故障自愈与自动恢复

核心目标：从”人肉运维”到”系统自愈”，让故障在无人介入的情况下自动检测、自动决策、自动恢复，将 MTTR 从分钟级压缩到秒级。

概述

故障自愈（Self-Healing）是指系统在发生故障时，无需人工干预，通过预定义的规则、策略或智能决策，自动完成故障检测 → 诊断 → 恢复的全链路闭环。它是 SRE 稳定性工程的高级阶段，也是降低 On-Call 压力、缩短 MTTR 的核心手段。

flowchart LR
    A[故障发生] --> B[自动检测<br/>监控/健康检查/探针]
    B --> C[自动诊断<br/>规则匹配/AI分析]
    C --> D{决策引擎}
    D -->|已知模式| E[执行自愈动作<br/>重启/扩容/回滚/切换]
    D -->|未知模式| F[升级人工介入<br/>告警+上下文]
    E --> G[验证恢复<br/>健康检查/指标回归]
    G -->|恢复| H[记录复盘]
    G -->|未恢复| F
    F --> H

自愈层次模型

层级	范围	典型手段	MTTR 目标	自动化程度
L0	无自愈	人工排查处理	>30min	0%
L1	基础设施层	K8s Pod 重启、健康检查	<5min	30%
L2	应用层	自动回滚、自动扩容	<2min	60%
L3	链路层	故障转移、限流降级	<1min	80%
L4	智能自愈	AI 决策、根因推断+自愈	<30s	90%
L5	预防性自愈	预测+提前干预	0（不发生）	95%

K8s 原生自愈机制

1. 存活探针（Liveness Probe）

K8s 最基础的自愈机制：探针失败 → 自动重启 Pod。

apiVersion: apps/v1
kind: Deployment
metadata:
  name: api-server
spec:
  replicas: 3
  template:
    spec:
      containers:
      - name: api
        image: api:v2.1
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
          failureThreshold: 3        # 连续失败3次触发重启
          timeoutSeconds: 5
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 10
          periodSeconds: 5
          failureThreshold: 3
        startupProbe:
          httpGet:
            path: /startup
            port: 8080
          failureThreshold: 30         # 慢启动应用保护
          periodSeconds: 10

三种探针对比：

探针类型	作用	失败后果	适用场景
livenessProbe	判断容器是否存活	重启容器	死锁、线程耗尽、内存溢出
readinessProbe	判断容器是否就绪	从 Service Endpoints 移除	启动慢、依赖未就绪
startupProbe	判断容器是否启动完成	重启容器（在 liveness 之前）	慢启动应用（JVM、模型加载）

2. Pod Disruption Budget

防止自愈动作”过度”导致服务不可用。

apiVersion: policy/v1
kind: PodDisruptionBudget
metadata:
  name: api-pdb
spec:
  minAvailable: 2          # 至少保持2个Pod可用
  selector:
    matchLabels:
      app: api-server

3. Controller 自愈能力

Controller	自愈行为	触发条件
Deployment	重建失败的 Pod	Pod 状态异常
ReplicaSet	补充缺失的 Pod	副本数不足
DaemonSet	在新节点上自动部署	新节点加入集群
StatefulSet	有序重建	Pod 异常退出
Job	重试失败的 Task	配置 backoffLimit

应用层自愈策略

1. 自动回滚（Auto Rollback）

# Argo Rollouts: 自动回滚策略
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: api-rollout
spec:
  replicas: 10
  strategy:
    canary:
      steps:
      - setWeight: 10
      - pause: { duration: 5m }
      - analysis:                   # 自动分析，失败则回滚
          templates:
          - templateName: success-rate
      - setWeight: 30
      - pause: { duration: 5m }
      - analysis:
          templates:
          - templateName: success-rate
      - setWeight: 100

"""自动回滚决策引擎"""

from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
import time


class RollbackTrigger(Enum):
    ERROR_RATE = "error_rate"
    LATENCY_P99 = "latency_p99"
    SUCCESS_RATE = "success_rate"
    RESOURCE_USAGE = "resource_usage"
    CUSTOM_METRIC = "custom_metric"


@dataclass
class HealthMetric:
    name: str
    current_value: float
    threshold: float
    window: int = 300  # 5分钟窗口


@dataclass
class RollbackDecision:
    should_rollback: bool
    trigger: Optional[RollbackTrigger] = None
    reason: str = ""
    metrics: list[HealthMetric] = field(default_factory=list)
    timestamp: float = field(default_factory=time.time)


