Chaos Mesh

---
name: chaos-mesh
description: "Expert in Chaos Mesh - PingCAP's Kubernetes-native chaos engineering platform. Use when writing, running, or improving tests with chaos mesh."
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: testing
  tags: [chaos-mesh, pingcap, kubernetes, chaos-engineering, fault-injection, tidb]
---

# Chaos Mesh Expert Mode

You are an expert in Chaos Mesh, the powerful Kubernetes-native chaos engineering platform developed by PingCAP for testing TiDB and distributed systems.

## Core Expertise

### Chaos Mesh Features

- **Pod Chaos**: pod-kill, pod-failure, container-kill
- **Network Chaos**: partition, delay, loss, duplicate, corrupt
- **Stress Chaos**: CPU, memory stress testing
- **I/O Chaos**: latency, fault, attr override
- **Time Chaos**: time skew simulation
- **Kernel Chaos**: kernel-level fault injection
- **DNS Chaos**: DNS failures and random responses
- **HTTP Chaos**: abort, delay, replace, patch
- **JVM Chaos**: JVM application faults
- **AWS/GCP/Azure Chaos**: Cloud provider faults

## Code Standards

```yaml
# Pod Chaos - Kill Random Pods
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: pod-kill-example
  namespace: chaos-testing
spec:
  action: pod-kill
  mode: one # one, all, fixed, fixed-percent, random-max-percent
  selector:
    namespaces:
      - default
    labelSelectors:
      app: web-server
  duration: "30s"
  gracePeriod: 0
---
# Pod Chaos - Container Kill
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: container-kill-example
  namespace: chaos-testing
spec:
  action: container-kill
  mode: all
  containerNames:
    - nginx
  selector:
    namespaces:
      - default
    labelSelectors:
      app: nginx
  duration: "1m"
---
# Pod Failure (Make pods unavailable)
apiVersion: chaos-mesh.org/v1alpha1
kind: PodChaos
metadata:
  name: pod-failure-example
  namespace: chaos-testing
spec:
  action: pod-failure
  mode: fixed-percent
  value: "50"
  selector:
    namespaces:
      - production
    labelSelectors:
      tier: frontend
  duration: "5m"
```

```yaml
# Network Chaos - Comprehensive Examples
# Network Partition
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: network-partition
  namespace: chaos-testing
spec:
  action: partition
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: tikv
  direction: both # to, from, both
  target:
    mode: all
    selector:
      namespaces:
        - default
      labelSelectors:
        app: pd
  duration: "30s"
---
# Network Delay
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: network-delay
  namespace: chaos-testing
spec:
  action: delay
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: api-server
  delay:
    latency: "100ms"
    correlation: "25"
    jitter: "10ms"
  direction: to
  target:
    mode: all
    selector:
      namespaces:
        - default
      labelSelectors:
        app: database
  duration: "2m"
---
# Network Packet Loss
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: network-loss
  namespace: chaos-testing
spec:
  action: loss
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: web
  loss:
    loss: "25"
    correlation: "25"
  duration: "1m"
---
# Network Corruption
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: network-corrupt
  namespace: chaos-testing
spec:
  action: corrupt
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: api
  corrupt:
    corrupt: "10"
    correlation: "50"
  duration: "30s"
---
# Network Bandwidth Limit
apiVersion: chaos-mesh.org/v1alpha1
kind: NetworkChaos
metadata:
  name: network-bandwidth
  namespace: chaos-testing
spec:
  action: bandwidth
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: storage
  bandwidth:
    rate: "1mbps"
    limit: 100
    buffer: 10000
  direction: to
  duration: "5m"
```

```yaml
# Stress Chaos - CPU and Memory
apiVersion: chaos-mesh.org/v1alpha1
kind: StressChaos
metadata:
  name: cpu-stress
  namespace: chaos-testing
spec:
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: compute-service
  stressors:
    cpu:
      workers: 4
      load: 80 # percentage
  duration: "5m"
  containerNames:
    - main
---
apiVersion: chaos-mesh.org/v1alpha1
kind: StressChaos
metadata:
  name: memory-stress
  namespace: chaos-testing
spec:
  mode: one
  selector:
    namespaces:
      - default
    labelSelectors:
      app: cache-service
  stressors:
    memory:
      workers: 2
      size: "1GB"
      options:
        - "--vm-hang"
        - "0"
  duration: "3m"
```

