Incident Management

Effective incident management minimizes the impact of production issues through structured response processes, clear communication, and continuous learning. A well-designed incident management system balances speed of resolution with thorough understanding.

Understanding Incident Management

Incident management is the process of responding to, resolving, and learning from production issues.

The Chaos of Unstructured Response

Without defined processes, incidents become chaotic:

Unstructured Incident Response:
├── Unclear who should respond
├── Multiple people working on same thing
├── No coordination
├── Customers unaware of issue
├── Duplicate communications
├── No clear resolution criteria
├── Lost context after resolution
└── Repeated similar incidents

Result: Prolonged outages, customer frustration, and team burnout.

Structured Incident Response

Clear processes enable effective resolution:

graph TD
    A[Alert Triggered] --> B[Incident Created]
    B --> C[Team Assembled]
    C --> D[Commander Assigned]
    D --> E[Impact Assessed]
    E --> F[Mitigation Started]
    F --> G[Resolution Achieved]
    G --> H[Customers Notified]
    H --> I[Postmortem Scheduled]
    
    style G fill:#d4f1d4
    style I fill:#e1f5ff

Incident Severity Levels

Consistent severity classification drives appropriate response.

Severity Definitions

Clear criteria for each level:

# incident-severity-levels.yaml
severity_levels:
  SEV1:
    name: Critical
    description: Complete service outage or data loss
    criteria:
      - All customers unable to use core functionality
      - Revenue-generating features completely down
      - Data breach or data loss occurring
      - Security vulnerability being actively exploited
    response_time: Immediate (< 5 minutes)
    page_who: All on-call engineers
    stakeholder_notification: Immediate
    executive_notification: Immediate
    status_page: Public update required
    example: "Payment processing completely down"
  
  SEV2:
    name: High
    description: Significant service degradation
    criteria:
      - Subset of customers affected
      - Core functionality degraded but accessible
      - Workaround exists but not ideal
      - Non-core features completely down
    response_time: < 15 minutes
    page_who: Primary on-call
    stakeholder_notification: Within 30 minutes
    executive_notification: If prolonged (> 2 hours)
    status_page: Update recommended
    example: "API latency 10x normal, causing timeouts"
  
  SEV3:
    name: Medium
    description: Minor service issues
    criteria:
      - Small number of customers affected
      - Non-critical functionality impacted
      - No revenue impact
      - Performance slightly degraded
    response_time: < 1 hour
    page_who: None (Slack notification)
    stakeholder_notification: Not required
    executive_notification: Not required
    status_page: Optional
    example: "Admin dashboard loading slowly"
  
  SEV4:
    name: Low
    description: Cosmetic or minor issues
    criteria:
      - No customer impact
      - Cosmetic issues only
      - Internal tools affected
    response_time: Next business day
    page_who: None
    stakeholder_notification: Not required
    executive_notification: Not required
    status_page: No
    example: "Monitoring dashboard display bug"

Severity assessment flowchart:

graph TD
    A[Issue Detected] --> B{Complete Outage?}
    B -->|Yes| C[SEV1]
    B -->|No| D{Customers Affected?}
    
    D -->|All/Most| E{Revenue Impact?}
    E -->|Yes| C
    E -->|No| F[SEV2]
    
    D -->|Some| G{Core Functionality?}
    G -->|Yes| F
    G -->|No| H[SEV3]
    
    D -->|None| I[SEV4]
    
    style C fill:#ff6b6b
    style F fill:#ffa500
    style H fill:#ffeb3b
    style I fill:#90ee90

Incident Response Process

Structured steps guide effective response.

Phase 1: Detection and Declaration

Recognize and declare incidents:

# Incident detection and declaration
from enum import Enum
from dataclasses import dataclass
from datetime import datetime

class IncidentSeverity(Enum):
    SEV1 = "critical"
    SEV2 = "high"
    SEV3 = "medium"
    SEV4 = "low"

