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 systemsChaos 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
"""
# Pre-flight checks
if self.safety.business_hours_only:
current_hour = datetime.now().hour
if 9 <= current_hour <= 17: # Business hours
experiment.status = ExperimentStatus.ABORTED
experiment.observations = ["Aborted: Business hours constraint"]
return experiment
# Baseline metrics
baseline_metrics = get_metrics()
print(f"Baseline metrics: {baseline_metrics}")
# Start experiment
experiment.status = ExperimentStatus.RUNNING
experiment.started_at = datetime.now()
experiment.observations = []
try:
# Inject chaos
print(f"Injecting chaos: {experiment.name}")
injector.inject()
# Monitor for max duration
start_time = datetime.now()
max_duration = timedelta(minutes=self.safety.max_duration_minutes)
while datetime.now() - start_time < max_duration:
time.sleep(10) # Check every 10 seconds
current_metrics = get_metrics()
experiment.observations.append(
f"{datetime.now().isoformat()}: {current_metrics}"
)
# Check safety constraints
if experiment.should_abort(current_metrics):
print("ABORTING: Safety constraint violated")
experiment.status = ExperimentStatus.ABORTED
break
else:
# Completed successfully
experiment.status = ExperimentStatus.SUCCESS
except Exception as e:
print(f"ERROR: {e}")
experiment.status = ExperimentStatus.FAILED
experiment.observations.append(f"Exception: {str(e)}")
finally:
# Always rollback
print("Rolling back chaos injection")
injector.rollback()
experiment.completed_at = datetime.now()
return experiment
# Example usage
def get_current_metrics() -> dict:
"""Fetch metrics from Prometheus."""
# In real implementation, query Prometheus
return {
'error_rate': 0.02, # 2%
'latency_p99': 0.45, # 450ms
}
safety = SafetyConstraints(business_hours_only=False)
runner = ChaosRunner(safety)
experiment = ChaosExperiment(
name="Kill 50% of API pods",
hypothesis="API remains available with 50% pod loss",
blast_radius="50% of API pods",
rollback_plan="Pods auto-restart via deployment",
success_criteria="Error rate <5%, latency p99 <1s",
)
injector = PodKiller(
namespace="production",
label_selector="app=api",
kill_percentage=0.5,
)
result = runner.run_experiment(experiment, injector, get_current_metrics)
print(f"Experiment status: {result.status.value}")Game Days
Scheduled chaos engineering practice sessions.
# gameday_plan.yaml
gameday:
date: "2024-02-15"
duration: "2 hours"
participants:
- SRE team
- Backend engineers
- On-call rotation
objectives:
- Test incident response procedures
- Validate monitoring and alerting
- Practice communication protocols
- Identify gaps in runbooks
scenarios:
- scenario: "Database Primary Failure"
inject: "Terminate primary database pod"
expected: "Automatic failover to replica in <30s"
- scenario: "API Service Overload"
inject: "Generate 10x normal traffic"
expected: "Rate limiting activates, no errors"
- scenario: "Network Partition"
inject: "Block traffic between API and database"
expected: "Circuit breaker opens, graceful degradation"
success_criteria:
- "All scenarios handled without escalation"
- "MTTR <30 minutes for all scenarios"
- "Documentation updated with learnings"
- "Action items created for gaps"
safety:
- "Run in staging environment first"
- "VP Engineering notified beforehand"
- "Abort plan ready for each scenario"
- "Customer support team on standby"Chaos Engineering Maturity
from enum import IntEnum
class ChaosMaturity(IntEnum):
"""Chaos engineering maturity levels."""
NONE = 0 # No chaos testing
AD_HOC = 1 # Occasional manual tests
SCHEDULED = 2 # Regular game days
CONTINUOUS = 3 # Automated in CI/CD
CULTURE = 4 # Embedded in development
def assess_maturity(practices: dict[str, bool]) -> ChaosMaturity:
"""Assess chaos engineering maturity level."""
if not practices.get('any_chaos_testing'):
return ChaosMaturity.NONE
if not practices.get('regular_game_days'):
return ChaosMaturity.AD_HOC
if not practices.get('automated_chaos'):
return ChaosMaturity.SCHEDULED
if not practices.get('chaos_in_cicd'):
return ChaosMaturity.CONTINUOUS
return ChaosMaturity.CULTURE
# Example assessment
current_practices = {
'any_chaos_testing': True,
'regular_game_days': True,
'automated_chaos': True,
'chaos_in_cicd': False,
}
maturity = assess_maturity(current_practices)
print(f"Chaos maturity: {maturity.name}")
# Target: CONTINUOUS or CULTURE