AIOps从监控到自愈：构建具备闭环能力的智能运维Agent系统架构设计

一、为什么大多数AIOps项目停在"告警"这一步

观察近三年AIOps项目的落地情况，可以发现一个明显的断点：绝大多数项目完成了异常检测和根因定位的功能，但极少有项目真正实现从"发现问题"到"自动修复"的闭环。

断裂的原因有两个。一是安全信任问题。故障自愈意味着机器有权限在生产环境执行变更操作——重启服务、切换流量、扩容节点。运维团队对AI的判断准确率天然存疑，赋予自愈权限的心理门槛极高。二是工程复杂度。检测和定位是单向的数据处理管道，自愈则需要引入决策→执行→验证→回滚的状态机，涉及变更管理、权限控制和执行审计等一系列工程问题。

但闭环的价值恰恰也在于此。能检测故障和能自动修复故障，对业务连续性的影响是指数级的差异。一条告警推送到运维人员的手机，从看到→理解→决策→执行，即使最优秀的运维工程师也需要3-5分钟。而这3-5分钟对金融交易系统来说可能意味着千万级的损失。

二、AIOps自愈Agent的架构设计

flowchart TD
    A[可观测性数据采集层] --> B[异常检测引擎]
    B --> C{检测到异常?}
    C -->|否| A
    C -->|是| D[根因分析引擎]
    D --> E[自愈决策引擎]
    
    E --> F{决策匹配}
    F -->|命中已知场景| G[执行预定义自愈策略]
    F -->|低置信度| H[人工审批流程]
    F -->|未知场景| I[创建人工处理工单+知识沉淀]
    
    G --> J{执行结果验证}
    J -->|成功恢复| K[记录Runbook+更新置信度]
    J -->|恢复失败| L[触发回滚操作]
    L --> M[升级至人工处理]
    
    H --> G
    
    K --> N[告警关闭]
    M --> N
    I --> N
    
    subgraph 安全边界
        S1[操作权限沙箱<br/>仅允许预定义操作集]
        S2[变更必审计<br/>所有操作记录完整日志]
        S3[熔断保护<br/>同一故障3次自愈失败后锁定]
    end
    
    E -.-> S1
    G -.-> S2
    L -.-> S3

AIOps自愈Agent由五个核心模块组成：

异常检测引擎：负责从Prometheus指标、ELK日志和分布式追踪中识别系统异常。这是整个Agent的感知层。

根因分析引擎：在检测到异常后，通过因果推断（如第一篇的方法）或调用链分析定位根因服务。输出结构化的根因报告：{root_cause_service, anomaly_type, confidence_score}。

自愈决策引擎：这是整个Agent的大脑。它维护一个自愈策略知识库（Runbook + 历史处理记录），将根因分析结果映射到自愈动作。决策引擎必须回答三个问题——该执行什么操作、操作的安全边界是什么、失败后的回滚方案是什么。

执行引擎：通过Kubernetes API、Ansible或自定义Operator执行修复操作。执行引擎运行在操作权限沙箱内，仅能调用决策引擎授权的API。

验证引擎：在修复操作完成后，验证系统指标是否在规定时间内恢复到正常范围。如果3次验证周期后仍未恢复，触发回滚和升级流程。

三、自愈策略知识库的构建与演化

flowchart LR
    A[历史故障工单] --> B[人工提取自愈策略]
    B --> C[Runbook模板化]
    C --> D[自愈策略库]
    
    E[实时自愈执行] --> F{结果判定}
    F -->|成功| G[策略置信度+1]
    F -->|失败| H[策略置信度-1 触发人工复盘]
    
    G --> D
    H --> I[策略优化更新]
    I --> D
    
    D --> J[策略版本管理 Git]

自愈策略知识库不是一次性构建完成的，而是一个持续演化的系统。初始版本可以从历史故障工单中提取。每条策略定义为：

# 自愈策略定义文件（YAML格式）
# 存储在Git仓库中，通过GitOps同步到Agent
apiVersion: autohealing.ops/v1
kind: HealingPolicy
metadata:
  name: deployment-image-pull-backoff
  description: "Pod因镜像拉取失败陷入ImagePullBackOff时自动扩容重建"
  # 策略适用范围（命名空间+标签选择器）
  scope:
    namespaces: ["production", "staging"]
    labelSelector:
      matchLabels:
        app.kubernetes.io/managed-by: "autohealing"
  # 策略优先级（数字越小优先级越高）
  priority: 10
  # 该策略是否处于激活状态
  enabled: true
