多云环境下混沌工程与压测的协同剖析：从指标到根因的闭环

chaos-pytest/
├── requirements.txt
├── run.py                 # 启动所有服务+混沌+压测
├── config.py              # 全局配置（端口、混沌概率等）
├── services/
│   ├── __init__.py
│   ├── service_a.py       # 入口服务，调用B
│   ├── service_b.py       # 中间服务，调用C
│   └── service_c.py       # 叶子服务
├── chaos/
│   ├── __init__.py
│   └── injector.py        # 混沌注入函数（延迟&错误）
├── loadtesting/
│   ├── __init__.py
│   └── load_gen.py        # 异步压测生成器
├── monitoring/
│   ├── __init__.py
│   └── metrics.py         # 指标数据结构+聚合
├── tracing/
│   ├── __init__.py
│   └── tracer.py          # 上下文管理（trace_id生成）
└── analysis/
    ├── __init__.py
    └── root_cause.py      # 根因定位逻辑

# config.py - 全局配置，多云环境模拟通过不同端口区分服务
CHAOS_CONFIG = {
    "service_a": {"delay_prob": 0.2, "delay_max_ms": 500, "error_prob": 0.1},
    "service_b": {"delay_prob": 0.3, "delay_max_ms": 800, "error_prob": 0.05},
    "service_c": {"delay_prob": 0.1, "delay_max_ms": 300, "error_prob": 0.02},
}

SERVICE_PORTS = {
    "a": 8001,
    "b": 8002,
    "c": 8003,
}

LOAD_CONFIG = {
    "qps": 50,               # 每秒请求数
    "duration_sec": 30,       # 压测持续时间
}

TRACING_HEADER = "X-Trace-Id"

# tracing/tracer.py - 简单的trace_id生成与传递
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from config import TRACING_HEADER

class TracingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        trace_id = request.headers.get(TRACING_HEADER) or str(uuid.uuid4())
        # 将trace_id注入请求状态，便于后续服务传递
        request.state.trace_id = trace_id
        response = await call_next(request)
        response.headers[TRACING_HEADER] = trace_id
        return response

def get_trace_id(request) -> str:
    return request.state.trace_id

# chaos/injector.py - 根据配置随机延迟或返回错误
import asyncio
import random
from config import CHAOS_CONFIG

def maybe_inject(service_name: str):
    """返回 (should_error, delay_seconds)"""
    cfg = CHAOS_CONFIG.get(service_name, {})
    error_prob = cfg.get("error_prob", 0)
    delay_prob = cfg.get("delay_prob", 0)
    delay_max = cfg.get("delay_max_ms", 100) / 1000.0

    should_error = random.random() < error_prob
    # 延迟和错误可同时触发，简化模型
    delay = 0
    if random.random() < delay_prob:
        delay = random.uniform(0, delay_max)
    return should_error, delay

async def apply_chaos(service_name: str):
    """应用混沌，返回是否应中断调用（返回503）并记录延迟"""
    should_error, delay = maybe_inject(service_name)
    if delay > 0:
        await asyncio.sleep(delay)
    if should_error:
        raise RuntimeError(f"Chaos injection: simulated error in {service_name}")
    return delay

# services/service_a.py - 入口服务，调用service_b
import json
import time
import httpx
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from config import SERVICE_PORTS, TRACING_HEADER
from tracing.tracer import TracingMiddleware, get_trace_id
from chaos.injector import apply_chaos
from monitoring.metrics import MetricsCollector

app = FastAPI()
app.add_middleware(TracingMiddleware)

metrics = MetricsCollector("service_a")

@app.get("/")
async def root(request: Request):
    trace_id = get_trace_id(request)
    start = time.time()
    try:
        # 先应用自身混沌
        await apply_chaos("service_a")
        # 调用下游 service_b
        async with httpx.AsyncClient() as client:
            # 透传trace_id
            headers = {TRACING_HEADER: trace_id}
            resp = await client.get(f"http://localhost:{SERVICE_PORTS['b']}/", headers=headers, timeout=5)
            resp.raise_for_status()
            result = resp.json()
        duration = time.time() - start
        metrics.record(duration, True)
        return {"service": "a", "trace_id": trace_id, "duration": round(duration, 3), "downstream": result}
    except Exception as e:
        duration = time.time() - start
        metrics.record(duration, False)
        return JSONResponse(status_code=503, content={"service": "a", "trace_id": trace_id, "error": str(e)})

