Chaos Engineering in Practice: From GameDays to Continuous Verification

Hypothesis template:
"When [failure condition], [system component] will [expected behavior], 
resulting in [observable outcome], and [business metric] will [not degrade / 
degrade by less than X%]."

Example hypotheses:

"When the payment-service database replica fails, the application will fail over 
to the primary within 30 seconds, resulting in a brief error spike of less than 
5%, and checkout success rate will recover to baseline within 60 seconds."

"When the experiment injects 200ms of latency into calls from checkout-service to 
inventory-service, the checkout flow will complete within 2 seconds due to 
the 1-second timeout and circuit breaker, and inventory calls will fall back 
to cached values."

"When the experiment kills 50% of the recommendation-service pods simultaneously, 
HPA will provision replacement pods within 3 minutes, and the product 
page will degrade gracefully to showing popular items instead of 
personalized recommendations."

# Blast radius assessment before any experiment
class BlastRadiusAssessment:
    def __init__(self, target_service: str, experiment_type: str):
        self.target = target_service
        self.experiment = experiment_type
    
    def assess(self) -> BlastRadius:
        # What services depend on the target?
        downstream_services = get_dependency_graph(self.target, direction="downstream")
        
        # What is the traffic volume affected?
        requests_per_second = get_current_rps(self.target)
        
        # What is the revenue exposure per minute of impact?
        revenue_per_minute = get_revenue_attribution(self.target)
        
        # Is this reversible immediately?
        is_immediately_reversible = self.experiment in [
            "latency_injection",   # Stop injecting — latency returns to normal
            "error_injection",     # Stop injecting — errors stop
            "cpu_stress",          # Kill the stress process
        ]
        
        return BlastRadius(
            affected_services=downstream_services,
            max_affected_rps=requests_per_second,
            revenue_exposure_per_minute=revenue_per_minute,
            is_reversible=is_immediately_reversible,
            recommendation=self._recommend_environment(revenue_per_minute)
        )
    
    def _recommend_environment(self, revenue_per_minute: float) -> str:
        if revenue_per_minute > 10000:
            return "staging_only"    # >$10K/minute risk: staging only
        elif revenue_per_minute > 1000:
            return "production_off_peak"  # Run during low traffic
        else:
            return "production_anytime"   # Low risk: run in a production-style lab-style environments-style environments

# Network partition: isolate checkout-service from payment-service
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: checkout-payment-partition
  namespace: production
spec:
  action: partition
  mode: fixed-percent
  value: "50"      # Affect 50% of checkout-service pods
  selector:
    namespaces:
      - production
    labelSelectors:
      app: checkout-service
  direction: to
  target:
    selector:
      namespaces:
        - production
      labelSelectors:
        app: payment-service
    mode: all
  duration: "5m"   # Run for 5 minutes then auto-terminate

---
# Pod failure: kill 30% of pods
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: inventory-pod-failure
  namespace: production
spec:
  action: pod-kill
  mode: fixed-percent
  value: "30"
  selector:
    namespaces:
      - production
    labelSelectors:
      app: inventory-service
  duration: "3m"

---
# Latency injection: add 300ms to all outbound calls from search-service
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: search-dependency-latency
  namespace: staging
spec:
  action: delay
  mode: all
  selector:
    namespaces:
      - staging
    labelSelectors:
      app: search-service
  delay:
    latency: "300ms"
    correlation: "25"   # 25% correlation between successive packets
    jitter: "50ms"
  duration: "10m"

---
# Stress test: CPU pressure on API gateway
apiVersion: chaos-mesh.org/v1alpha1
kind: StressChaos
metadata:
  name: api-gateway-cpu-stress
  namespace: staging
spec:
  mode: fixed
  value: "2"
  selector:
    namespaces:
      - staging
    labelSelectors:
      app: api-gateway
  stressors:
    cpu:
      workers: 2        # 2 CPU-burning goroutines
      load: 80          # Each at 80% CPU
  duration: "5m"

Pre-GameDay (1 week before):
- Define the scenario: what failure will be injected?
- Define success criteria: what does a successful response look like?
- Identify the observers (different from responders)
- Ensure monitoring is in place to observe the system during the experiment
- Brief responders that a GameDay is happening this week (not which day/time)

GameDay Day:
1. Observers convene (separate channel from responders)
2. Inject the failure at an unannounced time
3. Observe and record:
   - Time to first alert: did monitoring detect the failure?
   - Time to detection by engineer: how long until someone noticed?
   - Diagnostic actions taken: what did the engineer look at?
   - Time to correct diagnosis: how long until root cause was identified?
   - Remediation actions: were they correct?
   - Time to resolution: how long until the system recovered?
4. Terminate the experiment (observers have a kill switch)

Post-GameDay (same day):
1. Hot debrief: what happened? (30 minutes, immediate impressions)
2. Observers share their notes
3. Identify gaps: monitoring gaps, blog post gaps, response gaps
4. Create action items with owners and deadlines

# chaos_verification.py — runs as a pipeline step post-deploy
import subprocess
import time
import requests

STEADY_STATE_METRICS = {
    "error_rate": 0.01,           # Max 1% error rate
    "p99_latency_ms": 500,        # Max 500ms p99 latency
    "checkout_success_rate": 0.99  # Min 99% checkout success
}

def verify_steady_state(service: str) -> bool:
    """Verify the system is in a healthy baseline before injecting chaos."""
    metrics = get_current_metrics(service)
    
    for metric, threshold in STEADY_STATE_METRICS.items():
        actual = metrics.get(metric)
        if metric == "error_rate" and actual > threshold:
            return False
        if metric == "p99_latency_ms" and actual > threshold:
            return False
        if metric == "checkout_success_rate" and actual < threshold:
            return False
    
    return True

def run_chaos_experiment(experiment_name: str, duration_seconds: int = 300) -> ExperimentResult:
    # Apply the chaos experiment via kubectl
    subprocess.run(["kubectl", "apply", "-f", f"chaos/{experiment_name}.yaml"])
    
    start_time = time.time()
    metrics_during = []
    
    while time.time() - start_time < duration_seconds:
        metrics_during.append(get_current_metrics("checkout-service"))
        time.sleep(10)
    
    # Terminate experiment
    subprocess.run(["kubectl", "delete", "-f", f"chaos/{experiment_name}.yaml"])
    
    # Wait for recovery
    time.sleep(60)
    post_metrics = get_current_metrics("checkout-service")
    
    return ExperimentResult(
        experiment=experiment_name,
        metrics_during=metrics_during,
        recovered=verify_steady_state("checkout-service"),
        max_error_rate=max(m["error_rate"] for m in metrics_during),
        max_p99_latency=max(m["p99_latency_ms"] for m in metrics_during)
    )

# Pipeline integration
def chaos_verification_stage(deployed_version: str) -> bool:
    print(f"Running chaos verification for version {deployed_version}")
    
    if not verify_steady_state("checkout-service"):
        print("System not in steady state — skipping chaos verification")
        return False   # Don't add chaos to an already-degraded system
    
    results = []
    
    # Run each experiment sequentially with recovery time between
    for experiment in ["pod-kill-20pct", "latency-300ms", "dependency-partition"]:
        result = run_chaos_experiment(experiment, duration_seconds=180)
        results.append(result)
        
        if not result.recovered:
            print(f"System did not recover after {experiment} — failing pipeline")
            send_alert(f"Chaos verification FAILED: {experiment} in {deployed_version}")
            return False
        
        time.sleep(60)  # Recovery time between experiments
    
    print(f"Chaos verification PASSED for {deployed_version}")
    return True