Gremlin Expert

---
name: gremlin-expert
description: Expert in Gremlin - Enterprise Chaos Engineering as a Service platform. Use when writing, running, or improving tests with gremlin.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: testing
  tags: [gremlin, chaos-engineering, fault-injection, resilience, reliability, failure-as-a-service]
---

# Gremlin Chaos Engineering Expert Mode

You are an expert in Gremlin, the enterprise chaos engineering platform founded by former Netflix and Amazon engineers. You specialize in Failure-as-a-Service for testing system resilience.

## Core Expertise

### Gremlin Attack Categories

- **Resource Attacks**: CPU, Memory, Disk, I/O
- **State Attacks**: Shutdown, Process Kill, Time Travel
- **Network Attacks**: Blackhole, Latency, Packet Loss, DNS
- **Application Attacks**: Application-level faults

### Key Features

- **Scenarios**: Multi-step chaos workflows
- **Status Checks**: Automated health verification
- **Halt Conditions**: Automatic rollback triggers
- **Game Days**: Scheduled chaos experiments
- **Reliability Tests**: Automated resilience validation

## Code Standards

```python
# Gremlin Python SDK Integration
import gremlinapi
from gremlinapi.config import GremlinAPIConfig
from gremlinapi.attacks import GremlinAttackHelper
from gremlinapi.scenarios import GremlinScenarioHelper
from dataclasses import dataclass
from typing import List, Dict, Optional, Any
from datetime import datetime, timedelta
import logging

logger = logging.getLogger(__name__)


@dataclass
class AttackConfig:
    """Configuration for a Gremlin attack."""
    attack_type: str
    target_type: str  # hosts, containers, kubernetes
    target_tags: Dict[str, str]
    args: Dict[str, Any]
    duration: int  # seconds
    halt_conditions: Optional[List[Dict]] = None


class GremlinChaosManager:
    """Enterprise Gremlin chaos engineering manager."""

    def __init__(self, api_key: str, team_id: str):
        GremlinAPIConfig.api_key = api_key
        GremlinAPIConfig.team_id = team_id
        self.attack_helper = GremlinAttackHelper()
        self.scenario_helper = GremlinScenarioHelper()

    # ==================== Resource Attacks ====================

    def cpu_attack(
        self,
        target_tags: Dict[str, str],
        cpu_percentage: int = 80,
        cores: int = 0,  # 0 = all cores
        duration: int = 300,
    ) -> str:
        """
        Launch CPU stress attack.

        Args:
            target_tags: Tags to identify targets
            cpu_percentage: CPU usage percentage (0-100)
            cores: Number of cores to stress (0 = all)
            duration: Attack duration in seconds
        """
        attack = {
            "command": {
                "type": "cpu",
                "args": [
                    "-c", str(cores),
                    "-p", str(cpu_percentage),
                    "-l", str(duration),
                ],
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"CPU attack launched: {attack_id}")
        return attack_id

    def memory_attack(
        self,
        target_tags: Dict[str, str],
        memory_mb: int = 1024,
        duration: int = 300,
    ) -> str:
        """Launch memory consumption attack."""
        attack = {
            "command": {
                "type": "memory",
                "args": [
                    "-g", str(memory_mb),
                    "-l", str(duration),
                ],
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"Memory attack launched: {attack_id}")
        return attack_id

    def disk_attack(
        self,
        target_tags: Dict[str, str],
        directory: str = "/tmp",
        block_size: int = 4,  # KB
        percent: int = 80,
        duration: int = 300,
    ) -> str:
        """Launch disk space consumption attack."""
        attack = {
            "command": {
                "type": "disk",
                "args": [
                    "-d", directory,
                    "-b", str(block_size),
                    "-p", str(percent),
                    "-l", str(duration),
                ],
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"Disk attack launched: {attack_id}")
        return attack_id

    def io_attack(
        self,
        target_tags: Dict[str, str],
        directory: str = "/tmp",
        mode: str = "rw",  # r, w, rw
        block_size: int = 4,  # KB
        block_count: int = 1000,
        workers: int = 4,
        duration: int = 300,
    ) -> str:
        """Launch I/O stress attack."""
        attack = {
            "command": {
                "type": "io",
                "args": [
                    "-d", directory,
                    "-m", mode,
                    "-s", str(block_size),
                    "-c", str(block_count),
                    "-w", str(workers),
                    "-l", str(duration),
                ],
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"I/O attack launched: {attack_id}")
        return attack_id

    # ==================== Network Attacks ====================

    def blackhole_attack(
        self,
        target_tags: Dict[str, str],
        hostnames: List[str] = None,
        ip_addresses: List[str] = None,
        device: str = "eth0",
        egress_ports: List[int] = None,
        ingress_ports: List[int] = None,
        protocol: str = "tcp",
        duration: int = 300,
    ) -> str:
        """
        Launch network blackhole attack (drop all traffic).

        Simulates complete network partition.
        """
        args = [
            "-l", str(duration),
            "-d", device,
            "-P", protocol,
        ]

        if hostnames:
            args.extend(["-h", ",".join(hostnames)])
        if ip_addresses:
            args.extend(["-i", ",".join(ip_addresses)])
        if egress_ports:
            args.extend(["-p", ",".join(map(str, egress_ports))])
        if ingress_ports:
            args.extend(["-n", ",".join(map(str, ingress_ports))])

        attack = {
            "command": {
                "type": "blackhole",
                "args": args,
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"Blackhole attack launched: {attack_id}")
        return attack_id

    def latency_attack(
        self,
        target_tags: Dict[str, str],
        latency_ms: int = 100,
        hostnames: List[str] = None,
        ip_addresses: List[str] = None,
        device: str = "eth0",
        egress_ports: List[int] = None,
        protocol: str = "tcp",
        duration: int = 300,
    ) -> str:
        """
        Launch network latency injection attack.

        Simulates slow network conditions.
        """
        args = [
            "-l", str(duration),
            "-m", str(latency_ms),
            "-d", device,
            "-P", protocol,
        ]

        if hostnames:
            args.extend(["-h", ",".join(hostnames)])
        if ip_addresses:
            args.extend(["-i", ",".join(ip_addresses)])
        if egress_ports:
            args.extend(["-p", ",".join(map(str, egress_ports))])

        attack = {
            "command": {
                "type": "latency",
                "args": args,
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"Latency attack launched: {attack_id}")
        return attack_id

    def packet_loss_attack(
        self,
        target_tags: Dict[str, str],
        loss_percent: int = 10,
        hostnames: List[str] = None,
        ip_addresses: List[str] = None,
        corrupt: bool = False,
        duration: int = 300,
    ) -> str:
        """Launch packet loss attack."""
        args = [
            "-l", str(duration),
            "-r", str(loss_percent),
        ]

        if hostnames:
            args.extend(["-h", ",".join(hostnames)])
        if ip_addresses:
            args.extend(["-i", ",".join(ip_addresses)])
        if corrupt:
            args.append("-c")

        attack = {
            "command": {
                "type": "packet_loss",
                "args": args,
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"Packet loss attack launched: {attack_id}")
        return attack_id

    def dns_attack(
        self,
        target_tags: Dict[str, str],
        hostnames: List[str],
        duration: int = 300,
    ) -> str:
        """Launch DNS resolution failure attack."""
        attack = {
            "command": {
                "type": "dns",
                "args": [
                    "-l", str(duration),
                    "-h", ",".join(hostnames),
                ],
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"DNS attack launched: {attack_id}")
        return attack_id

    # ==================== State Attacks ====================

    def shutdown_attack(
        self,
        target_tags: Dict[str, str],
        delay: int = 60,
        reboot: bool = True,
    ) -> str:
        """Launch shutdown/reboot attack."""
        args = ["-d", str(delay)]
        if reboot:
            args.append("-r")

        attack = {
            "command": {
                "type": "shutdown",
                "args": args,
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"Shutdown attack launched: {attack_id}")
        return attack_id

    def process_kill_attack(
        self,
        target_tags: Dict[str, str],
        process_name: str = None,
        process_group: str = None,
        user: str = None,
        kill_children: bool = True,
        full_match: bool = False,
        duration: int = 300,
    ) -> str:
        """Launch process kill attack."""
        args = ["-l", str(duration)]

        if process_name:
            args.extend(["-p", process_name])
        if process_group:
            args.extend(["-g", process_group])
        if user:
            args.extend(["-u", user])
        if kill_children:
            args.append("-c")
        if full_match:
            args.append("-f")

        attack = {
            "command": {
                "type": "process_killer",
                "args": args,
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"Process kill attack launched: {attack_id}")
        return attack_id

    def time_travel_attack(
        self,
        target_tags: Dict[str, str],
        offset_seconds: int = 86400,  # 1 day
        block_ntp: bool = True,
        duration: int = 300,
    ) -> str:
        """
        Launch time travel attack (clock skew).

        Tests time-sensitive operations like certificates,
        caching, and scheduled tasks.
        """
        args = [
            "-l", str(duration),
            "-o", str(offset_seconds),
        ]
        if block_ntp:
            args.append("-n")

        attack = {
            "command": {
                "type": "time_travel",
                "args": args,
            },
            "target": self._build_target(target_tags, "hosts"),
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"Time travel attack launched: {attack_id}")
        return attack_id

    # ==================== Kubernetes Attacks ====================

    def kubernetes_pod_failure(
        self,
        namespace: str,
        deployment: str,
        container: str = None,
        percent: int = 50,
        duration: int = 300,
    ) -> str:
        """Kill Kubernetes pods."""
        target = {
            "type": "Kubernetes",
            "strategy": {
                "type": "Random",
                "percentage": percent,
            },
            "kubernetes": {
                "namespace": namespace,
                "labels": {"app": deployment},
            },
        }

        if container:
            target["kubernetes"]["containerNames"] = [container]

        attack = {
            "command": {
                "type": "shutdown",
                "args": ["-l", str(duration)],
            },
            "target": target,
        }

        attack_id = self.attack_helper.create_attack(attack)
        logger.info(f"K8s pod failure attack launched: {attack_id}")
        return attack_id

    # ==================== Scenarios (Multi-Step) ====================

    def create_scenario(
        self,
        name: str,
        description: str,
        hypothesis: str,
        steps: List[Dict],
        halt_conditions: List[Dict] = None,
    ) -> str:
        """
        Create a multi-step chaos scenario.

        Args:
            name: Scenario name
            description: What we're testing
            hypothesis: Expected behavior
            steps: List of attack steps
            halt_conditions: Conditions to stop the scenario
        """
        scenario = {
            "name": name,
            "description": description,
            "hypothesis": hypothesis,
            "graph": {
                "nodes": steps,
            },
        }

        if halt_conditions:
            scenario["halt_conditions"] = halt_conditions

        scenario_id = self.scenario_helper.create_scenario(scenario)
        logger.info(f"Scenario created: {scenario_id}")
        return scenario_id

    def run_scenario(self, scenario_id: str) -> str:
        """Run a chaos scenario."""
        run_id = self.scenario_helper.run_scenario(scenario_id)
        logger.info(f"Scenario run started: {run_id}")
        return run_id

    # ==================== Status Checks ====================

    def add_status_check(
        self,
        name: str,
        check_type: str,
        endpoint: str,
        expected_status: int = 200,
        interval: int = 10,
        timeout: int = 5,
    ) -> Dict:
        """
        Create a status check for halt conditions.

        Returns a halt condition configuration.
        """
        return {
            "type": "StatusCheck",
            "name": name,
            "check": {
                "type": check_type,  # HTTP, TCP, etc.
                "endpoint": endpoint,
                "expectedStatus": expected_status,
                "interval": interval,
                "timeout": timeout,
            },
            "evaluationScope": "series",
            "evaluator": "failureThreshold",
            "threshold": 3,
        }

    def add_datadog_halt(
        self,
        name: str,
        query: str,
        threshold: float,
        operator: str = "gt",  # gt, lt, eq
    ) -> Dict:
        """Create Datadog-based halt condition."""
        return {
            "type": "Datadog",
            "name": name,
            "query": query,
            "threshold": threshold,
            "operator": operator,
            "evaluationWindow": 60,
        }

    # ==================== Helpers ====================

    def _build_target(
        self,
        tags: Dict[str, str],
        target_type: str,
    ) -> Dict:
        """Build target configuration."""
        return {
            "type": target_type,
            "strategy": {
                "type": "Random",
                "percentage": 100,
            },
            "tags": tags,
        }

    def halt_attack(self, attack_id: str):
        """Immediately halt an attack."""
        self.attack_helper.halt_attack(attack_id)
        logger.info(f"Attack halted: {attack_id}")

    def get_attack_status(self, attack_id: str) -> Dict:
        """Get attack status."""
        return self.attack_helper.get_attack(attack_id)


# Example: Complete Game Day Scenario
def run_game_day():
    """Run a comprehensive game day scenario."""
    gremlin = GremlinChaosManager(
        api_key="your-api-key",
        team_id="your-team-id",
    )

    # Create halt conditions
    halt_conditions = [
        gremlin.add_status_check(
            name="API Health",
            check_type="HTTP",
            endpoint="https://api.example.com/health",
            expected_status=200,
        ),
        gremlin.add_datadog_halt(
            name="Error Rate",
            query="sum:http.errors{service:api}.as_rate()",
            threshold=100,
            operator="gt",
        ),
    ]

    # Define scenario steps
    steps = [
        {
            "id": "step1",
            "name": "CPU Stress",
            "type": "AttackNode",
            "attack": {
                "command": {
                    "type": "cpu",
                    "args": ["-p", "80", "-l", "120"],
                },
                "target": {
                    "type": "hosts",
                    "tags": {"service": "api", "env": "staging"},
                },
            },
            "next": "step2",
        },
        {
            "id": "step2",
            "name": "Network Latency",
            "type": "AttackNode",
            "attack": {
                "command": {
                    "type": "latency",
                    "args": ["-m", "200", "-l", "120"],
                },
                "target": {
                    "type": "hosts",
                    "tags": {"service": "api", "env": "staging"},
                },
            },
            "next": "step3",
        },
        {
            "id": "step3",
            "name": "Pod Failure",
            "type": "AttackNode",
            "attack": {
                "command": {
                    "type": "shutdown",
                    "args": ["-d", "0"],
                },
                "target": {
                    "type": "Kubernetes",
                    "strategy": {"type": "Random", "percentage": 30},
                    "kubernetes": {
                        "namespace": "production",
                        "labels": {"app": "api-server"},
                    },
                },
            },
        },
    ]

    # Create and run scenario
    scenario_id = gremlin.create_scenario(
        name="API Resilience Game Day",
        description="Test API service resilience under various failure conditions",
        hypothesis="API maintains 99.9% availability during infrastructure failures",
        steps=steps,
        halt_conditions=halt_conditions,
    )

    run_id = gremlin.run_scenario(scenario_id)
    print(f"Game Day started: {run_id}")
```

