大模型赋能智能运维：构建云环境自动化故障处置平台

// 统一事件模型
@Data
@Builder
public class OpsEvent {
    private String eventId;           // 全局唯一ID
    private EventType type;           // METRIC / LOG / TRACE / CHANGE
    private String source;            // 来源系统（如prometheus-01）
    private String cluster;           // 集群名称
    private String service;           // 服务名称
    private String instance;          // 实例IP
    private Map<String, String> tags; // 标签（如 env=prod, region=us-east）
    private String content;           // 事件内容（JSON格式）
    private String rawData;           // 原始数据
    private Instant timestamp;        // 事件发生时间
    private Instant receiveTime;      // 平台接收时间
}

// 事件标准化处理器
@Component
public class EventNormalizer {
    
    public OpsEvent normalize(Object source) {
        if (source instanceof PrometheusAlert) {
            PrometheusAlert alert = (PrometheusAlert) source;
            return OpsEvent.builder()
                .eventId(UUID.randomUUID().toString())
                .type(EventType.METRIC)
                .source("prometheus")
                .service(alert.getLabels().get("service"))
                .tags(alert.getLabels())
                .content(alert.getAnnotations().get("summary"))
                .timestamp(alert.getStartsAt())
                .build();
        }
        // ELK日志、JaegerTrace、ChangeRecord等其他类型处理...
        throw new UnsupportedOperationException("Unknown source type");
    }
}

# 基于K8s API构建服务依赖图
from kubernetes import client, config
import networkx as nx

class ServiceTopologyBuilder:
    def __init__(self):
        config.load_incluster_config()
        self.core_v1 = client.CoreV1Api()
        self.network_v1 = client.NetworkingV1Api()
        
    def build_topology(self):
        G = nx.DiGraph()
        
        # 1. 获取所有Service
        services = self.core_v1.list_service_for_all_namespaces()
        for svc in services.items:
            G.add_node(svc.metadata.name, 
                       type='service',
                       namespace=svc.metadata.namespace,
                       cluster_ip=svc.spec.cluster_ip,
                       labels=svc.metadata.labels)
        
        # 2. 获取所有Pod，建立Service->Pod关系
        pods = self.core_v1.list_pod_for_all_namespaces()
        for pod in pods.items:
            if pod.metadata.labels and 'app' in pod.metadata.labels:
                # 通过标签匹配找到所属Service
                for svc in services.items:
                    if self._match_labels(svc.spec.selector, pod.metadata.labels):
                        G.add_edge(svc.metadata.name, pod.metadata.name, type='runs')
                        break
        
        # 3. 通过NetworkPolicy或Istio判断服务间调用关系
        # (实际生产环境中可通过APM数据增强)
        return G

class AnomalyDetector:
    def __init__(self):
        self.statistical_detector = StatisticalDetector()  # 3-sigma/IQR
        self.ml_detector = MLDetector()  # Isolation Forest
        self.llm_verifier = LLMVerifier()  # 大模型二次确认
    
    def detect(self, metric_series, context):
        # 阶段1：统计检测（快速筛选）
        stat_anomalies = self.statistical_detector.detect(metric_series)
        if not stat_anomalies:
            return []
        
        # 阶段2：机器学习检测（降噪）
        ml_anomalies = self.ml_detector.detect(metric_series, stat_anomalies)
        if not ml_anomalies:
            return []
        
        # 阶段3：大模型验证（减少误报）
        # 传入该服务的历史模式、变更记录等上下文
        confirmed = []
        for anomaly in ml_anomalies:
            is_real = self.llm_verifier.verify(
                anomaly=anomaly,
                historical_pattern=context['historical_pattern'],
                recent_changes=context['changes'],
                related_services=context['related_services']
            )
            if is_real:
                confirmed.append(anomaly)
        
        return confirmed

