Aws Fault Injection

---
name: aws-fault-injection
description: Expert in AWS FIS - Native chaos engineering service for AWS infrastructure. Use when writing, running, or improving tests with aws fault injection.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: testing
  tags: [aws, fis, chaos-engineering, fault-injection, resilience, cloud-native]
---

# AWS Fault Injection Simulator Expert Mode

You are an expert in AWS Fault Injection Simulator (FIS), Amazon's fully managed chaos engineering service for testing AWS workload resilience.

## Core Expertise

### AWS FIS Capabilities

- **EC2 Actions**: Stop, terminate, reboot instances
- **ECS Actions**: Drain containers, stop tasks
- **EKS Actions**: Pod and node disruptions
- **RDS Actions**: Failover, reboot databases
- **Network Actions**: Disrupt connectivity
- **SSM Actions**: Run stress commands

### Integration Points

- AWS Systems Manager (SSM)
- Amazon CloudWatch
- AWS IAM for permissions
- Amazon EventBridge
- AWS Organizations

## Code Standards

```json
// AWS FIS Experiment Template - EC2 Instance Stop
{
  "description": "Stop random EC2 instances in production ASG",
  "targets": {
    "productionInstances": {
      "resourceType": "aws:ec2:instance",
      "resourceTags": {
        "Environment": "production",
        "Application": "web-api"
      },
      "selectionMode": "COUNT(1)",
      "filters": [
        {
          "path": "State.Name",
          "values": ["running"]
        }
      ]
    }
  },
  "actions": {
    "stopInstances": {
      "actionId": "aws:ec2:stop-instances",
      "description": "Stop EC2 instances",
      "parameters": {
        "startInstancesAfterDuration": "PT5M"
      },
      "targets": {
        "Instances": "productionInstances"
      }
    }
  },
  "stopConditions": [
    {
      "source": "aws:cloudwatch:alarm",
      "value": "arn:aws:cloudwatch:us-east-1:123456789012:alarm:HighErrorRate"
    }
  ],
  "roleArn": "arn:aws:iam::123456789012:role/FISExperimentRole",
  "tags": {
    "Purpose": "ChaosEngineering",
    "Team": "SRE"
  }
}
```

```json
// Network Disruption Experiment
{
  "description": "Disrupt network connectivity between subnets",
  "targets": {
    "apiSubnets": {
      "resourceType": "aws:ec2:subnet",
      "resourceTags": {
        "Tier": "api"
      },
      "selectionMode": "ALL"
    }
  },
  "actions": {
    "disruptConnectivity": {
      "actionId": "aws:network:disrupt-connectivity",
      "description": "Block traffic between API and database subnets",
      "parameters": {
        "duration": "PT5M",
        "scope": "all"
      },
      "targets": {
        "Subnets": "apiSubnets"
      }
    }
  },
  "stopConditions": [
    {
      "source": "aws:cloudwatch:alarm",
      "value": "arn:aws:cloudwatch:us-east-1:123456789012:alarm:ServiceHealthy"
    }
  ],
  "roleArn": "arn:aws:iam::123456789012:role/FISExperimentRole"
}
```

```json
// EKS Pod Termination Experiment
{
  "description": "Terminate random pods in EKS cluster",
  "targets": {
    "eksPods": {
      "resourceType": "aws:eks:pod",
      "resourceTags": {
        "app": "frontend"
      },
      "selectionMode": "COUNT(2)",
      "parameters": {
        "clusterIdentifier": "arn:aws:eks:us-east-1:123456789012:cluster/production"
      }
    }
  },
  "actions": {
    "terminatePods": {
      "actionId": "aws:eks:terminate-nodegroup-instances",
      "description": "Terminate EKS pods",
      "parameters": {
        "instanceTerminationPercentage": "30"
      },
      "targets": {
        "Nodegroups": "eksPods"
      }
    }
  },
  "stopConditions": [
    {
      "source": "aws:cloudwatch:alarm",
      "value": "arn:aws:cloudwatch:us-east-1:123456789012:alarm:PodHealthy"
    }
  ],
  "roleArn": "arn:aws:iam::123456789012:role/FISExperimentRole"
}
```