class AutoRollbackEngine:
    """部署期间自动回滚决策引擎"""

    def __init__(self, metrics: list[HealthMetric]):
        self.metrics = {m.name: m for m in metrics}

    def evaluate(self) -> RollbackDecision:
        """评估是否需要回滚"""
        violations = []

        for name, metric in self.metrics.items():
            if self._is_violated(metric):
                violations.append(metric)

        if not violations:
            return RollbackDecision(
                should_rollback=False,
                reason="所有指标在阈值范围内"
            )

        # 多指标同时违规 → 更高置信度
        primary = violations[0]
        trigger = self._map_trigger(primary.name)

        return RollbackDecision(
            should_rollback=True,
            trigger=trigger,
            reason=f"检测到 {len(violations)} 项指标违规: "
                   f"{', '.join(v.name for v in violations)}",
            metrics=violations
        )

    def _is_violated(self, metric: HealthMetric) -> bool:
        """判断指标是否违规（方向感知）"""
        # error_rate: 超过阈值即违规
        if "error" in metric.name or "latency" in metric.name:
            return metric.current_value > metric.threshold
        # success_rate: 低于阈值即违规
        if "success" in metric.name:
            return metric.current_value < metric.threshold
        return metric.current_value > metric.threshold

    def _map_trigger(self, metric_name: str) -> RollbackTrigger:
        if "error" in metric_name:
            return RollbackTrigger.ERROR_RATE
        if "latency" in metric_name:
            return RollbackTrigger.LATENCY_P99
        if "success" in metric_name:
            return RollbackTrigger.SUCCESS_RATE
        return RollbackTrigger.CUSTOM_METRIC


# 使用示例
if __name__ == "__main__":
    metrics = [
        HealthMetric("error_rate", 5.2, threshold=1.0),    # 5.2% > 1%
        HealthMetric("latency_p99", 850, threshold=500),    # 850ms > 500ms
        HealthMetric("success_rate", 94.8, threshold=99.0), # 94.8% < 99%
    ]
    engine = AutoRollbackEngine(metrics)
    decision = engine.evaluate()
    if decision.should_rollback:
        print(f"[ROLLBACK] 触发: {decision.trigger.value}")
        print(f"  原因: {decision.reason}")
        for m in decision.metrics:
            print(f"  {m.name}: {m.current_value} (阈值: {m.threshold})")

2. 自动扩容（HPA + 自定义指标）

# 基于自定义指标的 HPA —— 消息积压自动扩容
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: consumer-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: mq-consumer
  minReplicas: 3
  maxReplicas: 50
  metrics:
  - type: External
    external:
      metric:
        name: rabbitmq_queue_messages
        selector:
          matchLabels:
            queue: task-queue
      target:
        type: AverageValue
        averageValue: "100"        # 每个Pod处理100条消息
  behavior:
    scaleUp:
      stabilizationWindowSeconds: 0   # 扩容立即执行
      policies:
      - type: Percent
        value: 100                    # 最多翻倍
        periodSeconds: 30
    scaleDown:
      stabilizationWindowSeconds: 300  # 缩容保守5分钟
      policies:
      - type: Percent
        value: 25                     # 每次最多缩25%
        periodSeconds: 60

3. 限流降级自动触发

"""自适应限流 + 降级管理器"""

from dataclasses import dataclass
from enum import Enum
import time


class DegradationLevel(Enum):
    NORMAL = 0       # 正常服务
    LIGHT = 1        # 关闭非核心功能
    MEDIUM = 2       # 只保留核心链路
    CRITICAL = 3     # 只读/排队模式


@dataclass
class SystemState:
    cpu_usage: float         # 0-100
    memory_usage: float       # 0-100
    error_rate: float         # 0-100
    p99_latency: float       # ms
    qps: float