spec:
  # 触发条件：基于PromQL表达式
  trigger:
    promql: |
      (
        kube_pod_status_phase{phase="Pending"} 
        * on(pod, namespace) 
        kube_pod_container_status_waiting_reason{reason="ImagePullBackOff"}
      ) > 0
    # 必须持续超过指定时长才触发（防止瞬时抖动）
    for: "2m"
    # 冷却时间：同一告警在此时间内不重复触发
    cooldown: "10m"
  
  # 自愈动作序列（按顺序执行）
  actions:
    # 动作1：记录当前状态用于回滚
    - name: "snapshot-current-state"
      type: "kubectl"
      command: |
        kubectl get pod ${POD_NAME} -n ${NAMESPACE} -o yaml > /tmp/healing-snapshot-${POD_NAME}.yaml
      timeout: "10s"
      # 该步骤失败是否阻断（true则失败后停止执行）
      critical: false
    
    # 动作2：删除故障Pod触发重建
    - name: "delete-failed-pod"
      type: "kubectl"
      command: |
        kubectl delete pod ${POD_NAME} -n ${NAMESPACE} --grace-period=30
      timeout: "60s"
      critical: true
    
    # 动作3：等待新Pod进入Ready状态
    - name: "wait-pod-ready"
      type: "kubectl"
      command: |
        kubectl wait --for=condition=Ready pod \
          -l app=${APP_LABEL} \
          -n ${NAMESPACE} \
          --timeout=300s
      timeout: "320s"
      critical: true
  
  # 回滚策略（自愈失败时的恢复操作）
  rollback:
    - name: "restore-from-snapshot"
      type: "kubectl"
      command: |
        # 快照文件可能不存在（如果快照步骤未执行），所以用|| true
        kubectl apply -f /tmp/healing-snapshot-${POD_NAME}.yaml || true
      timeout: "30s"
  
  # 验证规则：自愈完成后检查指标是否恢复
  verification:
    - promql: |
        kube_pod_status_phase{
          pod=~"${APP_LABEL}-.*",
          namespace="${NAMESPACE}",
          phase="Running"
        } >= ${EXPECTED_REPLICAS}
      for: "1m"
      # 3次验证都失败则触发回滚
      max_attempts: 3
      interval: "30s"

每条策略关联一个置信度评分。初始置信度设为0.5。每次成功执行后置信度增加0.05，每次失败后降低0.1。策略的置信度决定了执行模式：

• 置信度 ≥ 0.8：全自动执行，仅通知不审批
• 置信度 0.5-0.8：自动执行但需事后审核
• 置信度 < 0.5：必须人工审批后才执行

四、自愈Agent的安全边界与执行引擎

flowchart TD
    A[自愈决策引擎] --> B{操作权限校验}
    B -->|允许的操作| C[生成K8s API调用请求]
    B -->|禁止的操作| D[拒绝执行 记录告警]
    
    C --> E[操作审计日志<br/>记录: 谁触发/什么操作/作用对象/时间]
    E --> F[执行操作]
    F --> G{熔断检查}
    G -->|熔断触发| H[锁定该策略24h<br/>升级至人工介入]
    G -->|正常| I[验证结果]
    
    I --> J{指标恢复?}
    J -->|是| K[关闭告警 更新知识库]
    J -->|否| L[重试?]
    L -->|是 且 < 3次| F
    L -->|否 或 ≥ 3次| M[触发回滚+升级]

安全边界是自愈Agent设计的核心。它包含三层防护：

第一层：操作权限白名单。Agent只能执行知识库中预先定义的操作类型。在Kubernetes中，这意味着Agent的ServiceAccount只被授予特定的RBAC权限：

# AIOps自愈Agent的ServiceAccount和RBAC定义
apiVersion: v1
kind: ServiceAccount
metadata:
  name: autohealing-agent
  namespace: autohealing
  labels:
    app: aiops-autohealing

---
# ClusterRole：严格限制Agent的操作范围
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: autohealing-agent-role
rules:
# 仅允许读取Pod信息（用于快照备份）
- apiGroups: [""]
  resources: ["pods", "pods/log"]
  verbs: ["get", "list", "watch"]
# 仅允许删除Pod（通过delete触发重建）
# 不授予deletecollection防止批量删除
- apiGroups: [""]
  resources: ["pods"]
  verbs: ["delete"]
