# Chaos Engineering

> Chaos engineering principles and implementation

- Skill: `neuralblitz/chaos-engineering-3` (Agent Skill)
- Install (CLI): `npx skillmds@latest add neuralblitz/chaos-engineering-3`
- Raw SKILL.md: https://api.skillmd.com/api/skills/neuralblitz/chaos-engineering-3/raw
- Safety review: pending (external: skill-scanner PASS, skillspector PASS)
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- License: MIT
- Author: NeuralBlitz (https://skillmd.com/u/neuralblitz)
- Updated: 2026-09-22
- Page: https://skillmd.com/skills/neuralblitz/chaos-engineering-3

---

## What I do
- Design chaos experiments
- Implement fault injection
- Test system resilience
- Handle failures gracefully
- Design for resilience
- Monitor chaos experiments
- Analyze failure modes
- Improve system robustness

## When to use me
When improving system resilience or implementing chaos testing.

## Chaos Experiment Principles
```
Chaos Engineering Principles:

1. Build a Hypothesis
   Define steady state behavior
   Assume failure will happen

2. Vary Real-World Events
   Simulate real failures
   Test edge cases

3. Run Experiments in Production
   Real traffic, real failures
   Learn from production

4. Minimize Blast Radius
   Start small
   Contain failures

5. Automate Experiments
   Continuous validation
   Regular schedules
```

## LitmusChaos Experiment
```yaml
# chaos-engineering/pod-kill-experiment.yaml
apiVersion: litmuschaos.io/v1alpha1
kind: ChaosEngine
metadata:
  name: pod-kill-chaos
  namespace: litmus
spec:
  appinfo:
    appns: default
    applabel: "app=myapp"
    appkind: deployment
  chaosServiceAccount: litmus-admin
  experiments:
    - name: pod-delete
      spec:
        components:
          env:
            # Percentage of pods to kill
            - name: TOTAL_CHAOS_DURATION
              value: '30'
            - name: CHAOS_INTERVAL
              value: '10'
            - name: PODS_AFFECTED_PERC
              value: '50'
            - name: KILL_COUNT
              value: '3'


# Network delay experiment
apiVersion: litmuschaos.io/v1alpha1
kind: ChaosEngine
metadata:
  name: network-delay-chaos
spec:
  appinfo:
    appns: default
    applabel: "app=myapp"
    applkind: deployment
  chaosServiceAccount: litmus-admin
  experiments:
    - name: network-delay
      spec:
        components:
          env:
            - name: TOTAL_CHAOS_DURATION
              value: '60'
            - name: NETWORK_LATENCY
              value: '2000'
            - name: JITTER
              value: '100'
            - name: CONTAINER_RUNTIME
              value: 'docker'
            - name: TARGET_CONTAINER
              value: 'myapp'
```

## Python Chaos Framework
```python
import random
import time
from typing import Callable, Any
from functools import wraps


class ChaosMonkey:
    """Simple chaos monkey for fault injection."""
    
    def __init__(self, failure_rate: float = 0.1):
        self.failure_rate = failure_rate
        self.failures = []
        self.injected_failures = []
    
    def inject_failure(self, func: Callable) -> Callable:
        """Decorator to inject random failures."""
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            if self._should_inject_failure():
                self._record_injection("function_failure", func.__name__)
                raise ChaosInjectedError(
                    f"Injected failure in {func.__name__}"
                )
            return func(*args, **kwargs)
        return wrapper
    
    def inject_latency(self, func: Callable) -> Callable:
        """Decorator to inject random latency."""
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            if self._should_inject_failure():
                latency = random.uniform(100, 2000)  # ms
                time.sleep(latency / 1000)
                self._record_injection("latency", func.__name__, latency)
            return func(*args, **kwargs)
        return wrapper
    
    def inject_network_error(self, func: Callable) -> Callable:
        """Decorator to simulate network errors."""
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            if self._should_inject_failure():
                error = random.choice([
                    ConnectionRefusedError,
                    ConnectionResetError,
                    TimeoutError,
                    OSError,
                ])
                self._record_injection("network_error", func.__name__, str(error))
                raise error("Simulated network error")
            return func(*args, **kwargs)
        return wrapper
    
    def _should_inject_failure(self) -> bool:
        """Determine if failure should be injected."""
        return random.random() < self.failure_rate
    
    def _record_injection(
        self,
        failure_type: str,
        target: str,
        detail: Any = None
    ):
        """Record injected failure."""
        self.injected_failures.append({
            "type": failure_type,
            "target": target,
            "detail": detail,
            "timestamp": time.time(),
        })
    
    def get_injection_stats(self) -> dict:
        """Get statistics on injected failures."""
        return {
            "total_injections": len(self.injected_failures),
            "by_type": self._group_by("type"),
            "by_target": self._group_by("target"),
        }
    
    def _group_by(self, key: str) -> dict:
        """Group failures by key."""
        from collections import defaultdict
        groups = defaultdict(int)
        for f in self.injected_failures:
            groups[f[key]] += 1
        return dict(groups)


class ChaosInjectedError(Exception):
    """Exception raised when chaos is injected."""
    pass
```

## Resilience Patterns
```python
import asyncio
from typing import Callable, TypeVar, Optional
from dataclasses import dataclass


T = TypeVar('T')


@dataclass
class CircuitBreakerConfig:
    """Configuration for circuit breaker."""
    failure_threshold: int = 5
    success_threshold: int = 2
    timeout_seconds: int = 60


class CircuitBreaker:
    """Circuit breaker for external service calls."""
    
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"
    
    def __init__(self, config: CircuitBreakerConfig):
        self.config = config
        self.state = self.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
        self.callable_ref = None
    
    def __call__(self, func: Callable) -> Callable:
        self.callable_ref = func
        return self._execute
    
    def _execute(self, *args, **kwargs) -> Any:
        if self.state == self.OPEN:
            if self._should_attempt_reset():
                self.state = self.HALF_OPEN
            else:
                raise CircuitOpenError("Circuit breaker is open")
        
        try:
            result = self.callable_ref(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
    
    def _should_attempt_reset(self) -> bool:
        if self.last_failure_time is None:
            return True
        return (
            time.time() - self.last_failure_time
        ) >= self.config.timeout_seconds
    
    def _on_success(self) -> None:
        if self.state == self.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.config.success_threshold:
                self.state = self.CLOSED
                self.failure_count = 0
        else:
            self.failure_count = 0
    
    def _on_failure(self) -> None:
        self.failure_count += 1
        self.last_failure_time = time.time()
        
        if self.state == self.HALF_OPEN:
            self.state = self.OPEN
            self.success_count = 0
        elif self.failure_count >= self.config.failure_threshold:
            self.state = self.OPEN


class CircuitOpenError(Exception):
    """Raised when circuit breaker is open."""
    pass


# Bulkhead pattern
class Bulkhead:
    """Thread pool isolation (bulkhead pattern)."""
    
    def __init__(self, max_concurrent: int = 10):
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self.active_tasks = 0
    
    async def execute(self, coro) -> Any:
        """Execute with bulkhead isolation."""
        async with self.semaphore:
            self.active_tasks += 1
            try:
                return await coro
            finally:
                self.active_tasks -= 1
    
    @property
    def available(self) -> int:
        """Available concurrent slots."""
        return self.semaphore._value


# Retry with jitter
async def retry_with_jitter(
    func: Callable,
    *args,
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    **kwargs
) -> Any:
    """Retry with exponential backoff and jitter."""
    last_exception = None
    
    for attempt in range(max_retries):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            last_exception = e
            
            if attempt >= max_retries - 1:
                raise
            
            delay = min(
                base_delay * (2 ** attempt),
                max_delay
            )
            jitter = random.uniform(0, delay * 0.1)
            
            await asyncio.sleep(delay + jitter)
    
    raise last_exception
```

## Designing for Failure
```python
# Graceful degradation patterns

class ServiceHealth:
    """Track service health for degradation decisions."""
    
    def __init__(self, service_name: str):
        self.service_name = service_name
        self.success_count = 0
        self.failure_count = 0
        self.last_success_time = None
        self.last_failure_time = None
    
    @property
    def is_healthy(self) -> bool:
        """Check if service is considered healthy."""
        if self.failure_count > 10:
            return False
        return True
    
    @property
    def health_score(self) -> float:
        """Calculate health score (0-1)."""
        total = self.success_count + self.failure_count
        if total == 0:
            return 1.0
        return self.success_count / total
    
    def record_success(self):
        self.success_count += 1
        self.last_success_time = time.time()
    
    def record_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()


class DegradationManager:
    """Manage graceful degradation."""
    
    def __init__(self):
        self.services: Dict[str, ServiceHealth] = {}
    
    def get_service_health(self, name: str) -> ServiceHealth:
        if name not in self.services:
            self.services[name] = ServiceHealth(name)
        return self.services[name]
    
    def should_use_fallback(self, service_name: str) -> bool:
        """Determine if fallback should be used."""
        health = self.get_service_health(service_name)
        
        # Use fallback if service is unhealthy
        if not health.is_healthy:
            return True
        
        # Use fallback if health score is low
        if health.health_score < 0.8:
            return True
        
        return False
    
    def get_fallback_data(
        self,
        service_name: str,
        fallback_type: str = "cache"
    ) -> Any:
        """Get fallback data."""
        if fallback_type == "cache":
            return self._get_cached_fallback(service_name)
        elif fallback_type == "stale":
            return self._get_stale_fallback(service_name)
        elif fallback_type == "static":
            return self._get_static_fallback(service_name)
    
    def _get_cached_fallback(self, service_name: str) -> Any:
        """Get data from cache."""
        # Implementation for cache fallback
        pass
    
    def _get_stale_fallback(self, service_name: str) -> Any:
        """Get stale data."""
        pass
    
    def _get_static_fallback(self, service_name: str) -> Any:
        """Get static fallback data."""
        return {"error": "Service temporarily unavailable"}
```

## Best Practices
```
Chaos Engineering Best Practices:

1. Start with observation
   Define steady state first
   Monitor baseline behavior

2. Begin small
   Single service
   Small percentage

3. Have a stop button
   Kill switch for experiments
   Automatic timeouts

4. Measure impact
   Track metrics
   Compare to baseline

5. Automate experiments
   Regular schedules
   CI/CD integration

6. Learn from failures
   Document findings
   Improve resilience

7. Test critical paths
   Most used features first
   Customer-facing systems

8. Prepare for recovery
   Know how to stop experiments
   Have rollback plan

9. Communicate
   Notify stakeholders
   Runbooks available

10. Make it routine
    Regular game days
    Continuous improvement
```