Chaos Engineering with SLOs: Testing Reliability Under Real-World Failures

1. Steady state hypothesis: "System maintains SLO during [failure]"
2. Inject failure
3. Measure: does SLO hold?
4. If SLO violated: you found a reliability gap — fix it
5. If SLO held: increase experiment scope or document as validated

# Install Chaos Monkey via Spinnaker (original) or standalone
# Or use simian-army for specific experiments

# Alternative: kube-monkey for Kubernetes
kubectl apply -f https://raw.githubusercontent.com/asobti/kube-monkey/master/manifests/configmap.yaml

# experiments/pod-kill-experiment.yaml
apiVersion: litmuschaos.io/v1alpha1
kind: ChaosEngine
metadata:
  name: pod-kill-test
  namespace: production
spec:
  appinfo:
    appns: production
    applabel: 'app=myapp'
    appkind: deployment
  chaosServiceAccount: litmus-admin
  
  experiments:
    - name: pod-delete
      spec:
        components:
          env:
            - name: TOTAL_CHAOS_DURATION
              value: "300"       # 5 minutes
            - name: CHAOS_INTERVAL
              value: "30"        # Kill a pod every 30 seconds
            - name: FORCE
              value: "false"     # Graceful shutdown
            - name: PODS_AFFECTED_PERC
              value: "50"        # Kill 50% of pods
        probe:
          # SLO probe: check availability during chaos
          - name: slo-availability-check
            type: httpProbe
            httpProbe/inputs:
              url: "https://api.example.com/health"
              insecureSkipVerify: false
              method:
                get:
                  criteria: ==
                  responseCode: "200"
            mode: Continuous
            runProperties:
              probeTimeout: 5
              interval: 5
              retry: 2

# chaos/chaos_harness.py
import time
import threading
import requests
import statistics
from dataclasses import dataclass, field
from typing import Callable, Optional
import subprocess
import random

@dataclass
class SLOMeasurement:
    timestamp: float
    success: bool
    latency_ms: float

@dataclass
class ChaosExperimentResult:
    experiment_name: str
    duration_seconds: int
    measurements: list[SLOMeasurement] = field(default_factory=list)
    
    @property
    def success_rate(self) -> float:
        if not self.measurements:
            return 0
        return sum(m.success for m in self.measurements) / len(self.measurements)
    
    @property
    def p99_latency_ms(self) -> float:
        latencies = sorted(m.latency_ms for m in self.measurements if m.success)
        if not latencies:
            return 0
        return latencies[int(len(latencies) * 0.99)]
    
    def slo_met(self, availability_slo: float = 0.999, latency_slo_p99_ms: float = 500) -> bool:
        return (
            self.success_rate >= availability_slo and
            self.p99_latency_ms <= latency_slo_p99_ms
        )

class ChaosHarness:
    def __init__(self, target_url: str):
        self.target_url = target_url
    
    def measure_slo(
        self,
        duration_seconds: int,
        interval_seconds: float = 1.0,
    ) -> list[SLOMeasurement]:
        """Continuously probe the target and record SLI measurements."""
        measurements = []
        end_time = time.time() + duration_seconds
        
        while time.time() < end_time:
            start = time.perf_counter()
            try:
                resp = requests.get(self.target_url, timeout=5)
                latency_ms = (time.perf_counter() - start) * 1000
                measurements.append(SLOMeasurement(
                    timestamp=time.time(),
                    success=200 <= resp.status_code < 300,
                    latency_ms=latency_ms,
                ))
            except Exception as e:
                latency_ms = (time.perf_counter() - start) * 1000
                measurements.append(SLOMeasurement(
                    timestamp=time.time(),
                    success=False,
                    latency_ms=latency_ms,
                ))
            
            time.sleep(interval_seconds)
        
        return measurements
    
    def run_experiment(
        self,
        experiment_name: str,
        failure_fn: Callable,
        duration_seconds: int = 300,
    ) -> ChaosExperimentResult:
        """
        Run a chaos experiment while measuring SLO compliance.
        """
        result = ChaosExperimentResult(
            experiment_name=experiment_name,
            duration_seconds=duration_seconds,
        )
        
        # Start SLO measurement in background thread
        measurement_thread_complete = threading.Event()
        measurements = []
        
        def measure_thread():
            measurements.extend(self.measure_slo(duration_seconds))
            measurement_thread_complete.set()
        
        thread = threading.Thread(target=measure_thread)
        thread.start()
        
        # Wait a bit, then inject failure
        time.sleep(10)  # Let baseline establish
        print(f"Injecting failure: {experiment_name}")
        failure_fn()
        
        # Wait for experiment to complete
        measurement_thread_complete.wait()
        result.measurements = measurements
        
        return result

# tests/chaos/test_pod_failure.py
import pytest
import subprocess
import time
from chaos.chaos_harness import ChaosHarness

AVAILABILITY_SLO = 0.999
LATENCY_P99_SLO_MS = 500

@pytest.fixture
def harness():
    return ChaosHarness("https://staging.example.com/api/products")

def kill_random_pod():
    """Kill a random pod in the production namespace."""
    # Get pod list
    result = subprocess.run(
        ["kubectl", "get", "pods", "-n", "production", 
         "-l", "app=myapp", "-o", "name"],
        capture_output=True, text=True
    )
    pods = result.stdout.strip().split('\n')
    
    if not pods:
        pytest.skip("No pods found")
    
    target_pod = random.choice(pods)
    subprocess.run(
        ["kubectl", "delete", target_pod, "-n", "production", "--grace-period=0"],
        check=True,
    )
    print(f"Killed pod: {target_pod}")