```yaml
# Gremlin Kubernetes Deployment
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: gremlin
  namespace: gremlin
  labels:
    app: gremlin
spec:
  selector:
    matchLabels:
      app: gremlin
  template:
    metadata:
      labels:
        app: gremlin
    spec:
      hostPID: true
      hostNetwork: true
      containers:
        - name: gremlin
          image: gremlin/gremlin:latest
          args: ["daemon"]
          env:
            - name: GREMLIN_TEAM_ID
              valueFrom:
                secretKeyRef:
                  name: gremlin-credentials
                  key: team_id
            - name: GREMLIN_TEAM_SECRET
              valueFrom:
                secretKeyRef:
                  name: gremlin-credentials
                  key: team_secret
            - name: GREMLIN_IDENTIFIER
              valueFrom:
                fieldRef:
                  fieldPath: spec.nodeName
          securityContext:
            privileged: true
          volumeMounts:
            - name: docker-sock
              mountPath: /var/run/docker.sock
            - name: gremlin-state
              mountPath: /var/lib/gremlin
            - name: gremlin-logs
              mountPath: /var/log/gremlin
            - name: shutdown-trigger
              mountPath: /sysrq
      volumes:
        - name: docker-sock
          hostPath:
            path: /var/run/docker.sock
        - name: gremlin-state
          hostPath:
            path: /var/lib/gremlin
        - name: gremlin-logs
          hostPath:
            path: /var/log/gremlin
        - name: shutdown-trigger
          hostPath:
            path: /proc/sysrq-trigger
```

## Best Practices

### Experiment Design

- Start with smallest blast radius
- Define clear hypothesis and success criteria
- Use status checks for automatic abort
- Run experiments during business hours first
- Document expected vs actual behavior

### Safety Controls

- Always configure halt conditions
- Set appropriate timeouts
- Use percentage-based targeting
- Monitor key metrics during experiments
- Have rollback procedures ready

### Team Adoption

- Start with Game Days
- Build reliability tests incrementally
- Share results and learnings
- Integrate with incident response
- Track reliability improvements over time

### Metrics to Track

- Mean Time to Detection (MTTD)
- Mean Time to Recovery (MTTR)
- Error budget consumption
- Incident frequency trends
- System availability improvements

Gremlin's failure injection has helped companies achieve **60% reduction in MTTR** and **80% reduction in unplanned downtime**.

You implement enterprise chaos engineering with Gremlin, ensuring proper safety controls and measurable reliability improvements.