# 暴露指标接口
@app.get("/metrics")
async def get_metrics():
    return metrics.summary()

# monitoring/metrics.py - 线程安全指标聚合
import time
from collections import deque
from threading import Lock

class MetricsCollector:
    def __init__(self, service_name, window_sec=10):
        self.service = service_name
        self.window = window_sec
        self.lock = Lock()
        # 存储 (timestamp, latency, success)
        self.records = deque()

    def record(self, latency: float, success: bool):
        with self.lock:
            self.records.append((time.time(), latency, success))
            # 清理旧数据
            cutoff = time.time() - self.window
            while self.records and self.records[0][0] < cutoff:
                self.records.popleft()

    def summary(self):
        with self.lock:
            total = len(self.records)
            if total == 0:
                return {"service": self.service, "error_rate": 0, "avg_latency": 0, "p99": 0}
            errors = sum(1 for _, _, s in self.records if not s)
            latencies = sorted([l for l, _,_ in self.records])  # 简化，准确需排序
            avg_latency = sum(l for l, _, _ in self.records) / total
            p99_index = int(total * 0.99) - 1
            p99 = latencies[p99_index] if p99_index >= 0 else 0
            return {
                "service": self.service,
                "error_rate": errors / total,
                "avg_latency": round(avg_latency, 3),
                "p99_latency": round(p99, 3),
                "total_requests": total,
            }

# loadtesting/load_gen.py - 异步压测，按QPS发送请求
import asyncio
import httpx
import time
from config import SERVICE_PORTS, LOAD_CONFIG, TRACING_HEADER
from monitoring.metrics import MetricsCollector

class LoadGenerator:
    def __init__(self, target_url: str, qps: int, duration: int):
        self.target = target_url
        self.qps = qps
        self.duration = duration
        self.collector = MetricsCollector("loadgen")  # 全局压测指标

    async def run(self):
        interval = 1.0 / self.qps
        start_time = time.time()
        async with httpx.AsyncClient() as client:
            while time.time() - start_time < self.duration:
                # 每间隔发一个请求，不等待结果可提高吞吐
                asyncio.create_task(self._send_one(client))
                await asyncio.sleep(interval)

        # 等待剩余任务完成（简化：再等待2秒）
        await asyncio.sleep(2)
        return self.collector.summary()

    async def _send_one(self, client):
        start = time.time()
        try:
            resp = await client.get(self.target, timeout=5)
            resp.raise_for_status()
            self.collector.record(time.time() - start, True)
        except Exception:
            self.collector.record(time.time() - start, False)

# analysis/root_cause.py - 基于各服务错误率判断根因
from monitoring.metrics import MetricsCollector

def analyze_root_cause(
    metrics_a: MetricsCollector,
    metrics_b: MetricsCollector,
    metrics_c: MetricsCollector,
    threshold: float = 0.1
) -> str:
    """
    简单根因规则：

    1. 如果所有服务错误率都低于阈值 -> 无故障
    2. 如果C错误率高，B和A正常 -> 根因在C
    3. 如果B错误率高，C正常且A正常 -> 根因在B（因为B即使自己正常也可能因为下游C？此处简化）
    4. 如果A错误率高，B和C正常 -> 根因在A
    5. 如果多个服务异常，选择第一个（A->B->C）最上游的异常服务
    """
    sa = metrics_a.summary()
    sb = metrics_b.summary()
    sc = metrics_c.summary()
    err_a = sa["error_rate"]
    err_b = sb["error_rate"]
    err_c = sc["error_rate"]

    # 上游优先：检查A->B->C
    if err_a > threshold:
        # 检查上游A是否真的异常（可能因下游B导致，但B也异常则根因在A自身？实际复杂）
        # 这里简化：如果A错误率高于阈值，就算B也高，仍认为A是根因（因为A是入口）
        return "service_a"
    elif err_b > threshold:
        return "service_b"
    elif err_c > threshold:
        return "service_c"
    else:
        return "no_fault"

# run.py - 启动三个服务，运行混沌压测，输出根因分析
import asyncio
import uvicorn
import multiprocessing
import sys
from config import SERVICE_PORTS, LOAD_CONFIG
from services.service_a import app as app_a
from services.service_b import app as app_b
from services.service_c import app as app_c
from loadtesting.load_gen import LoadGenerator
from monitoring.metrics import MetricsCollector
from analysis.root_cause import analyze_root_cause