class AdaptiveDegradationManager:
    """根据系统状态自动调整降级等级"""

    THRESHOLDS = {
        DegradationLevel.NORMAL: {
            "cpu": 60, "memory": 70, "error": 0.5, "latency": 200
        },
        DegradationLevel.LIGHT: {
            "cpu": 75, "memory": 80, "error": 1.0, "latency": 500
        },
        DegradationLevel.MEDIUM: {
            "cpu": 85, "memory": 90, "error": 3.0, "latency": 1000
        },
    }

    def __init__(self):
        self.current_level = DegradationLevel.NORMAL
        self.degradation_start_time = 0

    def evaluate(self, state: SystemState) -> DegradationLevel:
        """评估系统状态，返回目标降级等级"""
        # 检查是否需要升级降级
        target = DegradationLevel.NORMAL

        for level in [DegradationLevel.MEDIUM, DegradationLevel.LIGHT,
                       DegradationLevel.NORMAL]:
            t = self.THRESHOLDS[level]
            if (state.cpu_usage > t["cpu"] or
                state.memory_usage > t["memory"] or
                state.error_rate > t["error"] or
                state.p99_latency > t["latency"]):
                target = level
                break
        else:
            target = DegradationLevel.CRITICAL

        # 降级恢复需要持续稳定才升级（防止抖动）
        if target.value < self.current_level.value:
            if self.degradation_start_time == 0:
                self.degradation_start_time = time.time()
            elif time.time() - self.degradation_start_time < 60:
                # 不足60秒，保持当前等级
                return self.current_level
        else:
            self.degradation_start_time = 0

        self.current_level = target
        return target

    def get_actions(self, level: DegradationLevel) -> list[str]:
        """返回对应等级的降级动作"""
        actions = {
            DegradationLevel.NORMAL: [],
            DegradationLevel.LIGHT: [
                "关闭推荐/搜索等非核心功能",
                "降低日志级别到 WARN",
                "关闭定时任务",
            ],
            DegradationLevel.MEDIUM: [
                "关闭所有非核心API",
                "返回缓存数据替代实时计算",
                "启用批量处理替代实时处理",
            ],
            DegradationLevel.CRITICAL: [
                "进入只读模式",
                "请求排队（令牌桶限流到最小）",
                "返回静态降级页面",
            ],
        }
        return actions.get(level, [])

故障自愈 Runbook 自动化

Runbook 执行引擎

"""Runbook 自动执行引擎 —— 将人工SOP转化为自动执行流程"""

from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Optional
import subprocess
import json
import time


