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

1. 假设（Hypothesis）  
   → “当订单服务 Redis 延迟 > 300ms 时，API 错误率应 < 1%，且自动降级到本地缓存”

2. 注入（Injection）  
   → 使用 Chaos Mesh 对 redis-pod 注入 network-delay (500ms)

3. 验证（Verification）  
   → 自动采集指标：error_rate, p99_latency, fallback_hit_count
   → 判断是否满足假设

# models/experiment.py
class MetricRule(BaseModel):
    metric_name: str          # 如 "http_requests_error_rate"
    query: str                # PromQL 表达式
    threshold: float          # 阈值，如 0.01（1%）
    operator: Literal["<", ">", "<=", ">="] = "<"

class ChaosExperimentTemplate(BaseModel):
    name: str
    type: str  # "network", "pod", "stress"...
    config: Dict[str, Any]     # 故障参数
    verification_rules: List[MetricRule]  # ← 新增！
    target_service: str        # 如 "order-service"

{
  "name": "order-redis-delay-500ms",
  "type": "network",
  "config": {
    "target_label": {"app": "redis", "role": "order-cache"},
    "delay_ms": 500,
    "duration": "3m"
  },
  "verification_rules": [
    {
      "metric_name": "order_api_error_rate",
      "query": "rate(http_requests_total{service=\"order\", status=~\"5..\"}[1m]) / rate(http_requests_total{service=\"order\"}[1m])",
      "threshold": 0.01,
      "operator": "<"
    },
    {
      "metric_name": "fallback_cache_hit",
      "query": "sum(increase(order_fallback_cache_hit_total[3m]))",
      "threshold": 100,
      "operator": ">"
    }
  ],
  "target_service": "order-service"
}

# services/verifier.py
from prometheus_api_client import PrometheusConnect

class VerificationEngine:
    def __init__(self):
        self.prom = PrometheusConnect(url="http://prometheus.monitoring.svc")

    def verify_experiment(self, experiment: ChaosExperimentTemplate) -> List[dict]:
        results = []
        for rule in experiment.verification_rules:
            try:
                # 执行 PromQL 查询（时间范围 = 实验期间）
                start_time = experiment.start_time
                end_time = experiment.end_time
                result = self.prom.custom_query_range(
                    query=rule.query,
                    start_time=start_time,
                    end_time=end_time,
                    step="15s"
                )

                # 提取最大值/平均值（根据规则类型）
                values = [float(v[1]) for v in result[0]["values"]]
                max_value = max(values) if values else 0

                # 判断是否通过
                passed = eval(f"{max_value} {rule.operator} {rule.threshold}")

                results.append({
                    "metric": rule.metric_name,
                    "actual": max_value,
                    "threshold": rule.threshold,
                    "operator": rule.operator,
                    "passed": passed
                })
            except Exception as e:
                results.append({
                    "metric": rule.metric_name,
                    "error": str(e),
                    "passed": False
                })
        return results

# tasks/chaos.py
@celery.task
def run_chaos_experiment(experiment_id: str):
    # 1. 获取实验配置
    exp = get_experiment_by_id(experiment_id)

    # 2. 提交 Chaos Mesh CR
    apply_chaos_experiment(exp.config)

    # 3. 等待实验结束（轮询状态）
    wait_for_experiment_complete(exp.name)

    # 4. 自动验证
    verifier = VerificationEngine()
    results = verifier.verify_experiment(exp)

    # 5. 更新数据库 & 通知
    update_experiment_result(experiment_id, results)
    if not all(r["passed"] for r in results):
        send_alert_to_slack(f"❌ 实验 {exp.name} 验证失败！")

<template>
  <div class="experiment-report">
    <h3>验证结果</h3>
    <el-table :data="verificationResults">
      <el-table-column prop="metric" label="指标" />
      <el-table-column label="实际值" width="120">
        <template #default="{ row }">
          <span :style="{ color: row.passed ? 'green' : 'red' }">
            {{ row.actual.toFixed(4) }}
          </span>
        </template>
      </el-table-column>
      <el-table-column prop="threshold" label="阈值" />
      <el-table-column label="状态" width="80">
        <template #default="{ row }">
          <el-tag :type="row.passed ? 'success' : 'danger'">
            {{ row.passed ? '通过' : '失败' }}
          </el-tag>
        </template>
      </el-table-column>
    </el-table>

    <div class="metrics-chart">
      <EChart :option="chartOption" />
    </div>
  </div>
</template>

# 生成 Grafana 深链接
def get_grafana_link(service: str, start: int, end: int) -> str:
    return (
        f"https://grafana.example.com/d/order-dashboard?var-service={service}"
        f"&from={start*1000}&to={end*1000}&refresh=10s"
    )

def validate_promql(query: str) -> bool:
    try:
        result = prom.custom_query(query, params={"time": time.time()})
        return True
    except:
        return False