AI 辅助的故障复现与回放：从人工描述到自动化场景重建

一、故障复现的效率困境：不可复现的 Bug 是最昂贵的 Bug

运维团队最头疼的问题不是"出了故障"，而是"故障无法复现"。一个间歇性的数据库连接超时，在凌晨 3 点出现了 5 分钟，天亮后一切正常。日志显示连接池耗尽，但无法确定是流量突增、慢查询阻塞还是网络抖动导致的。没有复现条件，就无法定位根因，更无法验证修复效果。

AI 辅助的故障复现方案，核心思路是：从监控数据、日志和链路追踪中提取故障时刻的系统状态，自动生成可复现的测试场景。通过"状态快照 + 流量回放 + 环境模拟"三要素，在隔离环境中重建故障现场。

二、故障复现的架构设计与状态重建机制

故障复现的核心挑战是"状态完整性"——故障时刻的系统状态由多个维度构成：应用状态（内存、连接池、缓存）、基础设施状态（CPU、内存、网络）、外部依赖状态（数据库、第三方 API）。完整复现需要同时重建所有维度的状态。

flowchart TB
    A[故障时刻 T] --> B[多维度状态采集]
    B --> C[指标快照: CPU/内存/网络]
    B --> D[日志快照: 错误日志/慢查询]
    B --> E[链路快照: 请求链路/耗时分布]
    B --> F[配置快照: 部署版本/参数配置]

    C --> G[状态重建引擎]
    D --> G
    E --> G
    F --> G

    G --> H[环境模拟]
    H --> I[流量回放: 重放故障时刻的请求]
    H --> J[负载模拟: 重建 CPU/内存压力]
    H --> K[故障注入: 模拟网络延迟/丢包]

    I --> L[复现验证]
    J --> L
    K --> L

    L --> M{故障复现?}
    M -->|是| N[根因定位]
    M -->|否| O[调整参数，重新尝试]
    O --> G

三、生产级实现：故障复现引擎

# fault_replayer.py — AI 辅助故障复现引擎
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime, timedelta
import json

@dataclass
class SystemSnapshot:
    timestamp: datetime
    cpu_usage: float
    memory_usage: float
    network_in_mbps: float
    network_out_mbps: float
    active_connections: int
    slow_queries: List[Dict]
    error_logs: List[Dict]
    deployment_version: str

@dataclass
class TrafficSample:
    timestamp: datetime
    method: str
    path: str
    headers: Dict[str, str]
    body: Optional[str]
    response_code: int
    latency_ms: float

@dataclass
class ReplayScenario:
    name: str
    description: str
    snapshot: SystemSnapshot
    traffic: List[TrafficSample]
    fault_injections: List[Dict]
    expected_symptoms: List[str]

