Incident Management and Chaos Engineering

Incident Response Framework

Structured approach to incident management.

from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import List

class Severity(Enum):
    """Incident severity levels."""
    SEV1 = "critical"     # Complete outage, major customer impact
    SEV2 = "high"         # Partial outage, significant impact
    SEV3 = "medium"       # Degraded performance, some users affected
    SEV4 = "low"          # Minor issue, minimal impact

@dataclass
class Incident:
    """Incident tracking."""
    id: str
    title: str
    severity: Severity
    started_at: datetime
    detected_at: datetime
    resolved_at: datetime | None = None
    root_cause: str | None = None
    impact: str | None = None

    @property
    def detection_time(self) -> float:
        """Time from start to detection in minutes."""
        delta = self.detected_at - self.started_at
        return delta.total_seconds() / 60

    @property
    def mttr(self) -> float | None:
        """Mean Time To Repair in minutes."""
        if not self.resolved_at:
            return None
        delta = self.resolved_at - self.detected_at
        return delta.total_seconds() / 60

    @property
    def total_duration(self) -> float | None:
        """Total incident duration in minutes."""
        if not self.resolved_at:
            return None
        delta = self.resolved_at - self.started_at
        return delta.total_seconds() / 60

# Example incident
incident = Incident(
    id="INC-2024-001",
    title="Database connection pool exhaustion",
    severity=Severity.SEV2,
    started_at=datetime(2024, 1, 15, 14, 30),
    detected_at=datetime(2024, 1, 15, 14, 35),
    resolved_at=datetime(2024, 1, 15, 15, 10),
    root_cause="Connection leak in payment service",
    impact="Payment processing delayed for 15% of users"
)

print(f"Detection time: {incident.detection_time:.1f} minutes")
print(f"MTTR: {incident.mttr:.1f} minutes")
print(f"Total duration: {incident.total_duration:.1f} minutes")

Incident Response Runbook

# incident_response.yaml
incident_response:
  detection:
    - "Acknowledge alert in PagerDuty"
    - "Join #incident-response Slack channel"
    - "Create incident doc from template"
    - "Assess severity (SEV1-4)"

  sev1_response:  # Critical - all hands
    - "Page on-call lead + backup"
    - "Notify VP Engineering immediately"
    - "Start Zoom war room"
    - "Assign incident commander"
    - "Assign communication lead"
    - "Post status update every 15 minutes"

  sev2_response:  # High - team response
    - "Page on-call engineer"
    - "Notify team lead"
    - "Create incident channel"
    - "Post status update every 30 minutes"

  roles:
    incident_commander:
      - "Coordinate response efforts"
      - "Make decisions quickly"
      - "Delegate tasks"
      - "Communicate with stakeholders"

    communication_lead:
      - "Post regular status updates"
      - "Notify affected customers"
      - "Update status page"
      - "Summarize timeline"

    on_call_engineer:
      - "Investigate root cause"
      - "Implement fixes"
      - "Verify resolution"
      - "Document actions taken"

  resolution:
    - "Verify metrics returned to normal"
    - "Monitor for 30 minutes"
    - "Post final status update"
    - "Schedule postmortem within 48 hours"
    - "Close incident"

Blameless Postmortem Template

# Postmortem: [Incident Title]

**Date:** 2024-01-15
**Authors:** [Names]
**Status:** Complete
**Severity:** SEV2

## Summary

One-paragraph summary of what happened, impact, and resolution.

## Impact

- **Duration:** 40 minutes (14:30 - 15:10 UTC)
- **Users affected:** ~15% of payment transactions
- **Revenue impact:** Estimated $X delayed
- **SLO impact:** Consumed 2.3% of monthly error budget

## Timeline (all times UTC)

| Time  | Event |
|-------|-------|
| 14:30 | Deployment of payment-service v2.3.0 completed |
| 14:32 | Error rate begins increasing |
| 14:35 | Alert fires: HighErrorRate |
| 14:36 | On-call engineer acknowledges |
| 14:40 | Incident declared SEV2 |
| 14:45 | Root cause identified: connection leak |
| 14:50 | Rollback initiated |
| 14:55 | Rollback completed |
| 15:00 | Error rate returns to normal |
| 15:10 | Incident resolved, monitoring continued |

## Root Cause

The payment-service v2.3.0 deployment introduced a connection leak in the
database connection pool. The new retry logic was not properly closing
connections on timeout, causing the pool to exhaust after ~20 minutes.

## Resolution

Rolled back to payment-service v2.2.1, which immediately resolved the issue.

## Detection

**What went well:**
- Alert fired within 5 minutes of issue start
- Clear runbook helped quick diagnosis

**What could be improved:**
- Could have caught in staging with longer load test
- Database connection pool metrics not monitored

## Action Items

| Action | Owner | Priority | Due Date |
|--------|-------|----------|----------|
| Add connection pool monitoring | @alice | P0 | 2024-01-20 |
| Extend staging load tests to 30min | @bob | P1 | 2024-01-25 |
| Review all resource cleanup in retry logic | @charlie | P1 | 2024-01-30 |
| Add integration test for connection leaks | @dave | P2 | 2024-02-05 |

## Lessons Learned

**What went well:**
- Quick detection and response
- Effective team communication
- Clear rollback procedure

**What didn't go well:**
- Issue not caught in pre-production testing
- No monitoring for connection pool exhaustion