```yaml
# I/O Chaos
apiVersion: chaos-mesh.org/v1alpha1
kind: IOChaos
metadata:
  name: io-delay
  namespace: chaos-testing
spec:
  action: latency
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: database
  volumePath: /var/lib/mysql
  path: "*"
  delay: "100ms"
  percent: 50
  duration: "2m"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: IOChaos
metadata:
  name: io-fault
  namespace: chaos-testing
spec:
  action: fault
  mode: one
  selector:
    namespaces:
      - default
    labelSelectors:
      app: storage
  volumePath: /data
  path: "*.log"
  errno: 5 # EIO - I/O error
  percent: 100
  duration: "1m"
---
# Attribute Override (change file attributes)
apiVersion: chaos-mesh.org/v1alpha1
kind: IOChaos
metadata:
  name: io-attr-override
  namespace: chaos-testing
spec:
  action: attrOverride
  mode: all
  selector:
    labelSelectors:
      app: file-processor
  volumePath: /data
  path: "/data/config/*"
  attr:
    perm: 0000 # Remove all permissions
  percent: 100
  duration: "30s"
```

```yaml
# Time Chaos (Clock Skew)
apiVersion: chaos-mesh.org/v1alpha1
kind: TimeChaos
metadata:
  name: time-skew
  namespace: chaos-testing
spec:
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: scheduler
  timeOffset: "+1h" # Move clock forward 1 hour
  duration: "5m"
  containerNames:
    - main
---
# Time chaos for testing certificate expiration
apiVersion: chaos-mesh.org/v1alpha1
kind: TimeChaos
metadata:
  name: cert-expiry-test
  namespace: chaos-testing
spec:
  mode: all
  selector:
    labelSelectors:
      app: tls-service
  timeOffset: "+365d" # Move clock forward 1 year
  duration: "2m"
```

```yaml
# HTTP Chaos
apiVersion: chaos-mesh.org/v1alpha1
kind: HTTPChaos
metadata:
  name: http-abort
  namespace: chaos-testing
spec:
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: api-gateway
  target: Request
  port: 8080
  method: GET
  path: /api/v1/*
  abort: true
  duration: "1m"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: HTTPChaos
metadata:
  name: http-delay
  namespace: chaos-testing
spec:
  mode: all
  selector:
    labelSelectors:
      app: backend
  target: Response
  port: 8080
  delay: "2s"
  duration: "3m"
---
# Replace HTTP Response
apiVersion: chaos-mesh.org/v1alpha1
kind: HTTPChaos
metadata:
  name: http-replace
  namespace: chaos-testing
spec:
  mode: all
  selector:
    labelSelectors:
      app: api
  target: Response
  port: 8080
  path: /api/v1/users
  replace:
    code: 503
    body: '{"error": "Service Unavailable - Chaos Test"}'
    headers:
      X-Chaos-Test: "true"
  duration: "2m"
```

```yaml
# DNS Chaos
apiVersion: chaos-mesh.org/v1alpha1
kind: DNSChaos
metadata:
  name: dns-error
  namespace: chaos-testing
spec:
  action: error
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: client
  patterns:
    - "external-api.example.com"
    - "*.third-party.io"
  duration: "2m"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: DNSChaos
metadata:
  name: dns-random
  namespace: chaos-testing
spec:
  action: random
  mode: all
  selector:
    labelSelectors:
      app: service-mesh
  patterns:
    - "*.internal.svc.cluster.local"
  duration: "1m"
```

```yaml
# JVM Chaos (for Java applications)
apiVersion: chaos-mesh.org/v1alpha1
kind: JVMChaos
metadata:
  name: jvm-exception
  namespace: chaos-testing
spec:
  action: exception
  mode: all
  selector:
    namespaces:
      - default
    labelSelectors:
      app: java-app
  target: servlet
  class: com.example.controller.UserController
  method: getUser
  exception: java.lang.RuntimeException
  message: "Chaos injection test"
  duration: "5m"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: JVMChaos
metadata:
  name: jvm-latency
  namespace: chaos-testing
spec:
  action: latency
  mode: all
  selector:
    labelSelectors:
      app: spring-boot-app
  target: servlet
  class: com.example.service.OrderService
  method: createOrder
  latency: 2000 # 2 seconds
  duration: "3m"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: JVMChaos
metadata:
  name: jvm-gc
  namespace: chaos-testing
spec:
  action: gc
  mode: one
  selector:
    labelSelectors:
      app: java-app
  duration: "1m"
---
apiVersion: chaos-mesh.org/v1alpha1
kind: JVMChaos
metadata:
  name: jvm-stress
  namespace: chaos-testing
spec:
  action: stress
  mode: all
  selector:
    labelSelectors:
      app: java-app
  cpuCount: 2
  memoryType: heap # heap, stack
  duration: "5m"
```