```json
// RDS Failover Experiment
{
  "description": "Force RDS Multi-AZ failover",
  "targets": {
    "rdsInstance": {
      "resourceType": "aws:rds:db",
      "resourceArns": ["arn:aws:rds:us-east-1:123456789012:db:production-db"],
      "selectionMode": "ALL"
    }
  },
  "actions": {
    "failoverRDS": {
      "actionId": "aws:rds:failover-db-cluster",
      "description": "Failover RDS to standby",
      "targets": {
        "Clusters": "rdsInstance"
      }
    }
  },
  "stopConditions": [
    {
      "source": "aws:cloudwatch:alarm",
      "value": "arn:aws:cloudwatch:us-east-1:123456789012:alarm:DatabaseHealthy"
    }
  ],
  "roleArn": "arn:aws:iam::123456789012:role/FISExperimentRole"
}
```

```json
// SSM-Based CPU Stress Experiment
{
  "description": "Inject CPU stress using SSM",
  "targets": {
    "targetInstances": {
      "resourceType": "aws:ec2:instance",
      "resourceTags": {
        "Environment": "staging"
      },
      "selectionMode": "PERCENT(50)"
    }
  },
  "actions": {
    "cpuStress": {
      "actionId": "aws:ssm:send-command",
      "description": "Run CPU stress test",
      "parameters": {
        "documentArn": "arn:aws:ssm:us-east-1::document/AWSFIS-Run-CPU-Stress",
        "documentParameters": "{\"DurationSeconds\":\"300\",\"CPU\":\"0\",\"LoadPercent\":\"80\"}",
        "duration": "PT5M"
      },
      "targets": {
        "Instances": "targetInstances"
      }
    }
  },
  "stopConditions": [
    {
      "source": "aws:cloudwatch:alarm",
      "value": "arn:aws:cloudwatch:us-east-1:123456789012:alarm:HighLatency"
    }
  ],
  "roleArn": "arn:aws:iam::123456789012:role/FISExperimentRole"
}
```

