MTTR 从 45 分钟降到 5 分钟：我们是怎么用 AI 实现故障自动修复闭环的

# runbook-oom-restart.yaml
apiVersion: selfhealing/v1
kind: Runbook
metadata:
  name: oom-auto-restart
  labels:
    alertType: OOMKilled
    severity: P2
    riskLevel: low
spec:
  trigger:
    alertName: OOMKilled
    matchLabels:
      app: "{{ .Labels.app }}"
  steps:
    - name: 检查最近重启次数
      action: check-restart-count
      params:
        namespace: "{{ .Labels.namespace }}"
        pod: "{{ .Labels.pod }}"
        windowMinutes: 10
        maxRestarts: 3

    - name: 记录当前内存配置
      action: get-resource-limits
      params:
        namespace: "{{ .Labels.namespace }}"
        deployment: "{{ .Labels.deployment }}"

    - name: 提升内存限制
      action: patch-deployment
      params:
        memoryLimit: "2Gi"  # 原 1Gi → 2Gi
        memoryRequest: "1Gi"
      rollback:
        memoryLimit: "1Gi"
        memoryRequest: "512Mi"

    - name: 验证服务恢复
      action: health-check
      params:
        endpoint: "http://{{ .Labels.app }}:8080/actuator/health"
        timeoutSeconds: 60
        expectedStatus: 200

#!/usr/bin/env python3
"""
Runbook 执行引擎
支持步骤编排、条件分支、自动回滚
"""
import subprocess
import json
import logging
from datetime import datetime

logger = logging.getLogger("runbook-engine")

class RunbookExecutor:
    def __init__(self, runbook: dict, context: dict):
        self.runbook = runbook
        self.context = context
        self.executed_steps = []  # 记录已执行步骤，用于回滚
        self.start_time = datetime.now()

    def execute(self) -> dict:
        """执行 Runbook 全部步骤"""
        steps = self.runbook["spec"]["steps"]
        for step in steps:
            try:
                result = self._execute_step(step)
                self.executed_steps.append({
                    "step": step["name"],
                    "status": "success",
                    "result": result,
                    "rollback": step.get("rollback")
                })
                logger.info(f"步骤 [{step['name']}] 执行成功")
            except Exception as e:
                logger.error(f"步骤 [{step['name']}] 执行失败: {e}")
                self._rollback()
                return {"status": "failed", "error": str(e)}

        return {
            "status": "success",
            "steps_completed": len(self.executed_steps),
            "duration_sec": (datetime.now() - self.start_time).seconds
        }

    def _execute_step(self, step: dict) -> dict:
        """执行单个步骤"""
        action = step["action"]
        params = step.get("params", {})

        if action == "check-restart-count":
            return self._check_restart_count(params)
        elif action == "patch-deployment":
            return self._patch_deployment(params)
        elif action == "health-check":
            return self._health_check(params)
        else:
            raise ValueError(f"未知 action: {action}")

    def _check_restart_count(self, params: dict) -> dict:
        """检查 Pod 最近重启次数"""
        ns = params["namespace"]
        pod = params["pod"]
        window = params["windowMinutes"]
        max_count = params["maxRestarts"]

        cmd = f"kubectl get pod {pod} -n {ns} -o json"
        result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
        pod_info = json.loads(result.stdout)
        restarts = pod_info["status"]["containerStatuses"][0]["restartCount"]

        if restarts >= max_count:
            raise Exception(
                f"重启次数 {restarts} 超过阈值 {max_count}，停止自愈，转人工"
            )
        return {"restartCount": restarts, "allowAutoFix": True}

    def _patch_deployment(self, params: dict) -> dict:
        """修改 Deployment 资源配置"""
        # 实际生产中会调用 K8s API
        logger.info(f"修改 Deployment: {params}")
        return {"patched": True}

    def _health_check(self, params: dict) -> dict:
        """服务健康检查"""
        import requests
        try:
            resp = requests.get(
                params["endpoint"],
                timeout=params.get("timeoutSeconds", 30)
            )
            return {"status": resp.status_code, "healthy": resp.status_code == 200}
        except requests.RequestException as e:
            raise Exception(f"健康检查失败: {e}")

    def _rollback(self):
        """回滚已执行的步骤"""
        logger.warning("开始回滚...")
        for step_info in reversed(self.executed_steps):
            rollback = step_info.get("rollback")
            if rollback:
                logger.info(f"回滚步骤 [{step_info['step']}]: {rollback}")
                # 执行回滚逻辑
        logger.info("回滚完成")

"""回滚保护装饰器"""
from functools import wraps

def with_rollback(rollback_func):
    """为自愈步骤添加回滚保护"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                result = func(*args, **kwargs)
                # 记录回滚上下文
                rollback_ctx = {
                    "step": func.__name__,
                    "timestamp": datetime.now().isoformat(),
                    "params": kwargs
                }
                # 保存回滚信息到 Redis，TTL 1 小时
                redis.setex(
                    f"rollback:{func.__name__}",
                    3600,
                    json.dumps(rollback_ctx)
                )
                return result
            except Exception as e:
                logger.error(f"步骤 {func.__name__} 失败，触发回滚: {e}")
                rollback_func(**kwargs)
                raise
        return wrapper
    return decorator

# 执行前校验模板参数
REQUIRED_PARAMS = ["app", "namespace", "deployment"]

def validate_context(context: dict) -> bool:
    missing = [p for p in REQUIRED_PARAMS if not context.get(p)]
    if missing:
        raise ValueError(f"缺少必填参数: {missing}")
    return True

def should_auto_heal(alert: dict) -> bool:
    """判断是否应触发自愈"""
    key = f"silence:{alert['name']}:{alert['labels']['app']}"
    if redis.exists(key):
        logger.info(f"告警 {alert['name']} 在静默期，跳过自愈")
        return False
    # 设置 5 分钟静默
    redis.setex(key, 300, "healing_in_progress")
    return True

# Crontab 配置：自愈 Runbook 定时健康检查
0 * * * * /opt/scripts/self_healing_health.sh  # 每小时检查自愈服务状态
0 3 * * * /opt/scripts/runbook_backup.sh       # 每天凌晨备份 Runbook
*/5 * * * * /opt/scripts/check_healing_storm.sh # 每5分钟检查自愈风暴