Chaos Engineering with Kubernetes and Python¶

# chaos-hypothesis.py
class SteadyStateHypothesis:
    def __init__(self):
        self.metrics = {
            "response_time_p95": "< 200ms",
            "error_rate": "< 0.1%",
            "availability": "> 99.9%",
            "throughput": "> 1000 req/s"
        }

    def validate_before_chaos(self):
        """Validate system is in steady state before chaos."""
        # Check metrics
        response_time = self.get_p95_response_time()
        error_rate = self.get_error_rate()
        availability = self.get_availability()

        assert response_time < 0.2, f"Response time too high: {response_time}"
        assert error_rate < 0.001, f"Error rate too high: {error_rate}"
        assert availability > 0.999, f"Availability too low: {availability}"

    def validate_during_chaos(self):
        """Validate system behavior during chaos."""
        # Monitor for cascading failures
        # Check if circuit breakers activate
        # Verify retry mechanisms work
        pass

    def validate_after_chaos(self):
        """Validate system recovers after chaos."""
        # Check if metrics return to steady state
        # Verify no permanent damage
        # Confirm system is healthy
        pass

# monitor-recovery.py
import time
import requests
from kubernetes import client, config

def monitor_pod_recovery():
    """Monitor pod recovery after chaos injection."""
    config.load_kube_config()
    v1 = client.CoreV1Api()

    # Get initial pod count
    initial_pods = v1.list_namespaced_pod(
        namespace="default",
        label_selector="app=nginx"
    )
    initial_count = len(initial_pods.items)
    print(f"Initial pod count: {initial_count}")

    # Wait for chaos to complete
    time.sleep(30)

    # Check pod count during chaos
    chaos_pods = v1.list_namespaced_pod(
        namespace="default",
        label_selector="app=nginx"
    )
    chaos_count = len(chaos_pods.items)
    print(f"Pod count during chaos: {chaos_count}")

    # Wait for recovery
    time.sleep(60)

    # Check final pod count
    final_pods = v1.list_namespaced_pod(
        namespace="default",
        label_selector="app=nginx"
    )
    final_count = len(final_pods.items)
    print(f"Final pod count: {final_count}")

    # Verify recovery
    assert final_count == initial_count, "Pods did not recover"
    print("✅ Pod recovery successful")

if __name__ == "__main__":
    monitor_pod_recovery()

# network-chaos-monitor.py
import time
import requests
import statistics

def monitor_network_chaos():
    """Monitor system behavior during network chaos."""
    base_url = "http://nginx-service.default.svc.cluster.local"

    response_times = []
    error_count = 0
    total_requests = 100

    print("Starting network chaos monitoring...")

    for i in range(total_requests):
        start_time = time.time()
        try:
            response = requests.get(base_url, timeout=5)
            response_time = time.time() - start_time
            response_times.append(response_time)

            if response.status_code != 200:
                error_count += 1

        except requests.exceptions.RequestException as e:
            error_count += 1
            print(f"Request {i} failed: {e}")

        time.sleep(0.1)  # 10 requests per second

    # Analyze results
    avg_response_time = statistics.mean(response_times)
    p95_response_time = statistics.quantiles(response_times, n=20)[18]  # 95th percentile
    error_rate = error_count / total_requests

    print(f"Average response time: {avg_response_time:.3f}s")
    print(f"P95 response time: {p95_response_time:.3f}s")
    print(f"Error rate: {error_rate:.2%}")

    # Verify system resilience
    assert error_rate < 0.1, f"Error rate too high: {error_rate}"
    assert p95_response_time < 1.0, f"P95 response time too high: {p95_response_time}"

if __name__ == "__main__":
    monitor_network_chaos()

# chaos-annotations.py
import time
from kubernetes import client, config

def annotate_chaos_event(event_type, description):
    """Add annotations to Prometheus for chaos events."""
    config.load_kube_config()
    v1 = client.CoreV1Api()

    # Create event annotation
    event = client.V1Event(
        metadata=client.V1ObjectMeta(
            name=f"chaos-{event_type}-{int(time.time())}",
            namespace="chaos-testing"
        ),
        involved_object=client.V1ObjectReference(
            kind="Pod",
            name="chaos-mesh-controller",
            namespace="chaos-testing"
        ),
        reason=event_type,
        message=description,
        first_timestamp=time.time(),
        last_timestamp=time.time()
    )

    v1.create_namespaced_event(
        namespace="chaos-testing",
        body=event
    )
    print(f"Chaos event annotated: {event_type} - {description}")

# Usage
annotate_chaos_event("pod-kill", "Nginx pod terminated for chaos experiment")
annotate_chaos_event("network-delay", "200ms network delay injected")
annotate_chaos_event("recovery", "System recovered from chaos experiment")

# failure-cascade-analysis.py
import networkx as nx
import matplotlib.pyplot as plt

def analyze_failure_cascade():
    """Analyze potential failure cascades in the system."""
    G = nx.DiGraph()

    # Add nodes (services)
    G.add_node("user", type="external")
    G.add_node("ingress", type="load_balancer")
    G.add_node("frontend", type="service")
    G.add_node("backend", type="service")
    G.add_node("database", type="data_store")
    G.add_node("cache", type="data_store")

    # Add edges (dependencies)
    G.add_edge("user", "ingress")
    G.add_edge("ingress", "frontend")
    G.add_edge("frontend", "backend")
    G.add_edge("backend", "database")
    G.add_edge("backend", "cache")

    # Analyze failure propagation
    print("Critical paths:")
    for path in nx.all_simple_paths(G, "user", "database"):
        print(f"  {' -> '.join(path)}")

    # Identify critical nodes
    centrality = nx.betweenness_centrality(G)
    critical_nodes = sorted(centrality.items(), key=lambda x: x[1], reverse=True)

    print("\nCritical nodes (by betweenness centrality):")
    for node, centrality in critical_nodes:
        print(f"  {node}: {centrality:.3f}")

    return G

if __name__ == "__main__":
    G = analyze_failure_cascade()

# chaos-best-practices.py
class ChaosBestPractices:
    def __init__(self):
        self.blast_radius = "small"
        self.duration = "short"
        self.frequency = "controlled"

    def start_small(self):
        """Start with small blast radius."""
        return {
            "scope": "single_pod",
            "duration": "30s",
            "frequency": "daily"
        }

    def measure_steady_state(self):
        """Always measure steady state before chaos."""
        return {
            "metrics": ["response_time", "error_rate", "throughput"],
            "baseline": "24_hours",
            "thresholds": {
                "response_time_p95": "< 200ms",
                "error_rate": "< 0.1%",
                "throughput": "> 1000 req/s"
            }
        }

    def run_in_staging_first(self):
        """Test in staging before production."""
        return {
            "staging": "full_chaos_experiments",
            "production": "controlled_experiments",
            "rollback": "automatic"
        }

    def document_hypotheses(self):
        """Document hypotheses and results."""
        return {
            "hypothesis": "System will recover within 30s",
            "metrics": "Pod count, response time, error rate",
            "results": "Actual recovery time and behavior",
            "learnings": "What we discovered about system resilience"
        }

    def automate_chaos(self):
        """Automate chaos experiments in CI/CD."""
        return {
            "trigger": "scheduled_or_manual",
            "scope": "non_critical_services",
            "monitoring": "real_time_alerts",
            "rollback": "automatic_on_failure"
        }