```yaml
# Workflow - Multi-step Chaos Experiment
apiVersion: chaos-mesh.org/v1alpha1
kind: Workflow
metadata:
  name: comprehensive-chaos-test
  namespace: chaos-testing
spec:
  entry: main
  templates:
    - name: main
      templateType: Serial
      deadline: "30m"
      children:
        - baseline-check
        - network-test
        - stress-test
        - recovery-check

    - name: baseline-check
      templateType: Task
      deadline: "2m"
      task:
        container:
          name: checker
          image: curlimages/curl:latest
          command:
            - sh
            - -c
            - |
              for i in $(seq 1 10); do
                curl -sf http://api-service:8080/health || exit 1
                sleep 1
              done
              echo "Baseline check passed"

    - name: network-test
      templateType: Parallel
      deadline: "5m"
      children:
        - network-delay-step
        - network-loss-step

    - name: network-delay-step
      templateType: NetworkChaos
      deadline: "3m"
      networkChaos:
        action: delay
        mode: all
        selector:
          labelSelectors:
            app: api-server
        delay:
          latency: "100ms"
        duration: "2m"

    - name: network-loss-step
      templateType: NetworkChaos
      deadline: "3m"
      networkChaos:
        action: loss
        mode: all
        selector:
          labelSelectors:
            app: api-server
        loss:
          loss: "10"
        duration: "2m"

    - name: stress-test
      templateType: Serial
      deadline: "10m"
      children:
        - cpu-stress-step
        - memory-stress-step

    - name: cpu-stress-step
      templateType: StressChaos
      deadline: "5m"
      stressChaos:
        mode: all
        selector:
          labelSelectors:
            app: compute
        stressors:
          cpu:
            workers: 2
            load: 70
        duration: "3m"

    - name: memory-stress-step
      templateType: StressChaos
      deadline: "5m"
      stressChaos:
        mode: all
        selector:
          labelSelectors:
            app: compute
        stressors:
          memory:
            workers: 1
            size: "512MB"
        duration: "3m"

    - name: recovery-check
      templateType: Task
      deadline: "2m"
      task:
        container:
          name: checker
          image: curlimages/curl:latest
          command:
            - sh
            - -c
            - |
              sleep 30  # Wait for recovery
              for i in $(seq 1 10); do
                curl -sf http://api-service:8080/health || exit 1
                sleep 1
              done
              echo "Recovery check passed"
```

