从零落地 Chaos Mesh：构建高可用微服务的混沌工程平台（一）

+---------------------+
|     Vue 前端        | ← 用户操作界面（实验创建/执行/报告）
+----------+----------+
           |
+----------v----------+
|   FastAPI 后端      | ← 实验编排、权限控制、结果聚合
+----------+----------+
           |
+----------v----------+
|   Chaos Mesh API    | ← 通过 k8s client 操作 CRD
+----------+----------+
           |
+----------v----------+
|   Kubernetes 集群   | ← 运行 Chaos Daemon 和目标应用
+---------------------+

# 添加 repo
helm repo add chaos-mesh https://charts.chaos-mesh.org

# 创建 namespace
kubectl create ns chaos-testing

# 关键配置 values.yaml
---
chaosDaemon:
  # 启用安全模式（必须！）
  securityMode: true
  # 限制 Chaos Daemon 只在特定节点运行（打污点）
  nodeSelector:
    chaos-ready: "true"

controllerManager:
  # 禁用 Dashboard（我们自研前端）
  enableDashboard: false

# 启用所需 Chaos 类型
enabledChaosList:
  - pod
  - network
  - stress
  - time

helm install chaos-mesh chaos-mesh/chaos-mesh \
  --namespace=chaos-testing \
  --version v2.7.0 \
  -f values.yaml

# models/experiment.py
class NetworkDelayExperiment(BaseModel):
    name: str
    namespace: str
    target_label: Dict[str, str]  # 如 {"app": "order-service"}
    delay_ms: int = 200
    duration: str = "5m"  # 支持 5m, 1h 等
    scheduler: Optional[str] = None  # cron 表达式

# services/chaos_mesh.py
def build_network_chaos_cr(experiment: NetworkDelayExperiment) -> dict:
    return {
        "apiVersion": "chaos-mesh.org/v1alpha1",
        "kind": "NetworkChaos",
        "metadata": {
            "name": experiment.name,
            "namespace": "chaos-testing"
        },
        "spec": {
            "action": "delay",
            "mode": "all",  # 或 one / fixed
            "selector": {
                "namespaces": [experiment.namespace],
                "labelSelectors": experiment.target_label
            },
            "delay": {
                "latency": f"{experiment.delay_ms}ms",
                "correlation": "0",
                "jitter": "0ms"
            },
            "duration": experiment.duration,
            "scheduler": {"cron": experiment.scheduler} if experiment.scheduler else None
        }
    }

from kubernetes import client

def apply_chaos_experiment(cr: dict):
    api = client.CustomObjectsApi()
    api.create_namespaced_custom_object(
        group="chaos-mesh.org",
        version="v1alpha1",
        namespace="chaos-testing",
        plural=f"{cr['kind'].lower()}s",
        body=cr
    )

<template>
  <ServiceTopology @select="onServiceSelect" />
  <ChaosForm :target-label="selectedLabel" />
</template>

<script setup>
const selectedLabel = ref({})

function onServiceSelect(label) {
  selectedLabel.value = label
}
</script>

<el-form :model="form">
  <el-form-item label="延迟时间 (ms)">
    <el-input-number v-model="form.delay_ms" :min="10" :max="5000" />
  </el-form-item>
  <el-form-item label="持续时间">
    <el-select v-model="form.duration">
      <el-option value="1m" />
      <el-option value="5m" />
      <el-option value="10m" />
    </el-select>
  </el-form-item>
</el-form>