@dataclass
class Incident:
    id: str
    title: str
    severity: IncidentSeverity
    detected_at: datetime
    declared_at: datetime
    affected_services: list
    customer_impact: str
    commander: str
    responders: list
    status: str

class IncidentManager:
    def declare_incident(
        self,
        title: str,
        severity: IncidentSeverity,
        affected_services: list,
        customer_impact: str
    ) -> Incident:
        """Declare a new incident and kickoff response"""
        
        incident_id = self.generate_incident_id()
        
        incident = Incident(
            id=incident_id,
            title=title,
            severity=severity,
            detected_at=datetime.now(),
            declared_at=datetime.now(),
            affected_services=affected_services,
            customer_impact=customer_impact,
            commander=None,
            responders=[],
            status="active"
        )
        
        # Create incident war room
        war_room = self.create_war_room(incident)
        
        # Page appropriate responders
        self.page_responders(incident)
        
        # Create initial status page update
        if severity in [IncidentSeverity.SEV1, IncidentSeverity.SEV2]:
            self.update_status_page(incident, "investigating")
        
        # Notify stakeholders
        self.notify_stakeholders(incident)
        
        # Create incident tracking ticket
        self.create_incident_ticket(incident)
        
        return incident
    
    def create_war_room(self, incident: Incident) -> str:
        """Create dedicated communication channel"""
        
        channel_name = f"incident-{incident.id}"
        
        slack.create_channel(channel_name)
        slack.set_topic(
            channel_name,
            f"{incident.severity.value.upper()}: {incident.title}"
        )
        
        # Pin incident information
        slack.pin_message(channel_name, f"""
        **Incident {incident.id}**
        Severity: {incident.severity.value}
        Title: {incident.title}
        Affected: {', '.join(incident.affected_services)}
        
        **Resources:**
        - Runbooks: https://runbooks.example.com
        - Metrics: https://grafana.example.com
        - Status Page: https://status.example.com
        - Incident Doc: https://docs.example.com/incidents/{incident.id}
        """)
        
        return channel_name
    
    def page_responders(self, incident: Incident):
        """Alert appropriate on-call engineers"""
        
        if incident.severity == IncidentSeverity.SEV1:
            # Page entire on-call rotation
            pagerduty.trigger_incident(
                title=f"SEV1: {incident.title}",
                urgency="high",
                escalation_policy="all-oncall"
            )
        elif incident.severity == IncidentSeverity.SEV2:
            # Page primary on-call
            pagerduty.trigger_incident(
                title=f"SEV2: {incident.title}",
                urgency="high",
                escalation_policy="primary-oncall"
            )
        else:
            # Slack notification only
            slack.notify_oncall_channel(
                f"Incident {incident.id} ({incident.severity.value}): {incident.title}"
            )

Phase 2: Response Coordination

Organize the response effort:

// Incident coordination roles
interface IncidentRoles {
  commander: string;        // Leads the response
  responders: string[];     // Investigate and fix
  scribe: string;          // Documents timeline
  communications: string;   // External updates
}

class IncidentCoordinator {
  assignRoles(incident: Incident): IncidentRoles {
    // Find available on-call engineers
    const oncall = this.getOncallEngineers();
    
    // Assign commander (most senior available)
    const commander = this.selectCommander(oncall, incident.severity);
    
    // Assign responders based on affected services
    const responders = this.selectResponders(
      incident.affectedServices,
      oncall
    );
    
    // Assign support roles for SEV1/SEV2
    let scribe = null;
    let communications = null;
    
    if (incident.severity === 'SEV1' || incident.severity === 'SEV2') {
      scribe = this.selectScribe(oncall);
      communications = this.selectCommunications();
    }
    
    const roles: IncidentRoles = {
      commander,
      responders,
      scribe,
      communications
    };
    
    // Notify everyone of their roles
    this.notifyRoles(incident, roles);
    
    return roles;
  }
  