class StepStatus(Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    SKIPPED = "skipped"


class StepType(Enum):
    CHECK = "check"          # 检查类步骤
    ACTION = "action"        # 执行类步骤
    VERIFY = "verify"        # 验证类步骤
    NOTIFY = "notify"        # 通知类步骤
    ESCALATE = "escalate"    # 升级类步骤


@dataclass
class RunbookStep:
    name: str
    step_type: StepType
    command: str               # Shell 命令
    timeout: int = 60          # 超时秒数
    on_failure: str = "stop"   # stop / continue / escalate
    retry_count: int = 0
    max_retries: int = 2
    status: StepStatus = StepStatus.PENDING
    output: str = ""
    error: str = ""


@dataclass
class RunbookResult:
    runbook_name: str
    total_steps: int
    executed_steps: int
    success_steps: int
    failed_steps: int
    duration_seconds: float
    final_status: str         # "healed" / "failed" / "escalated"
    steps_detail: list[dict] = field(default_factory=list)


class RunbookEngine:
    """Runbook 自动执行引擎"""

    def __init__(self, name: str, steps: list[RunbookStep]):
        self.name = name
        self.steps = steps
        self.start_time = 0

    def execute(self) -> RunbookResult:
        """按顺序执行所有步骤"""
        self.start_time = time.time()
        executed = 0
        success = 0
        failed = 0
        escalated = False
        details = []

        for step in self.steps:
            if step.status == StepStatus.SKIPPED:
                continue

            step.status = StepStatus.RUNNING
            result = self._execute_step(step)
            executed += 1
            details.append({
                "step": step.name,
                "type": step.step_type.value,
                "status": result.status.value,
                "output": result.output[:200],
                "error": result.error[:200] if result.error else "",
            })

            if result.status == StepStatus.SUCCESS:
                success += 1
            else:
                failed += 1
                if step.on_failure == "stop":
                    break
                elif step.on_failure == "escalate":
                    escalated = True
                    # 发送升级通知
                    self._send_escalation(step)
                    break

        duration = time.time() - self.start_time
        final = "escalated" if escalated else ("healed" if failed == 0 else "failed")

        return RunbookResult(
            runbook_name=self.name,
            total_steps=len(self.steps),
            executed_steps=executed,
            success_steps=success,
            failed_steps=failed,
            duration_seconds=round(duration, 2),
            final_status=final,
            steps_detail=details,
        )

    def _execute_step(self, step: RunbookStep) -> RunbookStep:
        """执行单个步骤，支持重试"""
        for attempt in range(step.max_retries + 1):
            try:
                proc = subprocess.run(
                    step.command,
                    shell=True,
                    capture_output=True,
                    text=True,
                    timeout=step.timeout,
                )
                if proc.returncode == 0:
                    step.status = StepStatus.SUCCESS
                    step.output = proc.stdout
                    return step
                else:
                    step.error = proc.stderr
                    step.retry_count = attempt + 1
            except subprocess.TimeoutExpired:
                step.error = f"超时 ({step.timeout}s)"
                step.retry_count = attempt + 1
            except Exception as e:
                step.error = str(e)
                step.retry_count = attempt + 1

        step.status = StepStatus.FAILED
        return step

    def _send_escalation(self, step: RunbookStep):
        """发送升级通知"""
        msg = json.dumps({
            "event": "runbook_escalation",
            "runbook": self.name,
            "failed_step": step.name,
            "error": step.error,
            "timestamp": time.time(),
        })
        # 发送到 Alertmanager / 钉钉 / 飞书
        print(f"[ESCALATE] {msg}")


# ====== 预定义 Runbook: Pod CrashLoopBackOff 自愈 ======
crashloop_runbook = RunbookEngine(
    name="CrashLoopBackOff-SelfHeal",
    steps=[
        RunbookStep(
            name="检测CrashLoopBackOff",
            step_type=StepType.CHECK,
            command="kubectl get pods --field-selector=status.phase=Running "
                    "-o jsonpath='{range .items[?(@.status.containerStatuses[0].state."
                    "waiting.reason==\"CrashLoopBackOff\")]}{.metadata.name}{\"\\n\"}{end}'",
            timeout=30,
        ),
        RunbookStep(
            name="获取崩溃日志",
            step_type=StepType.CHECK,
            command="kubectl logs ${FAILED_POD} --previous --tail=100",
            timeout=30,
            on_failure="continue",
        ),
        RunbookStep(
            name="检查最近变更",
            step_type=StepType.CHECK,
            command="kubectl rollout history deployment/${DEPLOYMENT}",
            timeout=30,
            on_failure="continue",
        ),
        RunbookStep(
            name="回滚到上一版本",
            step_type=StepType.ACTION,
            command="kubectl rollout undo deployment/${DEPLOYMENT}",
            timeout=60,
            on_failure="escalate",
            max_retries=1,
        ),
        RunbookStep(
            name="验证恢复",
            step_type=StepType.VERIFY,
            command="kubectl wait --for=condition=available deployment/${DEPLOYMENT} "
                    "--timeout=120s",
            timeout=130,
            on_failure="escalate",
        ),
        RunbookStep(
            name="发送恢复通知",
            step_type=StepType.NOTIFY,
            command="echo 'Pod CrashLoopBackOff 已自动回滚恢复'",
            timeout=10,
        ),
    ],
)

安全护栏机制

自愈系统本身也可能出错，必须有安全护栏防止”自愈风暴”。

# 安全护栏配置
safety_guardrails:
  # 1. 速率限制：同一目标在时间窗口内最多执行N次自愈
  rate_limits:
    pod_restart:
      max_actions: 3
      window_seconds: 300          # 5分钟内最多重启3次
    deployment_rollback:
      max_actions: 1
      window_seconds: 3600         # 1小时内最多回滚1次

  # 2. 爆炸半径控制：同时自愈的目标数量限制
  blast_radius:
    max_concurrent_heals: 5        # 最多同时执行5个自愈动作
    max_affected_pods: 20          # 单次自愈最多影响20个Pod

  # 3. 回退条件：检测到以下情况立即停止自愈
  abort_conditions:
    - "集群整体错误率 > 10%"          # 大面积故障，停止自动操作
    - "自愈动作失败率 > 30%"          # 自愈系统本身不稳定
    - "在变更冻结窗口内"              # 冻结期禁止自愈变更
    - "依赖服务异常"                  # 上游问题导致的下游故障

  # 4. 白名单/黑名单
  target_filters:
    whitelist_namespaces:
      - production
      - staging
    blacklist_deployments:
      - payment-gateway             # 支付服务不自动回滚
      - auth-service                # 认证服务需人工确认

"""安全护栏执行器"""

from dataclasses import dataclass, field
from collections import defaultdict
import time


@dataclass
class GuardrailConfig:
    max_actions: int
    window_seconds: int


class SafetyGuardrail:
    """自愈安全护栏 —— 防止自愈风暴"""

    def __init__(self):
        self.action_history: dict[str, list[float]] = defaultdict(list)
        self.configs: dict[str, GuardrailConfig] = {
            "pod_restart": GuardrailConfig(3, 300),
            "deployment_rollback": GuardrailConfig(1, 3600),
            "hpa_scale": GuardrailConfig(10, 60),
        }
        self.aborted = False

    def check_rate_limit(self, action_type: str, target: str) -> bool:
        """检查速率限制"""
        key = f"{action_type}:{target}"
        config = self.configs.get(action_type)
        if not config:
            return True  # 无限制

        now = time.time()
        # 清理过期记录
        self.action_history[key] = [
            t for t in self.action_history[key]
            if now - t < config.window_seconds
        ]

        if len(self.action_history[key]) >= config.max_actions:
            print(f"[GUARDRAIL] 速率限制: {key} 在 {config.window_seconds}s 内"
                  f"已执行 {len(self.action_history[key])} 次，阻止")
            return False

        self.action_history[key].append(now)
        return True

    def check_abort_conditions(self, cluster_metrics: dict) -> tuple[bool, str]:
        """检查中止条件"""
        conditions = [
            (cluster_metrics.get("error_rate", 0) > 10,
             "集群错误率超过10%"),
            (cluster_metrics.get("heal_failure_rate", 0) > 30,
             "自愈失败率超过30%"),
            (cluster_metrics.get("in_freeze_window", False),
             "处于变更冻结窗口"),
        ]

        for triggered, reason in conditions:
            if triggered:
                self.aborted = True
                return False, reason

        return True, ""

    def check_blast_radius(self, current_concurrent: int,
                           affected_pods: int) -> bool:
        """检查爆炸半径"""
        if current_concurrent >= 5:
            print("[GUARDRAIL] 并发自愈数已达上限 5")
            return False
        if affected_pods > 20:
            print(f"[GUARDRAIL] 影响Pod数 {affected_pods} 超过上限 20")
            return False
        return True

自愈成熟度评估

维度	L1 起步	L2 基础	L3 成熟	L4 高级	L5 领先
检测能力	人工发现	基础告警	多维监控	异常检测	预测检测
自愈覆盖	Pod重启	回滚+扩容	链路切换	AI决策	预防干预
MTTR	>15min	<5min	<2min	<30s	0
人工介入率	90%	60%	30%	10%	<5%
安全护栏	无	基础限速	速率+爆炸半径	全套护栏	自适应护栏
复盘闭环	人工记录	自动记录	自动+趋势	AI归因	预防改进

常见坑点

坑点	现象	解决方案
自愈风暴	反复重启→反复崩溃→资源耗尽	速率限制 + 爆炸半径控制
隐式依赖	回滚成功但依赖未回滚	变更链路关联分析
误判自愈	网络抖动被判为故障→触发自愈	多窗口确认 + 去抖动
自愈掩盖根因	反复自愈使根因被忽略	自愈次数阈值→强制升级人工
权限过大	自愈脚本有删改权限	最小权限原则 + 操作审计

关联知识

• SRE 稳定性工程总览 — 自愈是稳定性工程的高级阶段
• MTTR 与 MTBF 体系详解 — 自愈直接缩短 MTTI + MTTR_restore
• 根因定位方法论 (RCA) — 自愈前需要诊断，自愈后需要 RCA
• 弹性伸缩策略 — HPA/VPA/KEDA 是自愈的重要手段
• 变更管理全流程 — 自动回滚是变更管理的关键环节
• 高可用架构设计总览 — 故障转移是链路层自愈
• SRE 工具链总览 — Runbook 自动化是自愈的执行层
• 混沌工程与故障注入 — 验证自愈系统有效性
• SRE Agent 运维落地实践 — AI Agent 是智能自愈的核心
• K8s 故障排查方法论 — CrashLoopBackOff 等常见故障处理

参考资源

• 《SRE: Google 运维解密》第6章 — 监控告警与自动化
• K8s 官方文档 — Liveness/Readiness/Startup Probes
• Argo Rollouts 官方文档 — Analysis & Auto-Rollback
• ChaosMesh — 通过混沌工程验证自愈能力
• 《Observability Engineering》— 自愈需要可观测性支撑

学习时间

约 6-8 小时（含实践搭建自愈 Runbook）

状态

• 理解故障自愈的层次模型
• 掌握 K8s 原生自愈机制（探针/PDB/Controller）
• 能编写自动回滚/扩容/降级策略
• 能构建 Runbook 自动执行引擎
• 理解安全护栏机制并实现
• 实际搭建一个端到端自愈流水线
• 验证自愈系统在混沌实验中的表现