# 全局指标收集器（在每个服务中已实例化，此处为了方便分析复用）
metrics_map = {}

def run_service(app, port):
    uvicorn.run(app, host="0.0.0.0", port=port, log_level="info")

def start_services():
    procs = []
    for app, port, name in [(app_a, SERVICE_PORTS["a"], "a"),
                             (app_b, SERVICE_PORTS["b"], "b"),
                             (app_c, SERVICE_PORTS["c"], "c")]:
        p = multiprocessing.Process(target=run_service, args=(app, port))
        p.start()
        procs.append((name, p, port))
    return procs

def get_metrics_from_service(port):
    # 通过HTTP获取每个服务的指标摘要（实际应使用共享内存，这里简易调用）
    import httpx
    try:
        resp = httpx.get(f"http://localhost:{port}/metrics", timeout=2)
        return resp.json()
    except Exception:
        return None

async def run_chaos_load():
    # 启动服务
    service_procs = start_services()
    await asyncio.sleep(2)  # 等待服务就绪

    # 创建压测生成器，以service_a为入口
    target = f"http://localhost:{SERVICE_PORTS['a']}/"
    loader = LoadGenerator(target, LOAD_CONFIG["qps"], LOAD_CONFIG["duration_sec"])
    print(f"[LOAD] Start load test: QPS={LOAD_CONFIG['qps']}, duration={LOAD_CONFIG['duration_sec']}s")
    summary = await loader.run()
    print(f"[LOAD] Final global metrics: {summary}")

    # 收集各服务指标（从/metrics端点）
    metrics_a = get_metrics_from_service(SERVICE_PORTS["a"])
    metrics_b = get_metrics_from_service(SERVICE_PORTS["b"])
    metrics_c = get_metrics_from_service(SERVICE_PORTS["c"])
    print(f"[METRICS A] {metrics_a}")
    print(f"[METRICS B] {metrics_b}")
    print(f"[METRICS C] {metrics_c}")

    # 模拟分析（实际要传入Collector对象，但这里复用字典）
    # 我们用从HTTP获取的json构造collector的summary类似对象
    # 简化：直接调用analyze_root_cause需要Collector，改为传递字典
    # 重新定义分析函数接受字典
    def analyze_simple(ma, mb, mc, thr=0.1):
        if ma and ma["error_rate"] > thr:
            return "service_a"
        if mb and mb["error_rate"] > thr:
            return "service_b"
        if mc and mc["error_rate"] > thr:
            return "service_c"
        return "no_fault"

    root_cause = analyze_simple(metrics_a, metrics_b, metrics_c)
    print(f"[RESULT] Probable root cause: {root_cause}")

    # 停止服务（简化：发送终止信号）
    for name, proc, port in service_procs:
        proc.terminate()
        proc.join()

if __name__ == "__main__":
    asyncio.run(run_chaos_load())

fastapi==0.104.1
uvicorn[standard]==0.24.0
httpx==0.25.1

# 创建虚拟环境（推荐）
python3 -m venv venv
source venv/bin/activate

# 安装依赖
pip install -r requirements.txt

python run.py

[LOAD] Start load test: QPS=50, duration=30s
[LOAD] Final global metrics: {'service': 'loadgen', 'error_rate': 0.12, 'avg_latency': 0.234, 'p99_latency': 0.789, 'total_requests': 1500}
[METRICS A] {'service': 'service_a', 'error_rate': 0.12, ...}
[METRICS B] {'service': 'service_b', 'error_rate': 0.05, ...}
[METRICS C] {'service': 'service_c', 'error_rate': 0.02, ...}
[RESULT] Probable root cause: service_a

# 终端1
uvicorn services.service_a:app --port 8001
# 终端2
uvicorn services.service_b:app --port 8002
# 终端3
uvicorn services.service_c:app --port 8003

curl http://localhost:8001/ -H "X-Trace-Id: test123"

{"service":"a","trace_id":"test123","duration":0.045,"downstream":{"service":"b","trace_id":"test123","duration":0.032,"downstream":{"service":"c","trace_id":"test123","duration":0.01}}}

curl http://localhost:8001/metrics