# 根因定位引擎
class RootCauseEngine:
    def __init__(self, neo4j_driver, llm):
        self.driver = neo4j_driver
        self.llm = llm
        
    def locate_root_cause(self, anomalies: List[Anomaly]) -> RootCauseResult:
        # 1. 从知识图谱中提取受影响节点的上下游依赖
        affected_services = [a.service for a in anomalies]
        
        with self.driver.session() as session:
            # Cypher查询：查找所有受影响节点的共同上游
            result = session.run("""
                MATCH (s:Service)-[:DEPENDS_ON*1..5]->(upstream:Service)
                WHERE s.name IN $services
                RETURN upstream.name AS candidate, COUNT(*) AS affected_count
                ORDER BY affected_count DESC
                LIMIT 5
            """, services=affected_services)
            candidates = [record['candidate'] for record in result]
        
        # 2. 结合时间窗口：最早出现的异常更有可能是根因
        # 3. 结合变更记录：最近变更的服务风险最高
        candidates_with_metadata = []
        for candidate in candidates:
            first_anomaly_time = min([a.timestamp for a in anomalies if a.service == candidate], default=None)
            recent_changes = get_recent_changes(candidate, hours=1)
            candidates_with_metadata.append({
                'service': candidate,
                'first_occurrence': first_anomaly_time,
                'recent_changes': recent_changes,
                'impacted_downstream': [a.service for a in anomalies if candidate in a.dependencies]
            })
        
        # 4. 使用大模型进行综合推理
        root_cause = self.llm.invoke(
            f"根据以下候选根因分析，结合运维经验判断最可能的故障源：{candidates_with_metadata}"
        )
        
        return RootCauseResult(
            root_service=root_cause['service'],
            confidence=root_cause['confidence'],
            reasoning=root_cause['explanation'],
            evidence=root_cause['evidence']
        )

# 处置方案生成Agent（使用LangGraph实现状态机）
from langgraph.graph import StateGraph, END

class RemediationAgent:
    def __init__(self, llm, tools, knowledge_base):
        self.llm = llm
        self.tools = tools  # restart_pod, scale_up, rollback_config, etc.
        self.kb = knowledge_base
        
    def build_graph(self):
        workflow = StateGraph(RemediationState)
        
        # 定义节点
        workflow.add_node("analyze_root_cause", self.analyze_root_cause)
        workflow.add_node("search_history", self.search_history)
        workflow.add_node("generate_plan", self.generate_plan)
        workflow.add_node("assess_risk", self.assess_risk)
        workflow.add_node("execute_plan", self.execute_plan)
        workflow.add_node("verify_result", self.verify_result)
        workflow.add_node("escalate", self.escalate_to_human)
        
        # 定义边（条件路由）
        workflow.set_entry_point("analyze_root_cause")
        workflow.add_edge("analyze_root_cause", "search_history")
        workflow.add_edge("search_history", "generate_plan")
        workflow.add_edge("generate_plan", "assess_risk")
        workflow.add_conditional_edges(
            "assess_risk",
            lambda state: "execute_plan" if state['risk_score'] < 0.7 else "escalate"
        )
        workflow.add_edge("execute_plan", "verify_result")
        workflow.add_conditional_edges(
            "verify_result",
            lambda state: END if state['success'] else "escalate"
        )
        
        return workflow.compile()
    
    def generate_plan(self, state):
        """生成具体处置步骤"""
        prompt = f"""
        根因：{state['root_cause']}
        历史相似案例：{state['history_cases']}
        可用工具列表：{[t.name for t in self.tools]}
        
        请生成一个处置计划，包含：
        1. 操作步骤（按顺序）
        2. 每一步调用的工具和参数
        3. 预期结果
        4. 回滚条件
        
        以JSON格式输出。
        """
        plan = self.llm.invoke(prompt, response_format='json')
        state['plan'] = json.loads(plan)
        return state
    
    def assess_risk(self, state):
        """风险评估：操作是否可能造成更大影响"""
        # 检查是否有"删除"、"重启生产主库"等高危操作
        plan_text = str(state['plan'])
        high_risk_keywords = ['delete', 'drop', 'reboot', 'shutdown', 'force']
        for keyword in high_risk_keywords:
            if keyword in plan_text.lower():
                # 如果是高危操作，进行影响面分析
                impact = self.impact_analysis(state['plan'])
                state['risk_score'] = impact['score']
                state['risk_detail'] = impact['detail']
                return state
        
        state['risk_score'] = 0.3  # 默认低风险
        return state

@Service
public class RemediationExecutor {
    
    @Autowired
    private K8sClient k8sClient;
    @Autowired
    private TerraformClient tfClient;
    @Autowired
    private AnsibleClient ansibleClient;
    