class FaultReplayer:
    """故障复现引擎：从监控数据生成可复现的测试场景"""

    def generate_scenario(
        self,
        fault_time: datetime,
        duration_minutes: int,
        monitoring_data: Dict,
    ) -> ReplayScenario:
        """从故障时刻的监控数据生成复现场景"""

        # 步骤 1：提取故障时刻的系统快照
        snapshot = self._extract_snapshot(fault_time, monitoring_data)

        # 步骤 2：提取故障时间窗口的流量样本
        traffic = self._extract_traffic(
            fault_time, duration_minutes, monitoring_data
        )

        # 步骤 3：推断可能的故障注入点
        injections = self._infer_fault_injections(snapshot, traffic)

        # 步骤 4：生成场景描述
        description = self._generate_description(snapshot, injections)

        return ReplayScenario(
            name=f"replay-{fault_time.strftime('%Y%m%d-%H%M%S')}",
            description=description,
            snapshot=snapshot,
            traffic=traffic,
            fault_injections=injections,
            expected_symptoms=self._extract_symptoms(snapshot),
        )

    def _extract_snapshot(
        self, fault_time: datetime, data: Dict
    ) -> SystemSnapshot:
        """提取故障时刻的系统状态快照"""
        metrics = data.get("metrics", {})
        logs = data.get("logs", {})

        return SystemSnapshot(
            timestamp=fault_time,
            cpu_usage=metrics.get("cpu_usage", 0),
            memory_usage=metrics.get("memory_usage", 0),
            network_in_mbps=metrics.get("network_in", 0),
            network_out_mbps=metrics.get("network_out", 0),
            active_connections=metrics.get("connections", 0),
            slow_queries=logs.get("slow_queries", []),
            error_logs=logs.get("errors", []),
            deployment_version=data.get("version", "unknown"),
        )

    def _extract_traffic(
        self, fault_time: datetime, duration: int, data: Dict
    ) -> List[TrafficSample]:
        """提取故障时间窗口的流量样本"""
        samples = []
        raw_traffic = data.get("traffic", [])

        end_time = fault_time + timedelta(minutes=duration)
        for req in raw_traffic:
            req_time = datetime.fromisoformat(req["timestamp"])
            if fault_time <= req_time <= end_time:
                samples.append(TrafficSample(
                    timestamp=req_time,
                    method=req.get("method", "GET"),
                    path=req.get("path", "/"),
                    headers=req.get("headers", {}),
                    body=req.get("body"),
                    response_code=req.get("status", 200),
                    latency_ms=req.get("latency", 0),
                ))

        return samples

    def _infer_fault_injections(
        self, snapshot: SystemSnapshot, traffic: List[TrafficSample]
    ) -> List[Dict]:
        """推断可能的故障注入点"""
        injections = []

        # 推断 1：CPU 压力注入
        if snapshot.cpu_usage > 80:
            injections.append({
                "type": "cpu_stress",
                "target": "application",
                "parameters": {
                    "usage_percent": int(snapshot.cpu_usage),
                    "duration_seconds": 300,
                },
                "reason": f"故障时刻 CPU 使用率 {snapshot.cpu_usage:.1f}%",
            })

        # 推断 2：网络延迟注入
        slow_requests = [t for t in traffic if t.latency_ms > 1000]
        if len(slow_requests) > len(traffic) * 0.1:
            avg_latency = sum(t.latency_ms for t in slow_requests) / len(slow_requests)
            injections.append({
                "type": "network_delay",
                "target": "database",
                "parameters": {
                    "delay_ms": int(avg_latency * 0.5),
                    "jitter_ms": 50,
                },
                "reason": f"{len(slow_requests)} 个请求延迟超过 1s，平均 {avg_latency:.0f}ms",
            })

        # 推断 3：连接池耗尽注入
        if snapshot.active_connections > 500:
            injections.append({
                "type": "connection_exhaustion",
                "target": "connection_pool",
                "parameters": {
                    "max_connections": snapshot.active_connections,
                },
                "reason": f"活跃连接数 {snapshot.active_connections}，可能耗尽连接池",
            })

        return injections

    def _generate_description(
        self, snapshot: SystemSnapshot, injections: List[Dict]
    ) -> str:
        """生成场景描述"""
        parts = [
            f"故障时刻: {snapshot.timestamp.isoformat()}",
            f"系统状态: CPU {snapshot.cpu_usage:.1f}%, "
            f"内存 {snapshot.memory_usage:.1f}%, "
            f"连接数 {snapshot.active_connections}",
            f"部署版本: {snapshot.deployment_version}",
        ]

        if injections:
            parts.append("推断的故障注入:")
            for inj in injections:
                parts.append(f"  - {inj['type']}: {inj['reason']}")

        return "\n".join(parts)

    def _extract_symptoms(self, snapshot: SystemSnapshot) -> List[str]:
        """提取预期的故障症状"""
        symptoms = []
        if snapshot.cpu_usage > 80:
            symptoms.append("CPU 使用率超过 80%")
        if snapshot.slow_queries:
            symptoms.append(f"慢查询数量: {len(snapshot.slow_queries)}")
        if snapshot.error_logs:
            symptoms.append(f"错误日志数量: {len(snapshot.error_logs)}")
        return symptoms

四、边界分析与架构权衡

AI 辅助故障复现在生产落地中需要正视以下 Trade-off：

状态快照的完整性。监控系统通常以 15-60 秒的间隔采集指标，故障时刻的精确状态可能被采样间隔"模糊化"。例如，CPU 在 15 秒内可能从 30% 飙升到 100% 再回落，但监控只记录了平均值 65%。建议对关键指标使用 1 秒采集间隔，或使用 eBPF 实现内核级的高频采集。

流量回放的安全性。回放生产流量到测试环境可能包含敏感数据（用户信息、支付数据）。必须在回放前对敏感字段进行脱敏处理。同时，回放流量不应触发真实的副作用（如发送邮件、扣款），需要 Mock 外部依赖。

复现成功率。间歇性故障的复现成功率通常低于 50%，因为故障可能依赖特定的时序条件（如两个请求恰好同时到达）。建议多次回放并引入随机延迟，增加命中故障时序的概率。

适用边界：故障复现最适合可观测性良好的系统（有完整的监控、日志和链路追踪）。对于缺乏可观测性的遗留系统，复现所需的状态数据不足，效果有限。

五、总结

AI 辅助的故障复现，将排障从"人工描述"推进到"自动化场景重建"。核心架构：多维度状态采集 → 故障注入推断 → 流量回放验证。落地建议：第一，关键指标使用 1 秒采集间隔，确保状态快照的精度；第二，流量回放前必须脱敏和 Mock 外部依赖；第三，多次回放并引入随机延迟，提高间歇性故障的复现率。关键原则：故障复现不是"重放过去"，而是"理解过去"——复现场景的价值在于帮助定位根因，而非简单地重现现象。