**Where we got lucky:**
- Issue occurred during low-traffic period
- Only affected payment service, not critical systems

Chaos Engineering

Proactively test system resilience through controlled failure injection.

# chaos_experiment.py
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import Callable

class ExperimentStatus(Enum):
    """Chaos experiment lifecycle states."""
    PLANNED = "planned"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    ABORTED = "aborted"

@dataclass
class ChaosExperiment:
    """Define a chaos engineering experiment."""
    name: str
    hypothesis: str  # What we expect to happen
    blast_radius: str  # Scope of impact
    rollback_plan: str
    success_criteria: str
    status: ExperimentStatus = ExperimentStatus.PLANNED
    started_at: datetime | None = None
    completed_at: datetime | None = None
    observations: list[str] | None = None

    def should_abort(self, metrics: dict) -> bool:
        """Check if experiment should be aborted.

        Args:
            metrics: Current system metrics

        Returns:
            bool: True if experiment should abort
        """
        # Abort if error rate exceeds 10%
        if metrics.get('error_rate', 0) > 0.10:
            return True

        # Abort if latency p99 exceeds 2 seconds
        if metrics.get('latency_p99', 0) > 2.0:
            return True

        return False

# Example: Database failover experiment
db_failover_experiment = ChaosExperiment(
    name="Database Primary Failover",
    hypothesis="System automatically fails over to replica within 30s with <1% error rate",
    blast_radius="Single database instance, 50% of production traffic",
    rollback_plan="Restore primary database immediately, redirect traffic",
    success_criteria="- Failover completes in <30s\n- Error rate <1%\n- No data loss",
)

Chaos Testing Patterns

# chaos_patterns.py - Common chaos engineering patterns
import time
import random
from typing import Protocol

class ChaosInjector(Protocol):
    """Interface for chaos injection."""

    def inject(self) -> None:
        """Inject chaos into the system."""
        ...

    def rollback(self) -> None:
        """Remove chaos and restore normal operation."""
        ...

class LatencyInjector:
    """Inject artificial latency into requests."""

    def __init__(self, target_service: str, latency_ms: int):
        self.target_service = target_service
        self.latency_ms = latency_ms

    def inject(self) -> None:
        """Add latency using iptables or proxy."""
        # Example using tc (traffic control) on Linux
        import subprocess
        subprocess.run([
            "tc", "qdisc", "add", "dev", "eth0",
            "root", "netem", "delay", f"{self.latency_ms}ms"
        ])

    def rollback(self) -> None:
        """Remove latency."""
        import subprocess
        subprocess.run(["tc", "qdisc", "del", "dev", "eth0", "root"])

class PodKiller:
    """Kill pods to test resilience."""

    def __init__(self, namespace: str, label_selector: str, kill_percentage: float = 0.5):
        self.namespace = namespace
        self.label_selector = label_selector
        self.kill_percentage = kill_percentage
        self.killed_pods = []

    def inject(self) -> None:
        """Randomly kill pods matching selector."""
        import subprocess

        # Get pods
        result = subprocess.run(
            ["kubectl", "get", "pods", "-n", self.namespace,
             "-l", self.label_selector, "-o", "name"],
            capture_output=True,
            text=True
        )

        pods = result.stdout.strip().split('\n')
        num_to_kill = int(len(pods) * self.kill_percentage)
        pods_to_kill = random.sample(pods, num_to_kill)

        # Kill selected pods
        for pod in pods_to_kill:
            subprocess.run(["kubectl", "delete", pod, "-n", self.namespace])
            self.killed_pods.append(pod)

    def rollback(self) -> None:
        """Pods will be recreated by deployment controller."""
        # Wait for pods to be recreated
        time.sleep(30)

class NetworkPartition:
    """Simulate network partition between services."""

    def __init__(self, source_pod: str, target_service: str):
        self.source_pod = source_pod
        self.target_service = target_service

    def inject(self) -> None:
        """Block network traffic using iptables."""
        import subprocess
        subprocess.run([
            "kubectl", "exec", self.source_pod, "--",
            "iptables", "-A", "OUTPUT", "-d", self.target_service, "-j", "DROP"
        ])

    def rollback(self) -> None:
        """Restore network traffic."""
        import subprocess
        subprocess.run([
            "kubectl", "exec", self.source_pod, "--",
            "iptables", "-D", "OUTPUT", "-d", self.target_service, "-j", "DROP"
        ])

Chaos Experiment Runner

# chaos_runner.py - Safe chaos experiment execution
from dataclasses import dataclass
from datetime import datetime, timedelta
import time

@dataclass
class SafetyConstraints:
    """Safety constraints for chaos experiments."""
    max_error_rate: float = 0.10  # 10%
    max_latency_p99: float = 2.0  # 2 seconds
    max_duration_minutes: int = 15
    business_hours_only: bool = True

class ChaosRunner:
    """Safely execute chaos experiments with monitoring."""

    def __init__(self, safety: SafetyConstraints):
        self.safety = safety

    def run_experiment(
        self,
        experiment: ChaosExperiment,
        injector: ChaosInjector,
        get_metrics: Callable[[], dict],
    ) -> ChaosExperiment:
        """Execute chaos experiment safely.

        Args:
            experiment: Experiment definition
            injector: Chaos injector implementation
            get_metrics: Function to fetch current metrics

        Returns:
            Updated experiment with results
        """