AI驱动的自动化运维：智能监控、根因分析与自愈系统

class DynamicBaseline:
    def __init__(self, metric_name, history_days=30):
        self.metric = metric_name
        self.history_days = history_days
        self.baseline_model = None
    
    def train(self, historical_data):
        """训练动态基线模型"""
        # 提取时间特征
        features = self._extract_features(historical_data)
        # 小时、星期、月份、节假日
        
        # 训练预测模型
        from sklearn.ensemble import GradientBoostingRegressor
        self.baseline_model = GradientBoostingRegressor()
        self.baseline_model.fit(features, historical_data["value"])
    
    def detect_anomaly(self, current_value, timestamp):
        """基于动态基线检测异常"""
        features = self._extract_features({"timestamp": timestamp})
        predicted = self.baseline_model.predict(features)
        
        # 计算残差
        residual = current_value - predicted[0]
        
        # 基于历史残差分布判断
        z_score = residual / self.residual_std
        
        if abs(z_score) > 3:
            return {
                "anomaly": True,
                "severity": "critical" if abs(z_score) > 5 else "warning",
                "expected": predicted[0],
                "actual": current_value,
                "deviation": f"{(residual/predicted[0]*100):.1f}%"
            }
        
        return {"anomaly": False}
    
    def _extract_features(self, data):
        """提取时间特征"""
        ts = data["timestamp"]
        return [[ts.hour, ts.weekday(), ts.month, is_holiday(ts)]]

class CorrelationMonitor:
    def __init__(self):
        self.metrics = {}
    
    def add_metric(self, name, data):
        self.metrics[name] = data
    
    def find_correlations(self, target_metric, window="1h"):
        """找出与目标指标相关的其他指标"""
        target = self.metrics[target_metric]
        
        correlations = {}
        for name, data in self.metrics.items():
            if name == target_metric:
                continue
            
            # 计算滚动相关性
            corr = target.rolling(window).corr(data)
            
            # 找出高相关时段
            high_corr_periods = corr[abs(corr) > 0.7]
            
            if len(high_corr_periods) > 0:
                correlations[name] = {
                    "avg_correlation": corr.mean(),
                    "max_correlation": corr.max(),
                    "lag": self._find_optimal_lag(target, data)
                }
        
        return correlations

class TimeSeriesAnomalyDetector:
    def __init__(self):
        self.models = {
            "statistical": StatisticalDetector(),    # 统计方法
            "isolation_forest": IsolationForestDetector(),  # 孤立森林
            "lstm_ae": LSTMAutoEncoder(),            # LSTM自编码器
        }
    
    def detect(self, timeseries, method="ensemble"):
        """多模型集成的异常检测"""
        if method == "ensemble":
            results = {}
            for name, model in self.models.items():
                results[name] = model.detect(timeseries)
            
            # 投票：多数模型认为异常才算异常
            anomaly_votes = sum(1 for r in results.values() if r["anomaly"])
            
            return {
                "anomaly": anomaly_votes >= 2,  # 至少2个模型认为异常
                "confidence": anomaly_votes / len(self.models),
                "model_details": results
            }
        
        return self.models[method].detect(timeseries)

class LogAnomalyDetector:
    def __init__(self, llm):
        self.llm = llm
        self.pattern_cache = {}
    
    def detect(self, log_lines):
        """检测日志中的异常"""
        anomalies = []
        
        # 1. 基于模式的检测
        for line in log_lines:
            pattern = self._extract_pattern(line)
            if pattern not in self.pattern_cache:
                # 新模式，需要AI分析
                is_anomaly = self._ai_analyze(line)
                self.pattern_cache[pattern] = is_anomaly
            
            if self.pattern_cache[pattern]:
                anomalies.append({
                    "line": line,
                    "pattern": pattern,
                    "timestamp": extract_timestamp(line)
                })
        
        # 2. 日志频率异常
        frequency_anomaly = self._detect_frequency_anomaly(log_lines)
        if frequency_anomaly:
            anomalies.append(frequency_anomaly)
        
        # 3. AI根因分析
        if anomalies:
            root_cause = self._analyze_root_cause(anomalies)
            return {
                "anomalies": anomalies,
                "root_cause": root_cause,
                "severity": self._assess_severity(anomalies)
            }
        
        return {"anomalies": [], "severity": "normal"}
    
    def _ai_analyze(self, log_line):
        """用LLM判断日志是否异常"""
        return self.llm.generate(f"""
        判断以下日志是否表示异常：
        {log_line}
        
        异常标准：
        - ERROR级别
        - 包含异常堆栈
        - 非预期行为
        - 性能问题信号
        
        只回答 true 或 false。
        """).strip().lower() == "true"

