1. 故障自动化脚本框架搭建

为什么需要统一框架

故障自动化的核心不是“写一个脚本”，而是让注入、检测、恢复、验证四个阶段可插拔、可复用、可观测。若每个故障场景都从头写脚本，会出现三类问题：阶段边界模糊导致恢复逻辑混入检测代码；参数散落在各处无法批量回归；执行结果没有统一结构，后续无法自动判定成功或失败。

本案例搭建的框架把一次故障演练抽象为一条固定执行链：注入 → 检测 → 恢复 → 验证。每个阶段是一个独立接口，框架只负责按顺序调度、传递上下文、记录结果，具体实现由各场景的插件提供。

框架的四个阶段接口

四个阶段共享同一个上下文对象 ctx，用于在阶段之间传递故障标识、目标对象、参数和观测数据。接口约定如下：

• inject(ctx)：执行故障注入，返回注入是否成功及注入句柄（如进程号、规则ID、文件路径），供恢复阶段使用。
• detect(ctx)：检测故障是否已生效，返回布尔值与证据（日志行、端口状态、指标值）。
• recover(ctx)：依据注入句柄执行恢复动作，返回恢复是否成功。
• verify(ctx)：验证系统是否回到健康状态，返回最终判定与残留问题。

框架对每个阶段统一做三件事：记录开始/结束时间、捕获异常、把返回值写入 ctx.results。任一阶段抛出异常时，框架标记该阶段失败并继续执行恢复阶段，避免故障残留。

框架骨架代码

下面给出可直接运行的最小框架。它不包含任何具体故障逻辑，只定义接口、调度流程和结果结构，后续案例只需实现四个函数即可接入。

import time
import traceback
from dataclasses import dataclass, field
from typing import Any, Callable, Optional

@dataclass
class FaultContext:
    """贯穿四个阶段的上下文对象"""
    fault_id: str
    target: str
    params: dict = field(default_factory=dict)
    handle: Any = None          # inject 返回的注入句柄
    evidence: dict = field(default_factory=dict)  # detect 收集的证据
    results: dict = field(default_factory=dict)   # 各阶段返回值
    timings: dict = field(default_factory=dict)   # 各阶段耗时
    errors: dict = field(default_factory=dict)    # 各阶段异常信息

class FaultPlugin:
    """故障插件基类：子类实现四个阶段"""

    def inject(self, ctx: FaultContext) -> Any:
        raise NotImplementedError

    def detect(self, ctx: FaultContext) -> bool:
        raise NotImplementedError

    def recover(self, ctx: FaultContext) -> bool:
        raise NotImplementedError

    def verify(self, ctx: FaultContext) -> bool:
        raise NotImplementedError

class Runner:
    """调度器：按 inject → detect → recover → verify 顺序执行"""

    STAGES = ["inject", "detect", "recover", "verify"]

    def __init__(self, plugin: FaultPlugin):
        self.plugin = plugin

    def _run_stage(self, name: str, ctx: FaultContext) -> Optional[Any]:
        fn: Callable = getattr(self.plugin, name)
        start = time.time()
        try:
            result = fn(ctx)
            ctx.results[name] = result
            return result
        except Exception as exc:
            ctx.errors[name] = f"{type(exc).__name__}: {exc}"
            ctx.results[name] = None
            return None
        finally:
            ctx.timings[name] = round(time.time() - start, 4)

    def run(self, ctx: FaultContext) -> dict:
        # 阶段1：注入
        ctx.handle = self._run_stage("inject", ctx)

        # 阶段2：检测（即使注入失败也尝试检测，便于定位）
        detected = self._run_stage("detect", ctx)

        # 阶段3：恢复（无论检测结果如何都执行，避免故障残留）
        recovered = self._run_stage("recover", ctx)

        # 阶段4：验证
        verified = self._run_stage("verify", ctx)

        # 汇总判定
        passed = (
            ctx.errors.get("inject") is None
            and detected is True
            and recovered is True
            and verified is True
        )
        ctx.results["final"] = "PASS" if passed else "FAIL"
        return {
            "fault_id": ctx.fault_id,
            "final": ctx.results["final"],
            "results": ctx.results,
            "timings": ctx.timings,
            "errors": ctx.errors,
        }

用一个示例插件验证框架

为确认框架可运行，实现一个“删除临时文件”的示例插件：注入阶段删除目标文件，检测阶段确认文件不存在，恢复阶段从备份还原，验证阶段确认文件已恢复。

import os
import shutil

class TempFileDeletePlugin(FaultPlugin):
    """示例：删除临时文件并恢复"""

    def inject(self, ctx):
        path = ctx.target
        backup = path + ".bak"
        shutil.copy2(path, backup)
        os.remove(path)
        ctx.params["backup"] = backup
        return {"removed": path, "backup": backup}

    def detect(self, ctx):
        exists = os.path.exists(ctx.target)
        ctx.evidence["file_exists_after_inject"] = exists
        return not exists

    def recover(self, ctx):
        backup = ctx.params["backup"]
        shutil.copy2(backup, ctx.target)
        os.remove(backup)
        return os.path.exists(ctx.target)

    def verify(self, ctx):
        ok = os.path.exists(ctx.target)
        ctx.evidence["file_exists_after_recover"] = ok
        return ok

if __name__ == "__main__":
    # 准备测试文件
    test_file = "/tmp/fault_demo.txt"
    with open(test_file, "w") as f:
        f.write("healthy content")

    ctx = FaultContext(fault_id="demo-001", target=test_file)
    report = Runner(TempFileDeletePlugin()).run(ctx)

    print("最终判定:", report["final"])
    print("各阶段结果:", report["results"])
    print("各阶段耗时:", report["timings"])
    print("异常信息:", report["errors"])

预期输出中 final 为 PASS，results 包含 inject 的句柄字典、detect 的 True、recover 的 True、verify 的 True，errors 为空字典。若把 recover 中的还原语句注释掉，verify 会返回 False，最终判定变为 FAIL，说明框架能正确捕获恢复失败。

接入新故障场景的约定