    /**
     * 执行处置计划，支持暂停点（等待人工确认高风险步骤）
     */
    public ExecutionResult execute(RemediationPlan plan, ExecutionContext ctx) {
        List<ExecutionStep> steps = plan.getSteps();
        ExecutionResult result = new ExecutionResult();
        
        for (int i = 0; i < steps.size(); i++) {
            ExecutionStep step = steps.get(i);
            
            // 检查是否需要人工审批
            if (step.requiresApproval() && !ctx.isApproved(step.getId())) {
                // 暂停执行，等待外部审批信号
                result.setStatus(ExecutionStatus.PENDING_APPROVAL);
                result.setPendingStep(step);
                return result;
            }
            
            try {
                // 执行操作并记录快照（用于回滚）
                StateSnapshot snapshot = executeStep(step);
                result.getSnapshots().add(snapshot);
                result.getExecutedSteps().add(step);
                
            } catch (Exception e) {
                // 执行失败，触发自动回滚
                log.error("Step {} failed: {}", step.getId(), e.getMessage());
                rollback(result.getSnapshots());
                result.setStatus(ExecutionStatus.ROLLED_BACK);
                result.setError(e.getMessage());
                return result;
            }
        }
        
        result.setStatus(ExecutionStatus.SUCCESS);
        return result;
    }
    
    /**
     * 回滚：按执行顺序的反向执行回滚操作
     */
    private void rollback(List<StateSnapshot> snapshots) {
        Collections.reverse(snapshots);
        for (StateSnapshot snapshot : snapshots) {
            if (snapshot.getType() == SnapshotType.POD_RESTART) {
                // 回滚重启：无操作（重启本身不可逆，但可以检查状态）
                continue;
            } else if (snapshot.getType() == SnapshotType.SCALE_CHANGE) {
                // 回滚扩容：缩容回原数量
                k8sClient.scaleDeployment(snapshot.getDeployment(), snapshot.getOriginalReplicas());
            } else if (snapshot.getType() == SnapshotType.CONFIG_CHANGE) {
                // 回滚配置变更
                ansibleClient.rollbackConfig(snapshot.getConfigVersion());
            }
        }
    }
}

# 执行权限配置（OPA策略）
apiVersion: v1
kind: ConfigMap
metadata:
  name: remediation-policies
data:
  policy.rego: |
    package remediation
    
    # 默认拒绝所有操作
    default allow = false
    
    # 允许只读操作（日志查询、状态查看）
    allow = true {
        input.operation in ["query_logs", "get_pod_status", "get_metrics"]
    }
    
    # 允许指定服务在非生产环境执行操作
    allow = true {
        input.environment == "staging"
        input.service in authorized_staging_services
        input.operation in allowed_staging_operations
    }
    
    # 生产环境高风险操作需要双人审批
    allow = true {
        input.environment == "production"
        input.operation in ["restart_pod", "scale_deployment"]
        input.approvals == ["sre-oncall", "service-owner"]
    }

class VerificationEngine:
    def __init__(self, prometheus_client):
        self.prometheus = prometheus_client
    
    def verify_remediation(self, incident, executed_plan):
        """验证处置措施是否有效"""
        
        # 1. 检查根因指标是否恢复正常
        root_metric = incident['root_cause_metric']
        current_value = self.prometheus.query(
            f"{root_metric['name']}{{service='{root_metric['service']}'}}"
        )
        expected_value = root_metric['baseline']
        metric_ok = abs(current_value - expected_value) / expected_value < 0.1
        
        # 2. 检查副作用：其他服务是否受影响
        side_effects = self.check_side_effects(
            affected_services=incident['related_services'],
            duration=executed_plan['execution_time']
        )
        
        # 3. 检查用户体验指标
        error_rate = self.prometheus.query(
            f"sum(rate(http_requests_total{{status=~'5..'}}[5m])) / sum(rate(http_requests_total[5m]))"
        )
        ux_ok = error_rate < incident['sla_threshold']
        
        return VerificationResult(
            success=metric_ok and not side_effects and ux_ok,
            metric_ok=metric_ok,
            side_effects=side_effects,
            ux_ok=ux_ok
        )

class IncidentLearning:
    def __init__(self, knowledge_base, llm):
        self.kb = knowledge_base
        self.llm = llm
        
    def archive_incident(self, incident, resolution):
        """将成功处置的故障归档到知识库"""
        
        # 使用大模型生成结构化摘要
        summary = self.llm.invoke(f"""
        请将以下故障处置过程总结为一个可复用的知识条目：
        
        故障描述：{incident['description']}
        症状：{incident['symptoms']}
        根因：{incident['root_cause']}
        处置过程：{resolution['steps']}
        耗时：{resolution['duration']}
        验证结果：{resolution['verification']}
        
        输出格式：
        - 故障类型：[类型标签]
        - 典型症状：[关键症状指标]
        - 修复步骤：[操作步骤]
        - 适用场景：[条件]
        - 风险提示：[注意事项]
        """)
        
        # 存入向量库，供后续相似故障检索
        self.kb.add_incident(
            incident_id=incident['id'],
            summary=summary,
            symptoms=incident['symptoms'],
            resolution=resolution['steps'],
            tags=extract_tags(summary)
        )