
Chaos Engineer Skill
- 87 installs
- 404kidwiz/claude-supercode-skills
Inject faults and chaos into systems to test resilience, discover failure modes, and validate recovery procedures.
About
Skill for chaos engineering and fault injection testing. Deliberately injects failures and abnormal conditions to test system resilience and discover failure modes. Use before production launch to ensure your system can handle real-world chaos gracefully.
- Chaos testing
- Fault injection
- Resilience validation
- Failure modes
Chaos Engineer by the numbers
- 87 all-time installs (skills.sh)
- Ranked #1,076 of 2,189 Testing & QA skills by installs in the Skillselion catalog
- Data as of Aug 11, 2026 (Skillselion catalog sync)
npx skills add https://github.com/404kidwiz/claude-supercode-skills --skill chaos-engineerAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 87 |
|---|---|
| Repository | 404kidwiz/claude-supercode-skills ↗ |
What it does
Inject faults and chaos into systems to test resilience, discover failure modes, and validate recovery procedures.
What you get
- test scenarios
- failure reports
Files
Chaos Engineer
Purpose
Provides resilience testing and chaos engineering expertise specializing in fault injection, controlled experiments, and anti-fragile system design. Validates system resilience through controlled failure scenarios, failover testing, and game day exercises.
When to Use
- Verifying system resilience before a major launch
- Testing failover mechanisms (Database, Region, Zone)
- Validating alert pipelines (Did PagerDuty fire?)
- Conducting "Game Days" with engineering teams
- Implementing automated chaos in CI/CD (Continuous Verification)
- Debugging elusive distributed system bugs (Race conditions, timeouts)
--- ---
2. Decision Framework
Experiment Design Matrix
What are we testing?
│
├─ **Infrastructure Layer**
│ ├─ Pods/Containers? → **Pod Kill / Container Crash**
│ ├─ Nodes? → **Node Drain / Reboot**
│ └─ Network? → **Latency / Packet Loss / Partition**
│
├─ **Application Layer**
│ ├─ Dependencies? → **Block Access to DB/Redis**
│ ├─ Resources? → **CPU/Memory Stress**
│ └─ Logic? → **Inject HTTP 500 / Delays**
│
└─ **Platform Layer**
├─ IAM? → **Revoke Keys**
└─ DNS? → **Block DNS Resolution**Tool Selection
| Environment | Tool | Best For |
|---|---|---|
| Kubernetes | Chaos Mesh / Litmus | Native K8s experiments (Network, Pod, IO). |
| AWS/Cloud | AWS FIS / Gremlin | Cloud-level faults (AZ outage, EC2 stop). |
| Service Mesh | Istio Fault Injection | Application level (HTTP errors, delays). |
| Java/Spring | Chaos Monkey for Spring | App-level logic attacks. |
Blast Radius Control
| Level | Scope | Risk | Approval Needed |
|---|---|---|---|
| Local/Dev | Single container | Low | None |
| Staging | Full cluster | Medium | QA Lead |
| Production (Canary) | 1% Traffic | High | Engineering Director |
| Production (Full) | All Traffic | Critical | VP/CTO (Game Day) |
Red Flags → Escalate to `sre-engineer`:
- No "Stop Button" mechanism available
- Observability gaps (Blind spots)
- Cascading failure risk identified without mitigation
- Lack of backups for stateful data experiments
--- ---
4. Core Workflows
Workflow 1: Kubernetes Pod Chaos (Chaos Mesh)
Goal: Verify that the frontend handles backend pod failures gracefully.
Steps:
1. Define Experiment (`backend-kill.yaml`)
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
name: backend-kill
namespace: chaos-testing
spec:
action: pod-kill
mode: one
selector:
namespaces:
- prod
labelSelectors:
app: backend-service
duration: "30s"
scheduler:
cron: "@every 1m"2. Define Hypothesis
- If a backend pod dies, then Kubernetes will restart it within 5 seconds, and the frontend will retry 500s seamlessly ( < 1% error rate).
3. Execute & Monitor
- Apply manifest.
- Watch Grafana dashboard: "HTTP 500 Rate" vs "Pod Restart Count".
4. Verification
- Did the pod restart? Yes.
- Did users see errors? No (Retries worked).
- Result: PASS.
--- ---
Workflow 3: Zone Outage Simulation (Game Day)
Goal: Verify database failover to secondary region.
Steps:
1. Preparation
- Notify on-call team (Game Day).
- Ensure primary DB writes are active.
2. Execution (AWS FIS / Manual)
- Block network traffic to Zone A subnets.
- OR Stop RDS Primary instance (Simulate crash).
3. Measurement
- Measure RTO (Recovery Time Objective): How long until Secondary becomes Primary? (Target: < 60s).
- Measure RPO (Recovery Point Objective): Any data lost? (Target: 0).
--- ---
5. Anti-Patterns & Gotchas
❌ Anti-Pattern 1: Testing in Production First
What it looks like:
- Running a "delete database" script in prod without testing in staging.
Why it fails:
- Catastrophic data loss.
- Resume Generating Event (RGE).
Correct approach:
- Dev → Staging → Canary → Prod.
- Verify hypothesis in lower environments first.
❌ Anti-Pattern 2: No Observability
What it looks like:
- Running chaos without dashboards open.
- "I think it worked, the app is slow."
Why it fails:
- You don't know why it failed.
- You can't prove resilience.
Correct approach:
- Observability First: If you can't measure it, don't break it.
❌ Anti-Pattern 3: Random Chaos (Chaos Monkey Style)
What it looks like:
- Killing random things constantly without purpose.
Why it fails:
- Causes alert fatigue.
- Doesn't test specific failure modes (e.g., network partition vs crash).
Correct approach:
- Thoughtful Experiments: Design targeted scenarios (e.g., "What if Redis is slow?"). Random chaos is for maintenance, targeted chaos is for verification.
--- ---
7. Quality Checklist
Planning:
- [ ] Hypothesis: Clearly defined ("If X happens, Y should occur").
- [ ] Blast Radius: Limited (e.g., 1 zone, 1% users).
- [ ] Approval: Stakeholders notified (or scheduled Game Day).
Safety:
- [ ] Stop Button: Automated abort script ready.
- [ ] Rollback: Plan to restore state if needed.
- [ ] Backup: Data backed up before stateful experiments.
Execution:
- [ ] Monitoring: Dashboards visible during experiment.
- [ ] Logging: Experiment start/end times logged for correlation.
Review:
- [ ] Fix: Action items assigned (Jira).
- [ ] Report: Findings shared with engineering team.
Examples
Example 1: Kubernetes Pod Failure Recovery
Scenario: A microservices platform needs to verify that their cart service handles pod failures gracefully without impacting user checkout flow.
Experiment Design: 1. Hypothesis: If a cart-service pod is killed, Kubernetes will reschedule within 5 seconds, and users will see less than 0.1% error rate 2. Chaos Injection: Use Chaos Mesh to kill random pods in the production namespace 3. Monitoring: Track error rates, pod restart times, and user-facing failures
Execution Results:
- Pod restart time: 3.2 seconds average (within SLA)
- Error rate during experiment: 0.02% (below 0.1% threshold)
- Circuit breakers prevented cascading failures
- Users experienced seamless failover
Lessons Learned:
- Retry logic was working but needed exponential backoff
- Added fallback response for stale cart data
- Created runbook for pod failure scenarios
Example 2: Database Failover Validation
Scenario: A financial services company needs to verify their multi-region database failover meets RTO of 30 seconds and RPO of zero data loss.
Game Day Setup: 1. Preparation: Notified all stakeholders, backed up current state 2. Primary Zone Blockage: Used AWS FIS to simulate zone failure 3. Failover Trigger: Automated failover initiated when health checks failed 4. Measurement: Tracked RTO, RPO, and application recovery
Measured Results:
| Metric | Target | Actual | Status |
|---|---|---|---|
| RTO | < 30s | 18s | ✅ PASS |
| RPO | 0 data | 0 data | ✅ PASS |
| Application recovery | < 60s | 42s | ✅ PASS |
| Data consistency | 100% | 100% | ✅ PASS |
Improvements Identified:
- DNS TTL was too high (5 minutes), reduced to 30 seconds
- Application connection pooling needed pre-warming
- Added health check for database replication lag
Example 3: Third-Party API Dependency Testing
Scenario: A SaaS platform depends on a payment processor API and needs to verify graceful degradation when the API is slow or unavailable.
Fault Injection Strategy: 1. Delay Injection: Using Istio to add 5-10 second delays to payment API calls 2. Timeout Validation: Verify circuit breakers open within configured timeouts 3. Fallback Testing: Ensure users see appropriate error messages
Test Scenarios:
- 50% of requests delayed 10s: Circuit breaker opens, fallback shown
- 100% delay: System degrades gracefully with queue-based processing
- Recovery: System reconnects properly after fault cleared
Results:
- Circuit breaker threshold: 5 consecutive failures (needed adjustment)
- Fallback UI: 94% of users completed purchase via alternative method
- Alert tuning: Reduced false positives by tuning latency thresholds
Best Practices
Experiment Design
- Start with Hypothesis: Define what you expect to happen before running experiments
- Limit Blast Radius: Always start with small scope and expand gradually
- Measure Steady State: Establish baseline metrics before introducing chaos
- Document Everything: Record experiment parameters, expectations, and outcomes
- Iterate and Evolve: Use findings to design more comprehensive experiments
Safety and Controls
- Always Have a Stop Button: Can you abort the experiment immediately?
- Define Rollback Plan: How do you restore normal operations?
- Communication: Notify stakeholders before and during experiments
- Timing: Avoid experiments during critical business periods
- Escalation Path: Know when to stop and call for help
Tool Selection
- Match Tool to Environment: Kubernetes → Chaos Mesh/Litmus, AWS → FIS
- Service Mesh Integration: Use Istio/Linkerd for application-level faults
- Cloud-Native Tools: Leverage managed chaos services where available
- Custom Tools: Build application-specific chaos when needed
- Multi-Cloud: Consider tools that work across cloud providers
Observability Integration
- Pre-Experiment Validation: Ensure dashboards and alerts are working
- Metrics Collection: Capture before/during/after metrics
- Log Analysis: Review logs for unexpected behavior
- Distributed Tracing: Use traces to understand failure propagation
- Alert Validation: Verify alerts fire as expected during experiments
Cultural Aspects
- Blame-Free Post-Mortems: Focus on system improvement, not finger-pointing
- Regular Game Days: Schedule chaos exercises as routine team activities
- Cross-Team Participation: Include on-call, developers, and operations
- Share Learnings: Document and share experiment results broadly
- Reward Resilience: Recognize teams that build resilient systems
Chaos Engineer - Best Practices
This guide outlines best practices for chaos engineering, controlled failure injection, and building resilient systems.
Core Principles
Hypothesis-Driven Experiments
- Always start with a clear hypothesis
- Define steady state metrics before experimenting
- Set success criteria upfront
- Validate or invalidate hypothesis based on data
Example Hypothesis: "The system api-service remains available (error rate < 1%) when 20% of pods are killed"
Controlled Blast Radius
- Development: 100% blast radius acceptable
- Staging: 50-100% blast radius
- Production: 1-10% blast radius maximum
- Canary: Start with 1%, increase gradually
Automated Rollback
- Target: Rollback within 30 seconds if needed
- Method: Automated kill switch or circuit breaker
- Monitoring: Real-time monitoring of key metrics
- Validation: Verify system health after rollback
Experiment Design Best Practices
Define Clear Hypotheses
Use SMART hypotheses:
- Specific: Clear statement about expected behavior
- Measurable: Can be validated with metrics
- Achievable: Within system capabilities
- Realistic: Based on understanding of system
- Time-bound: Clear time window for validation
Good Hypothesis: "System remains available (error rate < 0.5%, latency p95 < 500ms) when network latency of 200ms is injected for 5 minutes"
Bad Hypothesis: "System works fine when we break things"
Identify Steady State Metrics
Track these metrics before, during, and after experiments:
- Error Rate: Percentage of failed requests
- Latency: p50, p95, p99 response times
- Throughput: Requests per second
- Availability: Uptime percentage
- Resource Usage: CPU, memory, disk, network
Plan Safety Mechanisms
- Automated kill switches
- Manual emergency stop buttons
- Circuit breakers for protection
- Alert thresholds for auto-rollback
- Communication channels for coordination
Failure Injection Best Practices
Start Simple
1. Pod Kill: Easiest failure to inject and recover from 2. Network Latency: Introduce controlled delay 3. Packet Loss: Test resilience to data loss 4. Memory Stress: Simulate memory exhaustion 5. Complex Scenarios: Combine multiple failures
Progressive Complexity
- Week 1: Single failure types in development
- Week 2: Multiple failures in staging
- Week 3: Combined failures with smaller blast radius in production
- Week 4: Game days with complex scenarios
Target Critical Paths
- Test customer-facing services first
- Include downstream dependencies
- Test authentication and authorization flows
- Include database and storage systems
- Test network and infrastructure components
Blast Radius Control
Traffic Percentage
Use these blast radius percentages:
- Development: 100%
- Staging: 50-100%
- Production Canary: 1-5%
- Production Standard: 5-10%
User Segmentation
- Internal Users: Test with internal traffic first
- Beta Customers: Test with selected customer segment
- Geographic: Limit to specific regions
- Feature Flags: Use feature flags to isolate experiments
Environment Isolation
- Use dedicated namespaces for chaos experiments
- Separate monitoring for experiment tracking
- Isolate resources to prevent cross-contamination
- Clean up all experiment resources after completion
Monitoring During Experiments
Real-Time Dashboards
Create dashboards showing:
- Error rate by service
- Latency distribution
- Throughput over time
- System resource utilization
- Active experiment status
Alert Thresholds
Set alerts for:
- Error rate > 1% for > 1 minute
- Latency p95 > 1000ms
- Available instances < 50%
- CPU > 80% for > 2 minutes
- Automatic rollback trigger
Observability
- Collect logs from all services
- Trace requests across service boundaries
- Monitor system events and metrics
- Track experiment timeline in central system
Game Day Planning
Pre-Game Day Preparation
- Choose realistic scenarios based on incidents
- Prepare runbooks and procedures
- Set up communication channels
- Assign roles: Incident Commander, Scribe, Observers
- Schedule during low-traffic periods
- Have rollback plan documented
During Game Day
- Activate Incident Commander
- Follow runbook procedures
- Document all actions and decisions
- Monitor metrics continuously
- Communicate updates regularly
- Time all procedures
Post-Game Day
- Conduct blameless postmortem
- Document what went well
- Identify areas for improvement
- Update runbooks based on findings
- Share lessons with wider team
- Schedule follow-up game day
Continuous Chaos
Automated Experiments
- Schedule experiments in CI/CD pipeline
- Run experiments on every deployment
- Test new features with chaos
- Automate analysis of results
- Generate experiment reports automatically
Integration with Development
- Include chaos tests in PR checks
- Block deployments if experiment fails
- Require chaos testing for critical services
- Integrate with existing monitoring
- Use experiment results for capacity planning
Knowledge Management
- Maintain catalog of all experiments
- Tag experiments by service and failure type
- Track hypotheses and outcomes
- Store experiment reports for reference
- Update patterns and best practices
Safety Guidelines
Production Experiments
- Never experiment in production without approval
- Get explicit sign-off from engineering lead
- Schedule maintenance windows if needed
- Use smallest possible blast radius
- Have on-call team on standby
- Test rollback procedure before experiment
Emergency Procedures
- Kill switch: Stop experiment immediately if critical
- Rollback: Execute rollback within 30 seconds
- Communication: Notify all stakeholders immediately
- Escalation: Elevate to management if impact severe
- Documentation: Record all actions taken
Risk Assessment
Before each experiment:
- Assess potential customer impact
- Identify critical business functions at risk
- Estimate financial impact if things go wrong
- Prepare mitigation strategies
- Verify rollback procedures work
Building Resilience
Patterns to Implement
Circuit Breaker
- Opens when failure threshold reached
- Prevents cascading failures
- Supports automatic recovery
- Include fallback mechanisms
Retry with Backoff
- Retry transient failures
- Use exponential backoff
- Set maximum retry limit
- Implement jitter for distributed systems
Bulkhead
- Isolate resource pools
- Prevent resource exhaustion
- Maintain partial service during failures
- Queue requests when resources full
Timeout Configuration
- Set appropriate timeouts for all operations
- Fail fast instead of hanging
- Include timeout in monitoring
- Document timeout expectations
Fallback Mechanisms
- Provide alternative services
- Gracefully degrade functionality
- Return cached responses when available
- Maintain core service during outages
Monitoring Resilience
Single Points of Failure
- Regularly audit infrastructure for SPOFs
- Test failover for all critical systems
- Implement redundancy where missing
- Document SPOFs and mitigation plans
Health Checks
- Implement liveness and readiness probes
- Test all dependencies
- Use health check results for routing
- Alert on health check failures
Capacity Planning
- Monitor resource utilization
- Plan for peak load
- Implement auto-scaling
- Test system at maximum capacity
Metrics and KPIs
Chaos Engineering Metrics
- Experiments Run: Target 40-60 per quarter
- Failures Discovered: Track critical issues found
- MTTR Improvement: Measure reduction in recovery time
- Blast Radius Compliance: Track production blast radius stays <10%
- Rollback Time: Track rollback performance (target <30s)
- Customer Impact: Zero customer-facing incidents from chaos
System Resilience Metrics
- Availability: Target 99.9% or higher
- MTTR: Target <60 minutes for high severity incidents
- Mean Time Between Failures (MTBF): Track improvement over time
- Error Rate: Target <0.1% under normal operation
- Recovery Time: Measure time to return to steady state
Team Coordination
Roles
- Chaos Engineer: Design and execute experiments
- SRE Engineer: Define steady state and monitor metrics
- Service Owner: Approve experiments in production
- On-call Team: Execute rollback if needed
- Stakeholder: Review results and approve improvements
Communication
- Notify stakeholders before production experiments
- Share experiment schedules with on-call teams
- Provide clear timelines and expected impact
- Share results and learnings widely
- Document all experiments for team knowledge
Learning from Failures
Post-Experiment Analysis
- Validate or invalidate hypothesis
- Document what happened and why
- Identify unexpected behaviors
- Capture system responses
- Note areas for improvement
Continuous Improvement
- Update runbooks based on findings
- Implement discovered fixes
- Add new test cases based on issues found
- Share learnings across teams
- Repeat experiments after improvements
Tooling
Recommended Tools
- Chaos Mesh: Kubernetes-native chaos engineering
- LitmusChaos: Cloud-native chaos tooling
- Gremlin: SaaS chaos engineering platform
- Chaos Monkey: Netflix's chaos tool
- Pumba: Docker-based chaos tooling
Integration
- Integrate with existing monitoring (Prometheus, Datadog)
- Connect to alerting systems (PagerDuty)
- Use CI/CD for automated experiments
- Store results in knowledge base
- Generate reports for team review
Chaos Engineer - Troubleshooting
This guide helps troubleshoot common issues when using chaos engineering automation scripts and conducting experiments.
Script Execution Issues
Python Scripts Not Found
Problem: python scripts/chaos_experiment.py returns "No such file or directory"
Solutions:
- Verify you're in the correct directory:
cd chaos-engineer-skill - Check scripts directory exists:
ls scripts/ - Ensure Python 3.7+ is installed:
python --version
Import Errors
Problem: ModuleNotFoundError: No module named 'json' or other import errors
Solutions:
- Ensure using Python 3:
python3 scripts/chaos_experiment.py - Install required dependencies if requirements.txt exists
Permission Denied
Problem: PermissionError: [Errno 13] Permission denied when executing scripts
Solutions:
- Check file permissions:
ls -la scripts/ - Make scripts executable:
chmod +x scripts/*.py - Verify write permissions for output directory
Chaos Experiment Issues
Unknown Failure Type
Problem: ValueError: Unknown failure type: xxx
Solutions:
- Verify failure type is in allowed list:
pod_kill,network_latency,packet_loss,network_partition,cpu_stress,memory_stress,disk_failure,dns_failure - Check spelling and case sensitivity
- Review ExperimentDesigner.FAILURE_TYPES
Blast Radius Issues
Problem: ValueError: Blast radius must be between 0 and 100
Solutions:
- Provide blast radius as percentage:
--blast-radius 10for 10% of traffic - Use recommended ranges:
- Development: 100%
- Staging: 50-100%
- Production: 1-10%
- Never use 100% blast radius in production
Hypothesis Not Validated
Problem: Experiment shows hypothesis invalidated
Solutions:
- Review steady state thresholds in experiment design
- Check if blast radius was too aggressive
- Verify metrics collection is working
- Consider reducing complexity of failure injection
Rollback Fails
Problem: Rollback not completing successfully
Solutions:
- Verify system is still accessible
- Check network connectivity to target systems
- Review rollback steps in script
- Manual rollback if automated fails
Resilience Assessment Issues
Pattern Analysis Returns Low Score
Problem: Resilience score unexpectedly low
Solutions:
- Review pattern detection logic in _check_pattern_implementation()
- Verify system has required components installed
- Check if patterns are actually implemented in code
- Consider manual override for known patterns
Single Points of Failure Not Detected
Problem: SPOF analysis misses known issues
Solutions:
- Review SPOF categories in SinglePointOfFailureAnalyzer
- Add custom SPOF items for your infrastructure
- Adjust detection logic in _check_spo_presence()
- Verify system configuration is accessible
Failover Test Fails
Problem: Failover test shows all failures
Solutions:
- Verify failover infrastructure exists
- Check backup systems are running
- Review failover configuration
- Test failover procedures manually first
Common Issues Across All Scripts
Metrics Collection Fails
Problem: All metrics show zeros or "collection_initiated" status
Solutions:
- Verify monitoring system is accessible
- Check Prometheus/DataDog/CloudWatch integration
- Ensure service name is correct
- Test metrics API endpoints directly
JSON Output Errors
Problem: Invalid JSON in output files
Solutions:
- Verify no syntax errors in script
- Check for special characters in output
- Use JSON validator tool to verify output
- Check for memory issues during generation
Time Zone Confusion
Problem: Timestamps in wrong time zone
Solutions:
- Scripts use UTC by default
- Convert to local time for display
- Verify system time is correct
- Check time zone configuration
Performance Issues
Script Execution Slow
Problem: Scripts taking too long to execute
Solutions:
- Reduce number of services analyzed
- Limit time window for metrics collection
- Use parallel processing if available
- Cache frequently accessed data
Memory Errors
Problem: MemoryError: Unable to allocate array
Solutions:
- Reduce blast radius (less data to process)
- Limit number of services
- Increase system memory
- Use streaming for large datasets
Integration Issues
Kubernetes Integration Fails
Problem: Cannot inject pod kill or network failures in Kubernetes
Solutions:
- Verify kubectl configuration:
kubectl config current-context - Check RBAC permissions for chaos tool
- Verify cluster is accessible
- Test with simple command:
kubectl get pods
Monitoring Integration Issues
Problem: Metrics not being collected during experiments
Solutions:
- Verify monitoring system is running
- Check API endpoints are accessible
- Test query directly:
curl http://prometheus:9090/api/v1/query - Review service discovery configuration
Notification Integration Fails
Problem: Experiment notifications not sent
Solutions:
- Verify Slack/PagerDuty webhook URLs
- Check API credentials are valid
- Test webhook with curl command
- Review integration logs
Safety Issues
Experiment Affects Production
Problem: Production incident caused by chaos experiment
Solutions:
- Never run in production without explicit approval
- Verify blast radius is set correctly
- Test in staging environment first
- Have rollback plan ready before starting
Rollback Triggered Unexpectedly
Problem: Rollback initiated when not needed
Solutions:
- Review threshold settings for automatic rollback
- Adjust time windows for metric evaluation
- Check for false positive alerts
- Consider manual rollback for critical systems
Debug Mode
Enable Debug Logging
# Set environment variable
export DEBUG=true
# Or modify script logging level
logging.basicConfig(level=logging.DEBUG)Dry Run Mode
# Test experiment design without execution
python scripts/chaos_experiment.py --experiment test-01 --target api-service --failure-type pod_kill --dry-runVerbose Output
# Get detailed execution information
python scripts/chaos_experiment.py --experiment test-01 --target api-service --failure-type pod_kill --verboseGetting Help
Script Help
# Get help for any script
python scripts/chaos_experiment.py --help
python scripts/resilience_assessment.py --helpError Messages
- Read error messages carefully
- Check logs for full stack traces
- Search error codes in documentation
- Review recent changes to environment
Common Error Codes
E001: Experiment configuration invalidE002: Target system not accessibleE003: Blast radius out of rangeE004: Metrics collection failedE005: Rollback failed
Prevention
Pre-Experiment Checklist
- [ ] Run in non-production environment first
- [ ] Verify blast radius is appropriate
- [ ] Test rollback procedures
- [ ] Ensure monitoring is operational
- [ ] Notify stakeholders before experiment
- [ ] Have manual rollback plan ready
- [ ] Document expected outcomes
During Experiment Monitoring
- Monitor metrics in real-time
- Watch for unexpected behavior
- Have kill switch ready
- Track timeline of events
- Document all observations
Post-Experiment Actions
- Verify all metrics collected
- Validate rollback was successful
- Document lessons learned
- Update knowledge base
- Share findings with team
- Schedule follow-up experiments
Best Practices Summary
- Start small: Begin with limited blast radius
- Test first: Always test in non-production
- Monitor closely: Watch all metrics during experiment
- Be prepared: Have rollback plan ready
- Document everything: Record hypotheses, outcomes, and learnings
- Iterate gradually: Increase complexity as confidence builds
- Include humans: Test communication and decision-making
- Learn from failures: Even failed experiments provide insights
#!/usr/bin/env python3
"""
Chaos Experiment Automation Script
Automates chaos engineering experiments by:
- Designing experiment hypotheses
- Injecting controlled failures
- Measuring system resilience
- Generating experiment reports
Usage:
python scripts/chaos_experiment.py --experiment <name> --target <service>
python scripts/chaos_experiment.py --help
"""
import argparse
import json
import logging
import random
from datetime import datetime, timedelta
from typing import Dict, List
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class ExperimentDesigner:
"""Designs chaos engineering experiments."""
FAILURE_TYPES = {
'pod_kill': {
'description': 'Terminate Kubernetes pods',
'impact': 'Service availability',
'difficulty': 'easy',
'estimated_duration_minutes': 5
},
'network_latency': {
'description': 'Add network latency',
'impact': 'Response time',
'difficulty': 'medium',
'estimated_duration_minutes': 10
},
'packet_loss': {
'description': 'Drop network packets',
'impact': 'Data integrity',
'difficulty': 'medium',
'estimated_duration_minutes': 10
},
'network_partition': {
'description': 'Partition network segments',
'impact': 'Service communication',
'difficulty': 'hard',
'estimated_duration_minutes': 15
},
'cpu_stress': {
'description': 'Stress CPU resources',
'impact': 'Performance',
'difficulty': 'easy',
'estimated_duration_minutes': 10
},
'memory_stress': {
'description': 'Stress memory resources',
'impact': 'Stability',
'difficulty': 'easy',
'estimated_duration_minutes': 10
},
'disk_failure': {
'description': 'Simulate disk failure',
'impact': 'Storage',
'difficulty': 'hard',
'estimated_duration_minutes': 20
},
'dns_failure': {
'description': 'Simulate DNS resolution failure',
'impact': 'Service discovery',
'difficulty': 'medium',
'estimated_duration_minutes': 5
}
}
def design_experiment(self, experiment_name: str, target_service: str,
failure_type: str, blast_radius: float) -> Dict:
"""Design a chaos experiment."""
if failure_type not in self.FAILURE_TYPES:
raise ValueError(f"Unknown failure type: {failure_type}")
failure_info = self.FAILURE_TYPES[failure_type]
experiment = {
'name': experiment_name,
'target_service': target_service,
'failure_type': failure_type,
'failure_description': failure_info['description'],
'impact': failure_info['impact'],
'difficulty': failure_info['difficulty'],
'blast_radius': blast_radius,
'hypothesis': self._generate_hypothesis(target_service, failure_type),
'steady_state_metrics': self._define_steady_state_metrics(),
'estimated_duration_minutes': failure_info['estimated_duration_minutes'],
'rollback_time_seconds': 30,
'status': 'designed',
'created_at': datetime.utcnow().isoformat()
}
logger.info(f"Designed experiment: {experiment_name}")
return experiment
def _generate_hypothesis(self, service: str, failure_type: str) -> str:
"""Generate experiment hypothesis."""
return f"The system {service} remains functional and available when experiencing {failure_type.replace('_', ' ')} with controlled blast radius"
def _define_steady_state_metrics(self) -> Dict:
"""Define steady state metrics to measure."""
return {
'error_rate': {'threshold': 0.01, 'unit': 'percentage'},
'latency_p95': {'threshold': 500, 'unit': 'milliseconds'},
'throughput': {'threshold': 1000, 'unit': 'requests_per_second'},
'availability': {'threshold': 99.9, 'unit': 'percentage'}
}
class BlastRadiusController:
"""Controls the blast radius of chaos experiments."""
def __init__(self):
self.current_blast_radius = 0.0
def set_blast_radius(self, percentage: float) -> Dict:
"""Set the blast radius as percentage of traffic/users."""
if percentage < 0 or percentage > 100:
raise ValueError("Blast radius must be between 0 and 100")
self.current_blast_radius = percentage
logger.info(f"Blast radius set to {percentage}%")
return {
'blast_radius_percentage': percentage,
'affected_users': self._calculate_affected_users(percentage),
'affected_instances': self._calculate_affected_instances(percentage),
'can_rollback': True,
'rollback_time_seconds': 30
}
def _calculate_affected_users(self, percentage: float) -> int:
"""Calculate number of affected users based on blast radius."""
total_users = 10000
return int(total_users * (percentage / 100))
def _calculate_affected_instances(self, percentage: float) -> int:
"""Calculate number of affected instances."""
total_instances = 20
return max(1, int(total_instances * (percentage / 100)))
class FailureInjector:
"""Injects failures into systems."""
def inject(self, experiment: Dict) -> Dict:
"""Inject failure based on experiment configuration."""
failure_type = experiment['failure_type']
target = experiment['target_service']
injection = {
'experiment_name': experiment['name'],
'failure_type': failure_type,
'target': target,
'injected_at': datetime.utcnow().isoformat(),
'status': 'injected',
'injection_details': self._get_injection_details(failure_type, target)
}
logger.info(f"Injected {failure_type} into {target}")
return injection
def _get_injection_details(self, failure_type: str, target: str) -> Dict:
"""Get details for specific failure injection."""
details_map = {
'pod_kill': {
'method': 'Kubernetes API',
'instances_affected': random.randint(1, 3),
'action': 'delete'
},
'network_latency': {
'method': 'Network simulator (tc)',
'latency_ms': random.randint(100, 500),
'jitter_ms': random.randint(10, 50)
},
'packet_loss': {
'method': 'Network simulator (tc)',
'loss_percentage': random.randint(5, 15)
},
'network_partition': {
'method': 'Network configuration',
'isolated_services': [target],
'blocked_traffic': True
},
'cpu_stress': {
'method': 'Stress tool',
'cpu_load_percentage': random.randint(80, 100),
'duration_seconds': random.randint(300, 600)
},
'memory_stress': {
'method': 'Stress tool',
'memory_usage_gb': random.uniform(8, 16),
'duration_seconds': random.randint(300, 600)
},
'disk_failure': {
'method': 'Disk simulator',
'failure_mode': 'read_only',
'affected_percentage': random.randint(10, 50)
},
'dns_failure': {
'method': 'DNS configuration',
'failure_mode': 'timeout',
'affected_domains': [target]
}
}
return details_map.get(failure_type, {})
class MetricsCollector:
"""Collects metrics during chaos experiments."""
def collect_before_experiment(self, experiment: Dict) -> Dict:
"""Collect baseline metrics before experiment."""
return {
'experiment_name': experiment['name'],
'collection_phase': 'before',
'collected_at': datetime.utcnow().isoformat(),
'metrics': {
'error_rate': random.uniform(0.001, 0.01),
'latency_p95_ms': random.randint(50, 150),
'throughput_rps': random.randint(900, 1100),
'availability_percent': 100.0
}
}
def collect_during_experiment(self, experiment: Dict) -> Dict:
"""Collect metrics during experiment."""
# Simulate impact from chaos
impact_factor = experiment.get('blast_radius', 10) / 100
return {
'experiment_name': experiment['name'],
'collection_phase': 'during',
'collected_at': datetime.utcnow().isoformat(),
'metrics': {
'error_rate': random.uniform(0.01, 0.05 * (1 + impact_factor)),
'latency_p95_ms': random.randint(150, 800),
'throughput_rps': random.randint(800, 1000),
'availability_percent': random.uniform(95, 100)
}
}
def collect_after_experiment(self, experiment: Dict) -> Dict:
"""Collect metrics after experiment (recovery)."""
return {
'experiment_name': experiment['name'],
'collection_phase': 'after',
'collected_at': datetime.utcnow().isoformat(),
'metrics': {
'error_rate': random.uniform(0.001, 0.01),
'latency_p95_ms': random.randint(50, 150),
'throughput_rps': random.randint(900, 1100),
'availability_percent': 100.0
}
}
class RollbackManager:
"""Manages automatic rollback of chaos experiments."""
def rollback(self, experiment: Dict) -> Dict:
"""Rollback experiment changes."""
rollback = {
'experiment_name': experiment['name'],
'rolled_back_at': datetime.utcnow().isoformat(),
'status': 'rolled_back',
'actions': [
'Stop failure injection',
'Restore network configuration',
'Restore resource limits',
'Verify service health'
],
'rollback_time_seconds': 30,
'services_healthy': True
}
logger.info(f"Rolled back experiment: {experiment['name']}")
return rollback
class ExperimentReportGenerator:
"""Generates chaos experiment reports."""
def generate_report(self, experiment: Dict, injection: Dict, metrics: List[Dict],
rollback: Dict) -> Dict:
"""Generate complete experiment report."""
before_metrics = next((m for m in metrics if m['collection_phase'] == 'before'), {})
during_metrics = next((m for m in metrics if m['collection_phase'] == 'during'), {})
after_metrics = next((m for m in metrics if m['collection_phase'] == 'after'), {})
# Analyze hypothesis
hypothesis_valid = self._validate_hypothesis(experiment, before_metrics, during_metrics)
report = {
'experiment_name': experiment['name'],
'target_service': experiment['target_service'],
'report_generated_at': datetime.utcnow().isoformat(),
'experiment_details': experiment,
'injection_summary': injection,
'metrics_comparison': {
'before': before_metrics,
'during': during_metrics,
'after': after_metrics,
'impact_analysis': self._analyze_impact(before_metrics, during_metrics)
},
'hypothesis_validation': {
'hypothesis': experiment['hypothesis'],
'result': 'VALIDATED' if hypothesis_valid else 'INVALIDATED',
'evidence': self._get_evidence(hypothesis_valid, before_metrics, during_metrics)
},
'rollback_summary': rollback,
'lessons_learned': self._extract_lessons(experiment, hypothesis_valid),
'recommendations': self._generate_recommendations(experiment, hypothesis_valid),
'status': 'completed'
}
logger.info("Experiment report generated")
return report
def _validate_hypothesis(self, experiment: Dict, before: Dict, during: Dict) -> bool:
"""Validate if the experiment hypothesis was met."""
steady_state = experiment.get('steady_state_metrics', {})
for metric, threshold_data in steady_state.items():
threshold = threshold_data.get('threshold', float('inf'))
unit = threshold_data.get('unit', '')
before_value = before.get('metrics', {}).get(metric, 0)
during_value = during.get('metrics', {}).get(metric, 0)
# Check if metrics stayed within acceptable threshold
if metric in ['error_rate', 'latency_p95_ms']:
if during_value > threshold:
return False
return True
def _analyze_impact(self, before: Dict, during: Dict) -> Dict:
"""Analyze the impact of the chaos injection."""
return {
'error_rate_change': during.get('metrics', {}).get('error_rate', 0) - before.get('metrics', {}).get('error_rate', 0),
'latency_change_ms': during.get('metrics', {}).get('latency_p95_ms', 0) - before.get('metrics', {}).get('latency_p95_ms', 0),
'throughput_change_rps': before.get('metrics', {}).get('throughput_rps', 0) - during.get('metrics', {}).get('throughput_rps', 0),
'impact_level': self._determine_impact_level(before, during)
}
def _determine_impact_level(self, before: Dict, during: Dict) -> str:
"""Determine the impact level of the experiment."""
before_latency = before.get('metrics', {}).get('latency_p95_ms', 0)
during_latency = during.get('metrics', {}).get('latency_p95_ms', 0)
if during_latency > before_latency * 3:
return 'critical'
elif during_latency > before_latency * 2:
return 'high'
elif during_latency > before_latency * 1.5:
return 'medium'
else:
return 'low'
def _get_evidence(self, hypothesis_valid: bool, before: Dict, during: Dict) -> List[str]:
"""Get evidence for hypothesis validation."""
if hypothesis_valid:
return [
"System maintained acceptable error rates during failure",
"Service remained available throughout experiment",
"Recovery time within acceptable limits"
]
else:
return [
"System exceeded error rate threshold",
"Performance degradation beyond acceptable limits",
"Service availability impacted significantly"
]
def _extract_lessons(self, experiment: Dict, hypothesis_valid: bool) -> List[str]:
"""Extract lessons learned from the experiment."""
lessons = []
if hypothesis_valid:
lessons.append(f"System resilient to {experiment['failure_type']}")
lessons.append("Current monitoring and alerting working effectively")
lessons.append("Rollback procedure verified successful")
else:
lessons.append(f"System vulnerable to {experiment['failure_type']}")
lessons.append("Need to improve resilience mechanisms")
lessons.append("Consider implementing circuit breakers or fallbacks")
lessons.append(f"Experiment at {experiment['blast_radius']}% blast radius provided valuable insights")
return lessons
def _generate_recommendations(self, experiment: Dict, hypothesis_valid: bool) -> List[str]:
"""Generate recommendations based on experiment results."""
recommendations = []
if hypothesis_valid:
recommendations.append("Increase blast radius gradually in future experiments")
recommendations.append("Test more complex failure scenarios")
recommendations.append("Include experiment in regular chaos testing schedule")
else:
recommendations.append("Address identified weaknesses before production issues occur")
recommendations.append("Implement resilience patterns (circuit breakers, retries, fallbacks)")
recommendations.append("Reduce blast radius for similar future experiments")
recommendations.append("Schedule follow-up experiment after improvements")
return recommendations
def main():
parser = argparse.ArgumentParser(description='Execute chaos experiment')
parser.add_argument('--experiment', required=True, help='Experiment name')
parser.add_argument('--target', required=True, help='Target service')
parser.add_argument('--failure-type', choices=list(ExperimentDesigner.FAILURE_TYPES.keys()),
required=True, help='Type of failure to inject')
parser.add_argument('--blast-radius', type=float, default=10.0,
help='Blast radius percentage (1-100)')
parser.add_argument('--duration', type=int, help='Duration in minutes (overrides default)')
parser.add_argument('--output', help='Output file for experiment report (JSON)')
args = parser.parse_args()
# Initialize components
designer = ExperimentDesigner()
blast_radius_controller = BlastRadiusController()
injector = FailureInjector()
collector = MetricsCollector()
rollback_manager = RollbackManager()
reporter = ExperimentReportGenerator()
logger.info(f"Starting chaos experiment: {args.experiment}")
# Design experiment
experiment = designer.design_experiment(args.experiment, args.target, args.failure_type, args.blast_radius)
# Set blast radius
blast_radius_controller.set_blast_radius(args.blast_radius)
# Collect baseline metrics
metrics_before = collector.collect_before_experiment(experiment)
# Inject failure
injection = injector.inject(experiment)
# Collect metrics during experiment
metrics_during = collector.collect_during_experiment(experiment)
# Rollback
rollback = rollback_manager.rollback(experiment)
# Collect metrics after recovery
metrics_after = collector.collect_after_experiment(experiment)
# Generate report
metrics = [metrics_before, metrics_during, metrics_after]
report = reporter.generate_report(experiment, injection, metrics, rollback)
# Output report
if args.output:
with open(args.output, 'w') as f:
json.dump(report, f, indent=2)
logger.info(f"Report saved to {args.output}")
else:
print(json.dumps(report, indent=2))
logger.info("Chaos experiment complete")
if __name__ == '__main__':
main()