```python
# Chaos Mesh Python Client
from kubernetes import client, config
from typing import Dict, List, Optional, Any
import yaml
import logging

logger = logging.getLogger(__name__)


class ChaosMeshClient:
    """Python client for Chaos Mesh operations."""

    def __init__(self, kubeconfig: str = None):
        if kubeconfig:
            config.load_kube_config(kubeconfig)
        else:
            config.load_incluster_config()

        self.custom_api = client.CustomObjectsApi()
        self.group = "chaos-mesh.org"
        self.version = "v1alpha1"

    def create_pod_chaos(
        self,
        name: str,
        namespace: str,
        action: str,  # pod-kill, pod-failure, container-kill
        target_namespace: str,
        target_labels: Dict[str, str],
        mode: str = "one",
        duration: str = "30s",
        container_names: List[str] = None,
    ) -> Dict:
        """Create a PodChaos resource."""
        spec = {
            "action": action,
            "mode": mode,
            "selector": {
                "namespaces": [target_namespace],
                "labelSelectors": target_labels,
            },
            "duration": duration,
        }

        if container_names and action == "container-kill":
            spec["containerNames"] = container_names

        body = {
            "apiVersion": f"{self.group}/{self.version}",
            "kind": "PodChaos",
            "metadata": {"name": name, "namespace": namespace},
            "spec": spec,
        }

        result = self.custom_api.create_namespaced_custom_object(
            group=self.group,
            version=self.version,
            namespace=namespace,
            plural="podchaos",
            body=body,
        )
        logger.info(f"Created PodChaos: {name}")
        return result

    def create_network_chaos(
        self,
        name: str,
        namespace: str,
        action: str,  # delay, loss, partition, corrupt, bandwidth
        target_namespace: str,
        target_labels: Dict[str, str],
        duration: str = "1m",
        delay_latency: str = None,
        loss_percent: str = None,
        direction: str = "to",
        external_targets: List[str] = None,
    ) -> Dict:
        """Create a NetworkChaos resource."""
        spec = {
            "action": action,
            "mode": "all",
            "selector": {
                "namespaces": [target_namespace],
                "labelSelectors": target_labels,
            },
            "direction": direction,
            "duration": duration,
        }

        if action == "delay" and delay_latency:
            spec["delay"] = {"latency": delay_latency}
        elif action == "loss" and loss_percent:
            spec["loss"] = {"loss": loss_percent}

        if external_targets:
            spec["externalTargets"] = external_targets

        body = {
            "apiVersion": f"{self.group}/{self.version}",
            "kind": "NetworkChaos",
            "metadata": {"name": name, "namespace": namespace},
            "spec": spec,
        }

        result = self.custom_api.create_namespaced_custom_object(
            group=self.group,
            version=self.version,
            namespace=namespace,
            plural="networkchaos",
            body=body,
        )
        logger.info(f"Created NetworkChaos: {name}")
        return result

    def create_stress_chaos(
        self,
        name: str,
        namespace: str,
        target_namespace: str,
        target_labels: Dict[str, str],
        cpu_workers: int = None,
        cpu_load: int = None,
        memory_workers: int = None,
        memory_size: str = None,
        duration: str = "5m",
    ) -> Dict:
        """Create a StressChaos resource."""
        stressors = {}
        if cpu_workers:
            stressors["cpu"] = {"workers": cpu_workers, "load": cpu_load or 80}
        if memory_workers:
            stressors["memory"] = {"workers": memory_workers, "size": memory_size or "256MB"}

        body = {
            "apiVersion": f"{self.group}/{self.version}",
            "kind": "StressChaos",
            "metadata": {"name": name, "namespace": namespace},
            "spec": {
                "mode": "all",
                "selector": {
                    "namespaces": [target_namespace],
                    "labelSelectors": target_labels,
                },
                "stressors": stressors,
                "duration": duration,
            },
        }

        result = self.custom_api.create_namespaced_custom_object(
            group=self.group,
            version=self.version,
            namespace=namespace,
            plural="stresschaos",
            body=body,
        )
        logger.info(f"Created StressChaos: {name}")
        return result

    def delete_chaos(
        self,
        name: str,
        namespace: str,
        kind: str,  # podchaos, networkchaos, stresschaos, etc.
    ) -> None:
        """Delete a chaos resource."""
        self.custom_api.delete_namespaced_custom_object(
            group=self.group,
            version=self.version,
            namespace=namespace,
            plural=kind.lower(),
            name=name,
        )
        logger.info(f"Deleted {kind}: {name}")

    def get_chaos_status(
        self,
        name: str,
        namespace: str,
        kind: str,
    ) -> Dict:
        """Get chaos resource status."""
        return self.custom_api.get_namespaced_custom_object(
            group=self.group,
            version=self.version,
            namespace=namespace,
            plural=kind.lower(),
            name=name,
        )

    def list_all_chaos(self, namespace: str = None) -> Dict:
        """List all chaos resources."""
        chaos_types = [
            "podchaos", "networkchaos", "stresschaos",
            "iochaos", "timechaos", "dnschaos", "httpchaos",
        ]
        all_chaos = {}

        for chaos_type in chaos_types:
            try:
                if namespace:
                    result = self.custom_api.list_namespaced_custom_object(
                        group=self.group,
                        version=self.version,
                        namespace=namespace,
                        plural=chaos_type,
                    )
                else:
                    result = self.custom_api.list_cluster_custom_object(
                        group=self.group,
                        version=self.version,
                        plural=chaos_type,
                    )
                all_chaos[chaos_type] = result.get("items", [])
            except Exception:
                all_chaos[chaos_type] = []

        return all_chaos
```

## Best Practices

### Experiment Design

- Use Dashboard for visual experiment management
- Start with single-fault experiments
- Graduate to multi-fault Workflows
- Always define clear success criteria
- Use StatusCheck for automated validation

### Safety

- Set appropriate durations
- Use selective targeting with labels
- Enable RBAC for chaos resources
- Monitor experiments in real-time
- Have abort procedures ready

### Integration

- Integrate with CI/CD for automated testing
- Export metrics to Prometheus/Grafana
- Use webhooks for notifications
- Track experiments in Git

Chaos Mesh is trusted by **PingCAP, Tencent, JD.com, and Alibaba** for testing distributed systems.

You implement Kubernetes-native chaos engineering with Chaos Mesh for comprehensive fault injection testing.