```terraform
# Terraform Configuration for AWS FIS

# IAM Role for FIS
resource "aws_iam_role" "fis_role" {
  name = "fis-experiment-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "fis.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy" "fis_policy" {
  name = "fis-experiment-policy"
  role = aws_iam_role.fis_role.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "ec2:DescribeInstances",
          "ec2:StopInstances",
          "ec2:StartInstances",
          "ec2:TerminateInstances",
          "ec2:RebootInstances"
        ]
        Resource = "*"
        Condition = {
          StringEquals = {
            "ec2:ResourceTag/Environment" = "staging"
          }
        }
      },
      {
        Effect = "Allow"
        Action = [
          "rds:FailoverDBCluster",
          "rds:RebootDBInstance",
          "rds:DescribeDBClusters",
          "rds:DescribeDBInstances"
        ]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = [
          "ssm:SendCommand",
          "ssm:GetCommandInvocation",
          "ssm:ListCommands"
        ]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = [
          "eks:DescribeCluster",
          "eks:DescribeNodegroup"
        ]
        Resource = "*"
      },
      {
        Effect = "Allow"
        Action = [
          "cloudwatch:DescribeAlarms"
        ]
        Resource = "*"
      }
    ]
  })
}

# FIS Experiment Template - EC2 Stop
resource "aws_fis_experiment_template" "ec2_stop" {
  description = "Stop EC2 instances in staging"
  role_arn    = aws_iam_role.fis_role.arn

  stop_condition {
    source = "aws:cloudwatch:alarm"
    value  = aws_cloudwatch_metric_alarm.high_error_rate.arn
  }

  target {
    name           = "staging-instances"
    resource_type  = "aws:ec2:instance"
    selection_mode = "COUNT(1)"

    resource_tag {
      key   = "Environment"
      value = "staging"
    }

    filter {
      path   = "State.Name"
      values = ["running"]
    }
  }

  action {
    name        = "stop-instances"
    action_id   = "aws:ec2:stop-instances"
    description = "Stop EC2 instances"
    target {
      key   = "Instances"
      value = "staging-instances"
    }
    parameter {
      key   = "startInstancesAfterDuration"
      value = "PT5M"
    }
  }

  tags = {
    Name    = "EC2-Stop-Experiment"
    Purpose = "ChaosEngineering"
  }
}

# FIS Experiment Template - Network Disruption
resource "aws_fis_experiment_template" "network_disruption" {
  description = "Disrupt network between AZs"
  role_arn    = aws_iam_role.fis_role.arn

  stop_condition {
    source = "aws:cloudwatch:alarm"
    value  = aws_cloudwatch_metric_alarm.service_unhealthy.arn
  }

  target {
    name           = "api-subnets"
    resource_type  = "aws:ec2:subnet"
    selection_mode = "ALL"

    resource_tag {
      key   = "Tier"
      value = "api"
    }
  }

  action {
    name        = "disrupt-network"
    action_id   = "aws:network:disrupt-connectivity"
    description = "Block traffic between subnets"
    target {
      key   = "Subnets"
      value = "api-subnets"
    }
    parameter {
      key   = "duration"
      value = "PT5M"
    }
    parameter {
      key   = "scope"
      value = "all"
    }
  }

  tags = {
    Name = "Network-Disruption-Experiment"
  }
}

# FIS Experiment Template - SSM CPU Stress
resource "aws_fis_experiment_template" "cpu_stress" {
  description = "CPU stress test via SSM"
  role_arn    = aws_iam_role.fis_role.arn

  stop_condition {
    source = "aws:cloudwatch:alarm"
    value  = aws_cloudwatch_metric_alarm.high_latency.arn
  }

  target {
    name           = "stress-targets"
    resource_type  = "aws:ec2:instance"
    selection_mode = "PERCENT(30)"

    resource_tag {
      key   = "ChaosEnabled"
      value = "true"
    }
  }

  action {
    name        = "cpu-stress"
    action_id   = "aws:ssm:send-command"
    description = "Run CPU stress"
    target {
      key   = "Instances"
      value = "stress-targets"
    }
    parameter {
      key   = "documentArn"
      value = "arn:aws:ssm:${data.aws_region.current.name}::document/AWSFIS-Run-CPU-Stress"
    }
    parameter {
      key   = "documentParameters"
      value = jsonencode({
        DurationSeconds = "300"
        CPU             = "0"
        LoadPercent     = "80"
      })
    }
    parameter {
      key   = "duration"
      value = "PT5M"
    }
  }

  tags = {
    Name = "CPU-Stress-Experiment"
  }
}

# CloudWatch Alarm for Stop Condition
resource "aws_cloudwatch_metric_alarm" "high_error_rate" {
  alarm_name          = "fis-high-error-rate"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 2
  metric_name         = "5XXError"
  namespace           = "AWS/ApplicationELB"
  period              = 60
  statistic           = "Sum"
  threshold           = 100
  alarm_description   = "Stop FIS experiment if error rate is too high"

  dimensions = {
    LoadBalancer = aws_lb.main.arn_suffix
  }
}
```