class RootCauseAnalyzer:
    def __init__(self, llm):
        self.llm = llm
    
    def analyze(self, incident):
        """根因分析"""
        # 1. 收集上下文
        context = {
            "alert": incident.alert_message,
            "metrics": incident.affected_metrics,
            "logs": incident.relevant_logs,
            "topology": incident.service_topology,
            "recent_changes": incident.recent_deployments,
        }
        
        # 2. AI推理根因
        analysis = self.llm.generate(f"""
        系统发生了告警，请分析根因：
        
        告警信息：{context['alert']}
        
        异常指标：
        {json.dumps(context['metrics'], indent=2)}
        
        相关日志（最近5分钟）：
        {context['logs'][:5000]}
        
        服务拓扑：
        {context['topology']}
        
        最近变更：
        {context['recent_changes']}
        
        请分析：
        1. 最可能的根因（按置信度排序，top 3）
        2. 影响范围评估
        3. 建议的排查步骤
        4. 临时缓解措施
        5. 永久修复方案
        
        格式：JSON
        """)
        
        return analysis
    
    def build_causal_graph(self, metrics, correlations):
        """构建因果图"""
        graph = {}
        for metric, corr in correlations.items():
            if abs(corr["avg_correlation"]) > 0.7:
                # 可能的因果关系
                graph[metric] = {
                    "parents": self._find_causes(metric, corr),
                    "children": self._find_effects(metric, corr),
                    "confidence": abs(corr["avg_correlation"])
                }
        
        return graph

class AutoHealingSystem:
    def __init__(self, llm):
        self.llm = llm
        self.playbooks = self._load_playbooks()
        self.safety_guard = SafetyGuard()
    
    def handle(self, incident):
        """处理告警事件"""
        # 1. 根因分析
        root_cause = self._analyze(incident)
        
        # 2. 匹配修复方案
        fix = self._match_playbook(root_cause)
        
        if fix:
            # 3. 安全检查
            if self.safety_guard.is_safe(fix):
                # 4. 执行修复
                result = self._execute(fix)
                
                # 5. 验证修复效果
                if self._verify_fix(incident):
                    return {"status": "auto_healed", "fix": fix}
                else:
                    return {"status": "fix_failed", "escalate": True}
            else:
                # 需要人工审批
                return {"status": "needs_approval", "fix": fix}
        
        else:
            # 没有匹配的修复方案，生成建议
            suggestion = self._generate_fix_suggestion(root_cause)
            return {"status": "needs_manual", "suggestion": suggestion}
    
    def _match_playbook(self, root_cause):
        """匹配预定义的修复方案"""
        playbooks = {
            "high_memory": {
                "condition": "内存使用率>90%",
                "action": "重启内存泄漏的服务",
                "command": "kubectl rollout restart deployment {service}",
                "risk_level": "low"
            },
            "disk_full": {
                "condition": "磁盘使用率>95%",
                "action": "清理日志和临时文件",
                "command": "find /var/log -name '*.log' -mtime +7 -delete",
                "risk_level": "low"
            },
            "high_latency": {
                "condition": "P99延迟>阈值",
                "action": "扩容服务实例",
                "command": "kubectl scale deployment {service} --replicas={current}+1",
                "risk_level": "medium"
            }
        }
        
        for name, playbook in playbooks.items():
            if self._matches(root_cause, playbook["condition"]):
                return playbook
        
        return None

class SafetyGuard:
    UNSAFE_ACTIONS = [
        "删除数据库",
        "删除用户数据",
        "关闭防火墙",
        "修改密码",
        "降低安全配置"
    ]
    
    def is_safe(self, fix):
        """检查修复方案是否安全"""
        for unsafe in self.UNSAFE_ACTIONS:
            if unsafe in fix.get("command", "").lower():
                return False
        
        # 高风险操作需要人工确认
        if fix.get("risk_level") == "high":
            return False
        
        # 生产环境影响范围检查
        if fix.get("scope", "").startswith("production"):
            return False
        
        return True

class MicroserviceHealing:
    def handle_high_error_rate(self, service_name, error_rate):
        """处理微服务错误率飙升"""
        # 1. 诊断
        diagnosis = self._diagnose(service_name, error_rate)
        
        # 2. 根据诊断结果选择修复策略
        if diagnosis["cause"] == "bad_deployment":
            # 回滚到上一版本
            return self._rollback(service_name, diagnosis["bad_version"])
        
        elif diagnosis["cause"] == "dependency_failure":
            # 降级依赖服务
            return self._circuit_break(service_name, diagnosis["dependency"])
        
        elif diagnosis["cause"] == "resource_exhaustion":
            # 自动扩容
            return self._scale_out(service_name, factor=2)
        
        elif diagnosis["cause"] == "traffic_spike":
            # 限流保护
            return self._enable_rate_limit(service_name, limit=diagnosis["normal_load"])
        
        else:
            # 未知原因，升级人工
            return self._escalate(service_name, diagnosis)

class AIOpsMetrics:
    def evaluate(self):
        return {
            "mttr": self._mean_time_to_recovery(),  # 平均恢复时间
            "mttd": self._mean_time_to_detect(),    # 平均检测时间
            "false_positive_rate": self._false_positive_rate(),  # 误报率
            "auto_heal_rate": self._auto_heal_rate(),  # 自动修复率
            "incident_reduction": self._incident_trend()  # 事故趋势
        }

# 典型改善：
# MTTR: 从45分钟→8分钟 (82%降低)
# MTTD: 从10分钟→30秒 (95%降低)
# 误报率: 从35%→8% (77%降低)
# 自动修复率: 0%→45%