  selectCommander(oncall: Engineer[], severity: string): string {
    // For SEV1, get most experienced engineer
    if (severity === 'SEV1') {
      return oncall
        .sort((a, b) => b.yearsExperience - a.yearsExperience)[0]
        .email;
    }
    
    // For lower severity, primary on-call
    return oncall.find(e => e.tier === 'primary').email;
  }
  
  runCommanderChecklist(incident: Incident): void {
    const checklist = [
      {
        task: "Assemble response team",
        completed: false
      },
      {
        task: "Assess severity and impact",
        completed: false
      },
      {
        task: "Start communication cadence",
        completed: false
      },
      {
        task: "Assign action items",
        completed: false
      },
      {
        task: "Update status page",
        completed: false
      },
      {
        task: "Coordinate with stakeholders",
        completed: false
      },
      {
        task: "Track progress and blockers",
        completed: false
      },
      {
        task: "Verify resolution",
        completed: false
      },
      {
        task: "Schedule postmortem",
        completed: false
      }
    ];
    
    // Post checklist to incident channel
    slack.postMessage(
      incident.warRoom,
      "Commander Checklist:\n" + 
      checklist.map((item, i) => 
        `${i + 1}. [ ] ${item.task}`
      ).join('\n')
    );
  }
}

Incident commander responsibilities:

Incident Commander Role:
├── Lead the response effort
├── Make strategic decisions
├── Coordinate responders
├── Manage communication cadence
├── Escalate when needed
├── Declare resolution
└── Ensure postmortem happens

Key Principles:
- Stay calm and focused
- Avoid getting into technical details
- Keep everyone informed
- Unblock responders
- Make decisions quickly
- Ask for help when needed

Phase 3: Investigation and Mitigation

Diagnose and resolve the issue:

// Incident investigation workflow
package incident

import (
    "time"
)

type Investigation struct {
    IncidentID string
    Hypothesis []Hypothesis
    Timeline   []TimelineEvent
    Actions    []Action
}

type Hypothesis struct {
    Description string
    Evidence    []string
    Tested      bool
    Confirmed   bool
}

type Action struct {
    Description string
    AssignedTo  string
    Status      string
    StartedAt   time.Time
    CompletedAt *time.Time
    Result      string
}

func (i *Investigation) ProposeHypothesis(description string) {
    hypothesis := Hypothesis{
        Description: description,
        Evidence:    []string{},
        Tested:      false,
        Confirmed:   false,
    }
    
    i.Hypothesis = append(i.Hypothesis, hypothesis)
    
    // Log to timeline
    i.AddTimelineEvent(
        "hypothesis_proposed",
        description,
    )
}

func (i *Investigation) TestHypothesis(index int, result bool, evidence string) {
    i.Hypothesis[index].Tested = true
    i.Hypothesis[index].Confirmed = result
    i.Hypothesis[index].Evidence = append(
        i.Hypothesis[index].Evidence,
        evidence,
    )
    
    status := "disproven"
    if result {
        status = "confirmed"
    }
    
    i.AddTimelineEvent(
        "hypothesis_tested",
        fmt.Sprintf("%s: %s", status, i.Hypothesis[index].Description),
    )
}

func (i *Investigation) AssignAction(description, assignee string) Action {
    action := Action{
        Description: description,
        AssignedTo:  assignee,
        Status:      "in_progress",
        StartedAt:   time.Now(),
    }
    
    i.Actions = append(i.Actions, action)
    
    i.AddTimelineEvent(
        "action_assigned",
        fmt.Sprintf("%s assigned to %s", description, assignee),
    )
    
    return action
}

func (i *Investigation) CompleteAction(index int, result string) {
    now := time.Now()
    i.Actions[index].Status = "completed"
    i.Actions[index].CompletedAt = &now
    i.Actions[index].Result = result
    
    i.AddTimelineEvent(
        "action_completed",
        fmt.Sprintf("%s: %s", i.Actions[index].Description, result),
    )
}

Investigation workflow:

sequenceDiagram
    participant C as Commander
    participant R1 as Responder 1
    participant R2 as Responder 2
    participant S as Systems
    
    C->>R1: Investigate logs
    C->>R2: Check recent deployments
    
    R1->>S: Query logs
    S-->>R1: Error patterns found
    R1->>C: Hypothesis: Database connection issue
    
    R2->>S: List deployments
    S-->>R2: Recent config change
    R2->>C: Hypothesis: Config caused issue
    
    C->>R2: Test config hypothesis
    R2->>S: Revert config
    S-->>R2: Issue persists
    R2->>C: Hypothesis disproven
    
    C->>R1: Proceed with database investigation
    R1->>S: Check connection pool
    S-->>R1: Pool exhausted
    R1->>C: Root cause confirmed
    
    C->>R1: Scale down and up
    R1->>S: Execute scale operation
    S-->>R1: Connections restored
    R1->>C: Issue resolved

Phase 4: Communication

Keep stakeholders informed:

# Incident communication management
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class CommunicationTemplate:
    severity: str
    audience: str
    frequency: timedelta
    template: str

communication_templates = [
    CommunicationTemplate(
        severity="SEV1",
        audience="customers",
        frequency=timedelta(minutes=15),
        template="""
        We are investigating an issue affecting {affected_services}.
        Customers may experience {impact}.
        Our team is actively working on a resolution.
        Next update in 15 minutes.
        """
    ),
    CommunicationTemplate(
        severity="SEV1",
        audience="internal",
        frequency=timedelta(minutes=5),
        template="""
        Status Update - {time_elapsed}
        Current Status: {status}
        Working Theory: {hypothesis}
        Actions In Progress: {actions}
        Blockers: {blockers}
        Next steps: {next_steps}
        """
    ),
    CommunicationTemplate(
        severity="SEV2",
        audience="customers",
        frequency=timedelta(minutes=30),
        template="""
        We are experiencing degraded performance in {affected_services}.
        Some customers may notice {impact}.
        We are working on a fix.
        Next update in 30 minutes.
        """
    ),
]