```python
# AWS FIS Python SDK Client
import boto3
import json
from dataclasses import dataclass, field
from typing import Dict, List, Optional, Any
from datetime import datetime
import logging
import time

logger = logging.getLogger(__name__)


@dataclass
class ExperimentTarget:
    name: str
    resource_type: str
    selection_mode: str
    resource_tags: Dict[str, str] = field(default_factory=dict)
    resource_arns: List[str] = field(default_factory=list)
    filters: List[Dict] = field(default_factory=list)


@dataclass
class ExperimentAction:
    name: str
    action_id: str
    description: str
    target_name: str
    parameters: Dict[str, str] = field(default_factory=dict)
    start_after: List[str] = field(default_factory=list)


@dataclass
class StopCondition:
    source: str
    value: str


class AWSFISClient:
    """Client for AWS Fault Injection Simulator."""

    def __init__(self, region_name: str = "us-east-1"):
        self.client = boto3.client("fis", region_name=region_name)
        self.region = region_name

    def create_experiment_template(
        self,
        description: str,
        role_arn: str,
        targets: List[ExperimentTarget],
        actions: List[ExperimentAction],
        stop_conditions: List[StopCondition],
        tags: Dict[str, str] = None,
    ) -> Dict:
        """Create a FIS experiment template."""
        template_targets = {}
        for target in targets:
            target_config = {
                "resourceType": target.resource_type,
                "selectionMode": target.selection_mode,
            }
            if target.resource_tags:
                target_config["resourceTags"] = target.resource_tags
            if target.resource_arns:
                target_config["resourceArns"] = target.resource_arns
            if target.filters:
                target_config["filters"] = target.filters
            template_targets[target.name] = target_config

        template_actions = {}
        for action in actions:
            action_config = {
                "actionId": action.action_id,
                "description": action.description,
                "targets": {"Instances": action.target_name},
            }
            if action.parameters:
                action_config["parameters"] = action.parameters
            if action.start_after:
                action_config["startAfter"] = action.start_after
            template_actions[action.name] = action_config

        template_stop_conditions = [
            {"source": sc.source, "value": sc.value}
            for sc in stop_conditions
        ]

        response = self.client.create_experiment_template(
            description=description,
            roleArn=role_arn,
            targets=template_targets,
            actions=template_actions,
            stopConditions=template_stop_conditions,
            tags=tags or {},
        )

        template_id = response["experimentTemplate"]["id"]
        logger.info(f"Created experiment template: {template_id}")
        return response["experimentTemplate"]

    def start_experiment(
        self,
        template_id: str,
        tags: Dict[str, str] = None,
    ) -> Dict:
        """Start an experiment from a template."""
        response = self.client.start_experiment(
            experimentTemplateId=template_id,
            tags=tags or {},
        )

        experiment_id = response["experiment"]["id"]
        logger.info(f"Started experiment: {experiment_id}")
        return response["experiment"]

    def stop_experiment(self, experiment_id: str) -> Dict:
        """Stop a running experiment."""
        response = self.client.stop_experiment(id=experiment_id)
        logger.info(f"Stopped experiment: {experiment_id}")
        return response["experiment"]

    def get_experiment(self, experiment_id: str) -> Dict:
        """Get experiment details."""
        response = self.client.get_experiment(id=experiment_id)
        return response["experiment"]

    def wait_for_experiment(
        self,
        experiment_id: str,
        timeout_seconds: int = 600,
        poll_interval: int = 10,
    ) -> Dict:
        """Wait for experiment to complete."""
        start_time = time.time()

        while True:
            experiment = self.get_experiment(experiment_id)
            state = experiment["state"]["status"]

            if state in ["completed", "stopped", "failed"]:
                logger.info(f"Experiment {experiment_id} finished with state: {state}")
                return experiment

            if time.time() - start_time > timeout_seconds:
                raise TimeoutError(
                    f"Experiment {experiment_id} did not complete within {timeout_seconds}s"
                )

            time.sleep(poll_interval)

    def list_experiments(
        self,
        max_results: int = 100,
    ) -> List[Dict]:
        """List all experiments."""
        response = self.client.list_experiments(maxResults=max_results)
        return response["experiments"]

    def list_experiment_templates(
        self,
        max_results: int = 100,
    ) -> List[Dict]:
        """List all experiment templates."""
        response = self.client.list_experiment_templates(maxResults=max_results)
        return response["experimentTemplates"]

    # Convenience methods for common experiments
    def create_ec2_stop_template(
        self,
        role_arn: str,
        target_tags: Dict[str, str],
        stop_alarm_arn: str,
        restart_after_minutes: int = 5,
        selection_count: int = 1,
    ) -> Dict:
        """Create EC2 stop experiment template."""
        return self.create_experiment_template(
            description="Stop random EC2 instances",
            role_arn=role_arn,
            targets=[
                ExperimentTarget(
                    name="ec2-targets",
                    resource_type="aws:ec2:instance",
                    selection_mode=f"COUNT({selection_count})",
                    resource_tags=target_tags,
                    filters=[{"path": "State.Name", "values": ["running"]}],
                )
            ],
            actions=[
                ExperimentAction(
                    name="stop-instances",
                    action_id="aws:ec2:stop-instances",
                    description="Stop EC2 instances",
                    target_name="ec2-targets",
                    parameters={
                        "startInstancesAfterDuration": f"PT{restart_after_minutes}M"
                    },
                )
            ],
            stop_conditions=[
                StopCondition(
                    source="aws:cloudwatch:alarm",
                    value=stop_alarm_arn,
                )
            ],
        )

    def create_cpu_stress_template(
        self,
        role_arn: str,
        target_tags: Dict[str, str],
        stop_alarm_arn: str,
        cpu_load_percent: int = 80,
        duration_minutes: int = 5,
        target_percent: int = 30,
    ) -> Dict:
        """Create CPU stress experiment template."""
        return self.create_experiment_template(
            description="CPU stress test via SSM",
            role_arn=role_arn,
            targets=[
                ExperimentTarget(
                    name="stress-targets",
                    resource_type="aws:ec2:instance",
                    selection_mode=f"PERCENT({target_percent})",
                    resource_tags=target_tags,
                )
            ],
            actions=[
                ExperimentAction(
                    name="cpu-stress",
                    action_id="aws:ssm:send-command",
                    description="Run CPU stress",
                    target_name="stress-targets",
                    parameters={
                        "documentArn": f"arn:aws:ssm:{self.region}::document/AWSFIS-Run-CPU-Stress",
                        "documentParameters": json.dumps({
                            "DurationSeconds": str(duration_minutes * 60),
                            "CPU": "0",
                            "LoadPercent": str(cpu_load_percent),
                        }),
                        "duration": f"PT{duration_minutes}M",
                    },
                )
            ],
            stop_conditions=[
                StopCondition(
                    source="aws:cloudwatch:alarm",
                    value=stop_alarm_arn,
                )
            ],
        )


# Example usage
def run_chaos_experiment():
    fis = AWSFISClient(region_name="us-east-1")

    # Create template
    template = fis.create_ec2_stop_template(
        role_arn="arn:aws:iam::123456789012:role/FISRole",
        target_tags={"Environment": "staging"},
        stop_alarm_arn="arn:aws:cloudwatch:us-east-1:123456789012:alarm:HighErrorRate",
        restart_after_minutes=5,
        selection_count=1,
    )

    # Start experiment
    experiment = fis.start_experiment(template["id"])

    # Wait for completion
    result = fis.wait_for_experiment(
        experiment["id"],
        timeout_seconds=600,
    )

    print(f"Experiment completed: {result['state']['status']}")
```

## Best Practices

### Experiment Design

- Use stop conditions to automatically halt on SLO breach
- Start with non-production environments
- Tag resources for precise targeting
- Use COUNT or PERCENT for gradual rollout

### Safety

- Implement CloudWatch alarms as stop conditions
- Use IAM conditions to limit blast radius
- Run experiments during business hours initially
- Have rollback procedures documented

### Integration

- Integrate with CI/CD pipelines
- Export results to CloudWatch
- Use EventBridge for notifications
- Track experiments in your incident management system

AWS FIS provides **native integration** with all AWS services, making it the preferred choice for AWS-centric chaos engineering.

You implement AWS-native chaos engineering with FIS for comprehensive resilience testing.