def throttle_cpu():
    """Limit CPU on all app containers to simulate resource contention."""
    subprocess.run([
        "kubectl", "patch", "deployment", "myapp",
        "-n", "production",
        "--patch", '{"spec":{"template":{"spec":{"containers":[{"name":"myapp","resources":{"limits":{"cpu":"100m"}}}]}}}}',
    ], check=True)

def restore_cpu():
    """Restore normal CPU limits."""
    subprocess.run([
        "kubectl", "patch", "deployment", "myapp",
        "-n", "production",
        "--patch", '{"spec":{"template":{"spec":{"containers":[{"name":"myapp","resources":{"limits":{"cpu":"1000m"}}}]}}}}',
    ], check=True)

@pytest.mark.chaos
def test_pod_kill_slo_maintained(harness):
    """System should maintain SLO when a pod is killed."""
    
    result = harness.run_experiment(
        experiment_name="random-pod-kill",
        failure_fn=kill_random_pod,
        duration_seconds=300,
    )
    
    print(f"\nChaos Experiment: {result.experiment_name}")
    print(f"Success rate: {result.success_rate:.4%}")
    print(f"p99 latency: {result.p99_latency_ms:.1f}ms")
    print(f"SLO met: {result.slo_met(AVAILABILITY_SLO, LATENCY_P99_SLO_MS)}")
    
    assert result.slo_met(AVAILABILITY_SLO, LATENCY_P99_SLO_MS), \
        f"SLO violated during pod-kill experiment:\n" \
        f"  Availability: {result.success_rate:.4%} (need {AVAILABILITY_SLO:.3%})\n" \
        f"  p99 latency: {result.p99_latency_ms:.1f}ms (need ≤{LATENCY_P99_SLO_MS}ms)"

@pytest.mark.chaos
def test_cpu_throttle_slo_maintained(harness):
    """System should maintain SLO under CPU constraints."""
    
    def throttle_and_restore():
        throttle_cpu()
        time.sleep(120)  # Throttle for 2 minutes
        restore_cpu()
    
    result = harness.run_experiment(
        experiment_name="cpu-throttle",
        failure_fn=throttle_and_restore,
        duration_seconds=300,
    )
    
    assert result.slo_met(AVAILABILITY_SLO, LATENCY_P99_SLO_MS), \
        f"SLO violated under CPU throttle:\n" \
        f"  p99 latency: {result.p99_latency_ms:.1f}ms — CPU saturation causing latency spike"

@pytest.mark.chaos
def test_database_latency_injection(harness):
    """System should handle database latency without violating availability SLO."""
    
    def inject_db_latency():
        # Use toxiproxy or similar to add network latency to DB connection
        subprocess.run([
            "toxiproxy-cli", "toxic", "add", 
            "--toxicName", "db-latency",
            "--type", "latency",
            "--attributes", "latency=200,jitter=50",
            "postgres"
        ], check=True)
    
    def remove_db_latency():
        subprocess.run([
            "toxiproxy-cli", "toxic", "remove",
            "--toxicName", "db-latency",
            "postgres"
        ], check=True)
    
    def inject_and_remove():
        inject_db_latency()
        time.sleep(180)  # 3 minutes of database latency
        remove_db_latency()
    
    result = harness.run_experiment(
        experiment_name="database-latency",
        failure_fn=inject_and_remove,
        duration_seconds=300,
    )
    
    # With 200ms DB latency, some latency increase is expected
    # But should remain available
    assert result.success_rate >= AVAILABILITY_SLO, \
        f"Availability dropped under DB latency: {result.success_rate:.4%}"

def generate_chaos_report(results: list[ChaosExperimentResult]) -> str:
    """Generate a human-readable chaos experiment report."""
    
    lines = ["# Chaos Engineering Report\n"]
    
    passed = [r for r in results if r.slo_met()]
    failed = [r for r in results if not r.slo_met()]
    
    lines.append(f"## Summary")
    lines.append(f"- Experiments run: {len(results)}")
    lines.append(f"- SLO maintained: {len(passed)}")
    lines.append(f"- SLO violated: {len(failed)}")
    lines.append("")
    
    if failed:
        lines.append("## ⚠️  Failures (Action Required)")
        for r in failed:
            lines.append(f"\n### {r.experiment_name}")
            lines.append(f"- Availability: {r.success_rate:.4%}")
            lines.append(f"- p99 latency: {r.p99_latency_ms:.0f}ms")
            lines.append(f"- **Gap**: System does not maintain SLO under this failure mode")
    
    lines.append("\n## ✓ Passed Experiments")
    for r in passed:
        lines.append(f"- {r.experiment_name}: {r.success_rate:.4%} availability, {r.p99_latency_ms:.0f}ms p99")
    
    return "\n".join(lines)

# .github/workflows/chaos-tests.yml
name: Chaos Engineering Tests

on:
  schedule:
    - cron: '0 2 * * 2'  # Weekly Tuesday at 2am (low traffic)
  workflow_dispatch:
    inputs:
      experiment:
        description: 'Specific experiment to run'
        required: false

jobs:
  chaos-staging:
    runs-on: ubuntu-latest
    environment: staging  # Requires approval for production
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Configure kubectl
        run: |
          echo "${{ secrets.KUBECONFIG }}" | base64 -d > ~/.kube/config
      
      - name: Run chaos experiments
        run: |
          pip install pytest requests
          pytest tests/chaos/ -v \
            --chaos-env staging \
            -m "chaos" \
            --tb=short \
            --timeout=600
      
      - name: Generate report
        if: always()
        run: python scripts/chaos_report.py --output chaos-report.md
      
      - name: Post report to Slack
        if: failure()
        run: |
          curl -X POST "$SLACK_WEBHOOK" \
            -H 'Content-type: application/json' \
            -d "{\"text\": \"⚠️ Chaos tests found SLO violations. See report: ${{ github.run_url }}\"}"