class IncidentCommunication:
    def __init__(self, incident: Incident):
        self.incident = incident
        self.last_update = {}
    
    def should_send_update(self, audience: str) -> bool:
        """Check if update is due"""
        template = self.get_template(audience)
        
        if audience not in self.last_update:
            return True
        
        elapsed = datetime.now() - self.last_update[audience]
        return elapsed >= template.frequency
    
    def send_customer_update(self, status: str, impact: str):
        """Post update to status page"""
        
        template = self.get_template("customers")
        
        message = template.template.format(
            affected_services=", ".join(self.incident.affected_services),
            impact=impact
        )
        
        status_page.post_update(
            incident_id=self.incident.id,
            status=status,
            message=message
        )
        
        self.last_update["customers"] = datetime.now()
    
    def send_internal_update(
        self,
        status: str,
        hypothesis: str,
        actions: list,
        blockers: list,
        next_steps: str
    ):
        """Post update to war room"""
        
        elapsed = datetime.now() - self.incident.declared_at
        hours = int(elapsed.total_seconds() // 3600)
        minutes = int((elapsed.total_seconds() % 3600) // 60)
        
        template = self.get_template("internal")
        
        message = template.template.format(
            time_elapsed=f"{hours}h {minutes}m",
            status=status,
            hypothesis=hypothesis,
            actions="\n".join(f"- {a}" for a in actions),
            blockers="\n".join(f"- {b}" for b in blockers) if blockers else "None",
            next_steps=next_steps
        )
        
        slack.post_message(self.incident.war_room, message)
        self.last_update["internal"] = datetime.now()

Phase 5: Resolution and Recovery

Verify fix and close incident:

// Resolution verification
interface ResolutionCriteria {
  errorRateNormal: boolean;
  latencyNormal: boolean;
  healthChecksPass: boolean;
  customerImpactGone: boolean;
  monitoringStable: boolean;
}

class IncidentResolution {
  async verifyResolution(incident: Incident): Promise<boolean> {
    const criteria = await this.checkResolutionCriteria(incident);
    
    const allCriteriaMet = Object.values(criteria).every(v => v === true);
    
    if (allCriteriaMet) {
      await this.closeIncident(incident);
      return true;
    }
    
    return false;
  }
  
  async checkResolutionCriteria(
    incident: Incident
  ): Promise<ResolutionCriteria> {
    const metrics = await this.getMetrics(incident.affectedServices);
    
    return {
      errorRateNormal: metrics.errorRate < 1.0,
      latencyNormal: metrics.p99Latency < 500,
      healthChecksPass: await this.checkHealth(incident.affectedServices),
      customerImpactGone: await this.verifyCustomerImpact(incident),
      monitoringStable: await this.checkMonitoringStability(
        incident,
        duration: 15 * 60  // 15 minutes stable
      )
    };
  }
  
  async closeIncident(incident: Incident): Promise<void> {
    // Update incident status
    incident.status = 'resolved';
    incident.resolvedAt = new Date();
    
    // Calculate duration
    const duration = incident.resolvedAt - incident.declaredAt;
    
    // Post resolution message
    await slack.postMessage(
      incident.warRoom,
      `Incident ${incident.id} has been resolved.
      Duration: ${this.formatDuration(duration)}
      
      Next steps:
      1. Monitor for 24 hours
      2. Postmortem scheduled for ${this.schedulePostmortem(incident)}
      3. War room will be archived in 48 hours`
    );
    
    // Update status page
    await statusPage.resolveIncident(
      incident.id,
      `The issue has been resolved. All services are operating normally.`
    );
    
    // Notify stakeholders
    await this.notifyResolution(incident);
    
    // Schedule postmortem
    await this.schedulePostmortem(incident);
  }
}

Postmortem Process

Learn from incidents to prevent recurrence.

Blameless Postmortem

Focus on systems, not individuals:

# Incident Postmortem Template

## Incident Overview
- **Incident ID**: INC-2024-08-15-001
- **Date**: 2024-08-15
- **Duration**: 2 hours 15 minutes
- **Severity**: SEV2
- **Services Affected**: Payment API, Order Service
- **Customer Impact**: 30% of payment attempts failed

## Timeline
| Time | Event |
|------|-------|
| 14:00 | Alert fired: High error rate in payment-api |
| 14:05 | Incident declared, team assembled |
| 14:15 | Identified database connection exhaustion |
| 14:30 | Attempted connection pool increase - no effect |
| 14:45 | Discovered connection leak in recent deployment |
| 15:00 | Rolled back to previous version |
| 15:15 | Error rate returning to normal |
| 15:30 | Monitoring stable, incident resolved |
| 16:15 | Final verification complete |

## Root Cause
A recent code change introduced a connection leak where database
connections were not properly closed in error handling paths.

## What Went Well
- Alert fired quickly (< 1 minute after issue started)
- Team assembled rapidly
- Clear communication throughout
- Rollback executed smoothly

## What Went Wrong
- Connection leak not caught in code review
- No automated tests for connection lifecycle
- Staging environment has smaller connection pool, masking issue
- Initial hypothesis (pool size) delayed correct diagnosis

## Action Items
- [ ] Add automated tests for database connection lifecycle (Owner: @alice, Due: 2024-08-22)
- [ ] Implement connection leak detection in CI pipeline (Owner: @bob, Due: 2024-08-29)
- [ ] Make staging environment match production configuration (Owner: @charlie, Due: 2024-09-05)
- [ ] Add runbook section for connection leak diagnosis (Owner: @alice, Due: 2024-08-20)
- [ ] Update code review checklist to include resource lifecycle (Owner: @bob, Due: 2024-08-20)

## Lessons Learned
- Staging must match production to catch issues
- Resource lifecycle should be explicit in tests
- Connection pool metrics should include leak detection

## Prevention
Long-term improvements to prevent similar incidents:
- Implement automated resource leak detection
- Add circuit breakers for database operations
- Enhance monitoring to detect leaks earlier

Key Takeaways