# 允许读取Deployment用于验证
- apiGroups: ["apps"]
  resources: ["deployments", "deployments/scale"]
  verbs: ["get", "list", "watch"]
# 允许扩缩容Deployment（受限的自愈操作）
- apiGroups: ["apps"]
  resources: ["deployments/scale"]
  verbs: ["update", "patch"]
# 允许创建Event用于记录自愈活动
- apiGroups: [""]
  resources: ["events"]
  verbs: ["create", "patch"]
# 明确禁止的操作（防止权限蔓延）
# - 禁止创建/删除命名空间
# - 禁止修改RBAC资源
# - 禁止访问Secret内容
# - 禁止exec/attach到Pod
# 这些规则通过"不列出"来实现，Kubernetes默认拒绝未授权的操作

第二层：熔断机制。同一故障场景如果在2小时内连续触发3次自愈且都失败，Agent自动锁定该策略24小时并升级至人工处理。熔断机制防止了"自愈→失败→再自愈→再失败"的死循环，这在实际情况中可能造成更大的损害。

"""
AIOps自愈Agent核心执行引擎
包含权限检查、熔断控制和执行审计
"""
import time
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from enum import Enum

# 配置日志格式
logging.basicConfig(
    level=logging.INFO,
    format='[%(asctime)s] [%(levelname)s] %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)
logger = logging.getLogger('autohealing')


class ActionStatus(Enum):
    """自愈动作执行状态"""
    PENDING = "pending"        # 等待执行
    RUNNING = "running"        # 执行中
    SUCCESS = "success"        # 执行成功
    FAILED = "failed"          # 执行失败
    ROLLED_BACK = "rolled_back"  # 已回滚
    CIRCUIT_BREAKER = "circuit_breaker"  # 熔断


@dataclass
class HealingExecution:
    """单次自愈执行记录"""
    incident_id: str
    policy_name: str
    start_time: datetime = field(default_factory=datetime.now)
    actions: List[Dict] = field(default_factory=list)
    status: ActionStatus = ActionStatus.PENDING
    end_time: Optional[datetime] = None
    error_message: Optional[str] = None


class AutoHealingAgent:
    """AIOps自愈Agent核心引擎"""
    
    # 熔断配置
    CIRCUIT_BREAKER_MAX_FAILURES = 3     # 2小时内最大失败次数
    CIRCUIT_BREAKER_WINDOW_HOURS = 2     # 熔断统计窗口
    CIRCUIT_BREAKER_LOCK_HOURS = 24      # 熔断后锁定时长
    
    def __init__(self):
        # 执行历史记录（生产环境应使用Redis等持久化存储）
        self.execution_history: Dict[str, List[HealingExecution]] = {}
        # 被熔断的策略列表 {policy_name: lock_until_timestamp}
        self.circuit_breaker_state: Dict[str, float] = {}
    
    def check_circuit_breaker(self, policy_name: str) -> Tuple[bool, str]:
        """
        检查指定策略是否处于熔断状态
        :param policy_name: 策略名称
        :return: (是否被熔断, 原因说明)
        """
        # 检查是否在锁定期间
        lock_until = self.circuit_breaker_state.get(policy_name)
        if lock_until is not None:
            if time.time() < lock_until:
                remaining = lock_until - time.time()
                hours = int(remaining // 3600)
                minutes = int((remaining % 3600) // 60)
                reason = (
                    f"策略 {policy_name} 处于熔断状态，"
                    f"剩余锁定时间: {hours}小时{minutes}分钟"
                )
                logger.warning(f"[熔断] {reason}")
                return True, reason
            else:
                # 锁定已过期，自动解除
                del self.circuit_breaker_state[policy_name]
                logger.info(f"[熔断] 策略 {policy_name} 锁定已过期，自动解除")
        
        # 检查窗口期内的失败次数
        window_start = datetime.now() - timedelta(
            hours=self.CIRCUIT_BREAKER_WINDOW_HOURS
        )
        
        recent_failures = [
            exec_record
            for exec_record in self.execution_history.get(policy_name, [])
            if exec_record.start_time >= window_start
            and exec_record.status in (
                ActionStatus.FAILED,
                ActionStatus.ROLLED_BACK
            )
        ]
        
        if len(recent_failures) >= self.CIRCUIT_BREAKER_MAX_FAILURES:
            # 触发熔断
            lock_until = time.time() + self.CIRCUIT_BREAKER_LOCK_HOURS * 3600
            self.circuit_breaker_state[policy_name] = lock_until
            
            reason = (
                f"策略 {policy_name} 在"
                f"{self.CIRCUIT_BREAKER_WINDOW_HOURS}小时内"
                f"失败 {len(recent_failures)} 次，触发熔断，"
                f"锁定 {self.CIRCUIT_BREAKER_LOCK_HOURS} 小时"
            )
            logger.critical(f"[熔断] {reason}")
            return True, reason
        
        return False, ""
    
    def execute_healing(
        self,
        incident_id: str,
        policy: Dict,
        context: Dict
    ) -> HealingExecution:
        """
        执行自愈策略
        :param incident_id: 事件ID（全局唯一）
        :param policy: 自愈策略定义
        :param context: 执行上下文（POD_NAME, NAMESPACE等变量）
        :return: 执行记录
        """
        policy_name = policy["metadata"]["name"]
        
        # 1. 熔断检查
        is_blocked, reason = self.check_circuit_breaker(policy_name)
        if is_blocked:
            execution = HealingExecution(
                incident_id=incident_id,
                policy_name=policy_name,
                status=ActionStatus.CIRCUIT_BREAKER,
                error_message=reason
            )
            logger.error(f"[自愈] 执行被熔断阻止: {reason}")
            return execution
        
        # 2. 创建执行记录
        execution = HealingExecution(
            incident_id=incident_id,
            policy_name=policy_name
        )
        
        logger.info(
            f"[自愈] 开始执行策略 {policy_name}, "
            f"事件ID: {incident_id}"
        )
        
        try:
            # 3. 按顺序执行每个action
            for action in policy["spec"]["actions"]:
                action_result = self._execute_action(
                    action, context
                )
                execution.actions.append(action_result)
                
                if not action_result.get("success", False):
                    logger.error(
                        f"[自愈] 动作 {action['name']} 执行失败: "
                        f"{action_result.get('error', '未知错误')}"
                    )
                    
                    # 关键步骤失败 -> 触发回滚
                    if action.get("critical", True):
                        self._rollback(
                            policy.get("spec", {}).get("rollback", []),
                            context,
                            execution
                        )
                        execution.status = ActionStatus.ROLLED_BACK
                        execution.end_time = datetime.now()
                        execution.error_message = (
                            f"关键步骤 {action['name']} 失败，已执行回滚"
                        )
                        self._record_execution(execution)
                        return execution
            
            # 4. 验证自愈效果
            verification_passed = self._verify_healing(
                policy.get("spec", {}).get("verification", []),
                context
            )
            
            if verification_passed:
                execution.status = ActionStatus.SUCCESS
                logger.info(
                    f"[自愈] 策略 {policy_name} 执行成功，"
                    f"验证通过"
                )
            else:
                execution.status = ActionStatus.FAILED
                execution.error_message = "验证未通过：指标未在预期时间内恢复"
                logger.error(
                    f"[自愈] 策略 {policy_name} 验证失败: "
                    f"{execution.error_message}"
                )
        
        except Exception as e:
            execution.status = ActionStatus.FAILED
            execution.error_message = f"执行异常: {str(e)}"
            logger.exception(
                f"[自愈] 策略 {policy_name} 执行异常"
            )
        
        execution.end_time = datetime.now()
        self._record_execution(execution)
        return execution
    
    def _execute_action(
        self,
        action: Dict,
        context: Dict
    ) -> Dict:
        """
        执行单个自愈动作
        :param action: 动作定义
        :param context: 上下文变量
        :return: {"success": bool, "output": str, "error": str}
        """
        action_name = action["name"]
        action_type = action["type"]
        timeout = int(action.get("timeout", "30").rstrip("s"))
        
        logger.info(f"[执行] {action_name} (类型={action_type})")
        
        try:
            if action_type == "kubectl":
                import subprocess
                
                # 替换命令中的模板变量
                command = action["command"]
                for key, value in context.items():
                    command = command.replace(
                        f"${{{key}}}", str(value)
                    )
                
                result = subprocess.run(
                    command,
                    shell=True,
                    capture_output=True,
                    text=True,
                    timeout=timeout
                )
                
                if result.returncode == 0:
                    return {
                        "success": True,
                        "output": result.stdout.strip(),
                        "action": action_name,
                        "duration_ms": int((time.time() - 
                            getattr(self, '_last_action_start', time.time())) * 1000)
                    }
                else:
                    return {
                        "success": False,
                        "error": (
                            f"退出码={result.returncode}, "
                            f"stderr={result.stderr.strip()[:200]}"
                        ),
                        "action": action_name
                    }
            else:
                return {
                    "success": False,
                    "error": f"不支持的动作类型: {action_type}",
                    "action": action_name
                }
        except subprocess.TimeoutExpired:
            return {
                "success": False,
                "error": f"动作超时（{timeout}秒）",
                "action": action_name
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "action": action_name
            }
    
    def _rollback(
        self,
        rollback_actions: List[Dict],
        context: Dict,
        execution: HealingExecution
    ) -> None:
        """
        执行回滚操作
        :param rollback_actions: 回滚动作列表
        :param context: 上下文变量
        :param execution: 当前执行记录
        """
        logger.warning(
            f"[回滚] 策略 {execution.policy_name} 开始回滚，"
            f"共 {len(rollback_actions)} 个回滚动作"
        )
        
        for action in rollback_actions:
            try:
                result = self._execute_action(action, context)
                if not result.get("success"):
                    # 回滚动作失败不中断，记录后继续执行
                    logger.error(
                        f"[回滚] 回滚动作 {action['name']} 失败: "
                        f"{result.get('error')}"
                    )
                else:
                    logger.info(f"[回滚] 回滚动作 {action['name']} 完成")
            except Exception as e:
                logger.error(f"[回滚] 回滚动作 {action['name']} 异常: {e}")
    
    def _verify_healing(
        self,
        verification_rules: List[Dict],
        context: Dict
    ) -> bool:
        """
        验证自愈效果（查询Prometheus）
        :param verification_rules: 验证规则列表
        :param context: 上下文
        :return: 验证是否通过
        """
        if not verification_rules:
            logger.info("[验证] 无验证规则，默认通过")
            return True
        
        for rule in verification_rules:
            max_attempts = rule.get("max_attempts", 3)
            interval = int(rule.get("interval", "30").rstrip("s"))
            
            for attempt in range(1, max_attempts + 1):
                logger.info(
                    f"[验证] 第 {attempt}/{max_attempts} 次尝试..."
                )
                
                # 查询Prometheus（简化实现，实际使用prometheus-api-client）
                # result = prometheus_client.query(rule["promql"])
                # if result_matches_expected:
                #     return True
                
                if attempt < max_attempts:
                    time.sleep(interval)
        
        return False
    
    def _record_execution(self, execution: HealingExecution) -> None:
        """
        记录执行历史
        :param execution: 执行记录
        """
        policy_name = execution.policy_name
        if policy_name not in self.execution_history:
            self.execution_history[policy_name] = []
        self.execution_history[policy_name].append(execution)
        
        # 同时输出审计日志
        audit_entry = {
            "timestamp": execution.start_time.isoformat(),
            "incident_id": execution.incident_id,
            "policy": policy_name,
            "status": execution.status.value,
            "duration_seconds": (
                (execution.end_time - execution.start_time).total_seconds()
                if execution.end_time and execution.start_time else None
            ),
            "error": execution.error_message
        }
        logger.info(f"[审计] {json.dumps(audit_entry, ensure_ascii=False)}")

第三层：全量审计日志。Agent的每一次操作——包括决策依据、执行动作、执行结果——都记录为不可篡改的审计日志，输出到独立的审计ELK集群。这既是安全合规的要求，也为事后的策略优化提供了数据基础。

五、总结

AIOps从监控到自愈的跨越，不是机器学习算法的突破，而是工程体系和安全设计的突破。五个核心引擎（检测、定位、决策、执行、验证）构成了一个完整的OODA（Observe-Orient-Decide-Act）闭环。其中决策引擎的策略置信度分级是信任建立的机制保障——从不信任到逐步信任，再到全自动执行，每个策略都走过了自己的成长路径。

落地建议：不要试图一步到位建设全场景自愈能力。从最成熟的场景开始——例如OOMKilled Pod自动重建或磁盘空间自动清理——这些场景的根因明确、修复动作简单、回滚成本低。跑通一个场景的完整闭环后，再复制模式扩展到更多场景。

最终目标是：Agent处理80%的已知故障模式，20%的未知场景由人工介入处理，且每次人工介入都沉淀为新的自愈策略。这个循环持续运转，Agent的能力边界就会不断扩展。