AI驅動SRE：智能告警根因分析與自愈系統完整指南 2026
DevOps运维
AI驅動SRE：智能告警根因分析與自愈系統完整指南 2026
凌晨3點，你的手機被200條告警轟炸，但真正需要處理的可能只有1條。這是SRE工程師的日常噩夢。傳統告警系統基於靜態閾值，無法區分症狀與根因，更無法自動修復。2026年，AI正在重塑SRE：從告警降噪到根因定位，從自動修復到自愈系統，AIOps已經從概念走向生產。
核心概念速覽
| 概念 | 說明 | 適用場景 |
|---|---|---|
| 告警降噪 | 合併/抑制/靜默冗餘告警 | 告警風暴 |
| 告警聚合 | 基於相關性將多個告警合併 | 根因定位 |
| 根因分析(RCA) | 定位故障的根本原因 | 故障排查 |
| Runbook | 標準化的故障修復流程 | 自動化修復 |
| 自愈系統 | 無需人工干預的自動修復 | 7x24運維 |
| AIOps | AI驅動的IT運維 | 智能運維 |
| 異常檢測 | 基於ML的異常識別 | 主動發現 |
| 混沌工程 | 主動注入故障驗證韌性 | 可靠性驗證 |
五大痛點分析
- 告警風暴：微服務架構下，一個服務故障引發級聯告警，5分鐘內產生數百條告警，真正的根因被淹沒
- 根因定位困難：平均MTTR（平均恢復時間）中，70%的時間花在定位根因上，而非修復
- 重複性修復：80%的運維操作是重複的（重啟服務、擴容、清理磁盤），但每次都需人工介入
- 知識斷層：資深工程師離職後，故障處理經驗隨之流失，新人無法快速上手
- 被動響應：傳統監控只能在故障發生後告警，無法預測和預防
分步實操：5個核心模式
模式一：告警降噪與聚合
運行環境：Python 3.12+ / Prometheus + Alertmanager
# alert_dedup.py - 智能告警降噪与聚合
# 依赖：pip install prometheus-client scikit-learn numpy
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional
import hashlib
import re
from collections import defaultdict
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.cluster import DBSCAN
class AlertSeverity(Enum):
CRITICAL = "critical"
WARNING = "warning"
INFO = "info"
class AlertStatus(Enum):
FIRING = "firing"
RESOLVED = "resolved"
SUPPRESSED = "suppressed"
ACKNOWLEDGED = "acknowledged"
@dataclass
class Alert:
"""告警数据模型"""
id: str
name: str
severity: AlertSeverity
status: AlertStatus
labels: dict[str, str]
annotations: dict[str, str]
fired_at: datetime
resolved_at: Optional[datetime] = None
fingerprint: str = ""
def __post_init__(self):
if not self.fingerprint:
# 基于标签生成指纹，用于去重
label_str = "|".join(f"{k}={v}" for k, v in sorted(self.labels.items()))
self.fingerprint = hashlib.md5(label_str.encode()).hexdigest()[:12]
@dataclass
class AlertGroup:
"""告警聚合组"""
group_id: str
root_alert: Alert
related_alerts: list[Alert] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.now)
root_cause_hypothesis: str = ""
@property
def total_alerts(self) -> int:
return 1 + len(self.related_alerts)
@property
def severity(self) -> AlertSeverity:
return self.root_alert.severity
class AlertDeduplicator:
"""告警去重器 - 基于指纹和窗口"""
def __init__(self, dedup_window: timedelta = timedelta(minutes=5)):
self.dedup_window = dedup_window
self.recent_alerts: dict[str, list[Alert]] = defaultdict(list)
def should_suppress(self, alert: Alert) -> bool:
"""判断告警是否应被抑制（重复告警）"""
fingerprint = alert.fingerprint
now = datetime.now()
# 清理过期记录
self.recent_alerts[fingerprint] = [
a for a in self.recent_alerts[fingerprint]
if now - a.fired_at < self.dedup_window
]
# 检查是否在去重窗口内已存在
if self.recent_alerts[fingerprint]:
return True # 抑制重复告警
self.recent_alerts[fingerprint].append(alert)
return False
class AlertAggregator:
"""告警聚合器 - 基于相关性和聚类"""
def __init__(self, similarity_threshold: float = 0.6):
self.similarity_threshold = similarity_threshold
self.alert_groups: list[AlertGroup] = []
self.vectorizer = TfidfVectorizer(max_features=1000)
def aggregate(self, alerts: list[Alert]) -> list[AlertGroup]:
"""聚合相关告警"""
if not alerts:
return []
# 1. 提取告警文本特征
texts = [
f"{a.name} {' '.join(a.labels.values())} {a.annotations.get('summary', '')}"
for a in alerts
]
# 2. TF-IDF向量化
tfidf_matrix = self.vectorizer.fit_transform(texts)
# 3. DBSCAN聚类
clustering = DBSCAN(
eps=1.0 - self.similarity_threshold,
min_samples=1,
metric="cosine"
).fit(tfidf_matrix.toarray())
# 4. 按聚类结果分组
groups: dict[int, list[Alert]] = defaultdict(list)
for i, label in enumerate(clustering.labels_):
groups[label].append(alerts[i])
# 5. 创建AlertGroup
result = []
for cluster_id, cluster_alerts in groups.items():
# 选择最严重的告警作为根告警
root_alert = min(
cluster_alerts,
key=lambda a: {"critical": 0, "warning": 1, "info": 2}[a.severity.value]
)
related = [a for a in cluster_alerts if a.id != root_alert.id]
group = AlertGroup(
group_id=f"group-{cluster_id}-{root_alert.fingerprint}",
root_alert=root_alert,
related_alerts=related,
root_cause_hypothesis=self._generate_hypothesis(root_alert, related),
)
result.append(group)
self.alert_groups.extend(result)
return result
def _generate_hypothesis(self, root: Alert, related: list[Alert]) -> str:
"""生成根因假设"""
if not related:
return f"单点故障：{root.name} - {root.annotations.get('summary', '')}"
# 基于关联告警推断根因
services = set()
for a in [root] + related:
if "service" in a.labels:
services.add(a.labels["service"])
if "namespace" in a.labels:
services.add(a.labels["namespace"])
if len(services) > 1:
return f"级联故障：{root.name} 影响了 {', '.join(services)}"
elif "node" in root.labels:
return f"节点故障：{root.labels['node']} 上的服务异常"
else:
return f"服务故障：{root.name} 可能是根因"
class AlertNoiseReducer:
"""告警降噪器 - 综合去重+聚合+静默"""
def __init__(self):
self.deduplicator = AlertDeduplicator()
self.aggregator = AlertAggregator()
def process_alerts(self, alerts: list[Alert]) -> list[AlertGroup]:
"""处理告警流：去重 → 聚合 → 分组"""
# 第一步：去重
unique_alerts = [a for a in alerts if not self.deduplicator.should_suppress(a)]
# 第二步：聚合
groups = self.aggregator.aggregate(unique_alerts)
return groups
def get_stats(self) -> dict:
"""获取降噪统计"""
total_suppressed = sum(
len(v) - 1 for v in self.deduplicator.recent_alerts.values() if len(v) > 1
)
return {
"total_groups": len(self.alert_groups),
"total_suppressed": total_suppressed,
"dedup_rate": f"{total_suppressed / max(total_suppressed + len(self.alert_groups), 1) * 100:.1f}%",
}
# 使用示例
if __name__ == "__main__":
reducer = AlertNoiseReducer()
alerts = [
Alert(id="1", name="HighCPU", severity=AlertSeverity.CRITICAL,
status=AlertStatus.FIRING, labels={"service": "api-gateway", "node": "node-1"},
annotations={"summary": "CPU使用率超过90%"}, fired_at=datetime.now()),
Alert(id="2", name="HighMemory", severity=AlertSeverity.WARNING,
status=AlertStatus.FIRING, labels={"service": "api-gateway", "node": "node-1"},
annotations={"summary": "内存使用率超过85%"}, fired_at=datetime.now()),
Alert(id="3", name="PodCrashLooping", severity=AlertSeverity.CRITICAL,
status=AlertStatus.FIRING, labels={"service": "api-gateway", "namespace": "prod"},
annotations={"summary": "Pod持续重启"}, fired_at=datetime.now()),
Alert(id="4", name="HighCPU", severity=AlertSeverity.CRITICAL,
status=AlertStatus.FIRING, labels={"service": "api-gateway", "node": "node-1"},
annotations={"summary": "CPU使用率超过90%"}, fired_at=datetime.now()), # 重复
]
groups = reducer.process_alerts(alerts)
for g in groups:
print(f"[{g.severity.value}] {g.root_cause_hypothesis} ({g.total_alerts} alerts)")
print(f"\nStats: {reducer.get_stats()}")
模式二：AI根因分析
基於知識圖譜和因果推斷的根因分析：
# root_cause_analyzer.py
# 依赖：pip install openai networkx pydantic
from openai import OpenAI
from pydantic import BaseModel, Field
from typing import Optional
import networkx as nx
from datetime import datetime
import json
class CausalNode(BaseModel):
"""因果图节点"""
id: str
type: str # service, infrastructure, config, deployment
name: str
status: str = "healthy" # healthy, degraded, down
metrics: dict[str, float] = Field(default_factory=dict)
class CausalEdge(BaseModel):
"""因果图边"""
source: str
target: str
relation: str # depends_on, causes, correlates_with
weight: float = 1.0
class RootCauseReport(BaseModel):
"""根因分析报告"""
incident_id: str
timestamp: datetime
root_causes: list[dict]
confidence: float
evidence: list[str]
recommended_actions: list[str]
causal_path: list[str]
class AIRootCauseAnalyzer:
"""AI根因分析器 - 知识图谱 + LLM推理"""
def __init__(self, api_key: str = ""):
self.client = OpenAI(api_key=api_key) if api_key else None
self.causal_graph = nx.DiGraph()
self._build_default_graph()
def _build_default_graph(self):
"""构建默认的微服务因果图"""
edges = [
("user-request", "api-gateway", "depends_on"),
("api-gateway", "auth-service", "depends_on"),
("api-gateway", "user-service", "depends_on"),
("api-gateway", "order-service", "depends_on"),
("order-service", "payment-service", "depends_on"),
("order-service", "inventory-service", "depends_on"),
("payment-service", "database-primary", "depends_on"),
("user-service", "database-primary", "depends_on"),
("user-service", "redis-cache", "depends_on"),
("api-gateway", "redis-cache", "depends_on"),
("database-primary", "database-replica", "replicates_to"),
("node-cpu", "api-gateway", "affects"),
("node-memory", "api-gateway", "affects"),
("network-latency", "api-gateway", "affects"),
("deployment", "api-gateway", "may_cause"),
("config-change", "api-gateway", "may_cause"),
]
for source, target, relation in edges:
self.causal_graph.add_edge(source, target, relation=relation)
def analyze(
self,
alert_group: dict,
metrics_snapshot: dict[str, dict],
) -> RootCauseReport:
"""执行根因分析"""
# 1. 基于图算法的候选根因
affected_services = self._identify_affected_services(alert_group)
candidates = self._find_root_candidates(affected_services)
# 2. 基于指标的异常评分
anomaly_scores = self._compute_anomaly_scores(metrics_snapshot, candidates)
# 3. LLM推理（如果可用）
if self.client:
llm_analysis = self._llm_reasoning(alert_group, candidates, anomaly_scores)
else:
llm_analysis = self._rule_based_reasoning(alert_group, candidates, anomaly_scores)
return llm_analysis
def _identify_affected_services(self, alert_group: dict) -> list[str]:
"""识别受影响的服务"""
services = set()
for alert in alert_group.get("alerts", []):
labels = alert.get("labels", {})
if "service" in labels:
services.add(labels["service"])
if "node" in labels:
services.add(labels["node"])
return list(services)
def _find_root_candidates(self, affected: list[str]) -> list[str]:
"""基于因果图找到候选根因节点"""
candidates = set()
for service in affected:
if service in self.causal_graph:
# 向上游追溯：找到所有可能影响该服务的节点
predecessors = nx.ancestors(self.causal_graph, service)
candidates.update(predecessors)
candidates.add(service)
return list(candidates)
def _compute_anomaly_scores(
self,
metrics: dict[str, dict],
candidates: list[str],
) -> dict[str, float]:
"""计算异常评分"""
scores = {}
for candidate in candidates:
if candidate in metrics:
m = metrics[candidate]
# 综合异常评分
cpu_score = min(m.get("cpu_usage", 0) / 100, 1.0)
mem_score = min(m.get("memory_usage", 0) / 100, 1.0)
error_score = min(m.get("error_rate", 0) / 10, 1.0)
latency_score = min(m.get("p99_latency_ms", 0) / 5000, 1.0)
scores[candidate] = (
cpu_score * 0.3 +
mem_score * 0.2 +
error_score * 0.3 +
latency_score * 0.2
)
return scores
def _llm_reasoning(
self,
alert_group: dict,
candidates: list[str],
anomaly_scores: dict[str, float],
) -> RootCauseReport:
"""使用LLM进行根因推理"""
prompt = f"""你是一位资深SRE工程师，请分析以下告警和指标数据，定位根因。
## 告警信息
{json.dumps(alert_group, indent=2, ensure_ascii=False)}
## 候选根因节点
{candidates}
## 异常评分
{json.dumps(anomaly_scores, indent=2)}
## 因果图路径
{self._get_causal_paths(candidates)}
请以JSON格式返回分析结果，包含：
1. root_causes: 根因列表，每个包含service、reason、confidence
2. evidence: 支持证据列表
3. recommended_actions: 推荐修复动作列表
4. causal_path: 从根因到症状的因果路径
"""
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "你是SRE根因分析专家，擅长从告警和指标中定位故障根因。"},
{"role": "user", "content": prompt},
],
response_format={"type": "json_object"},
temperature=0.1,
)
result = json.loads(response.choices[0].message.content)
return RootCauseReport(
incident_id=f"INC-{datetime.now().strftime('%Y%m%d%H%M%S')}",
timestamp=datetime.now(),
root_causes=result.get("root_causes", []),
confidence=max(
rc.get("confidence", 0.5) for rc in result.get("root_causes", [{"confidence": 0.5}])
),
evidence=result.get("evidence", []),
recommended_actions=result.get("recommended_actions", []),
causal_path=result.get("causal_path", []),
)
def _rule_based_reasoning(
self,
alert_group: dict,
candidates: list[str],
anomaly_scores: dict[str, float],
) -> RootCauseReport:
"""基于规则的根因推理（LLM不可用时的后备方案）"""
# 按异常评分排序
sorted_candidates = sorted(
anomaly_scores.items(),
key=lambda x: x[1],
reverse=True,
)
root_causes = []
for candidate, score in sorted_candidates[:3]:
if score > 0.3:
root_causes.append({
"service": candidate,
"reason": f"异常评分 {score:.2f}，指标异常",
"confidence": min(score, 0.95),
})
return RootCauseReport(
incident_id=f"INC-{datetime.now().strftime('%Y%m%d%H%M%S')}",
timestamp=datetime.now(),
root_causes=root_causes,
confidence=root_causes[0]["confidence"] if root_causes else 0.0,
evidence=[f"异常评分最高的服务: {sorted_candidates[0][0]}"] if sorted_candidates else [],
recommended_actions=self._generate_actions(root_causes),
causal_path=[c["service"] for c in root_causes],
)
def _get_causal_paths(self, candidates: list[str]) -> list[str]:
"""获取因果路径"""
paths = []
for i, src in enumerate(candidates):
for dst in candidates[i+1:]:
try:
path = nx.shortest_path(self.causal_graph, src, dst)
paths.append(" → ".join(path))
except nx.NetworkXNoPath:
pass
return paths
def _generate_actions(self, root_causes: list[dict]) -> list[str]:
"""生成修复动作"""
actions = []
for rc in root_causes:
service = rc["service"]
if "node" in service:
actions.append(f"检查节点 {service} 的资源使用情况，考虑重启或迁移")
elif "database" in service:
actions.append(f"检查数据库 {service} 的慢查询和连接数，考虑扩容")
elif "cache" in service:
actions.append(f"检查缓存 {service} 的命中率，考虑清理或扩容")
else:
actions.append(f"检查服务 {service} 的日志和指标，考虑重启或回滚")
return actions
# 使用示例
if __name__ == "__main__":
analyzer = AIRootCauseAnalyzer()
alert_group = {
"alerts": [
{"name": "HighCPU", "labels": {"service": "api-gateway", "node": "node-1"}},
{"name": "HighMemory", "labels": {"service": "api-gateway", "node": "node-1"}},
{"name": "PodCrashLooping", "labels": {"service": "api-gateway"}},
]
}
metrics = {
"node-1": {"cpu_usage": 95, "memory_usage": 88, "error_rate": 5.2, "p99_latency_ms": 3200},
"api-gateway": {"cpu_usage": 85, "memory_usage": 72, "error_rate": 8.1, "p99_latency_ms": 4500},
"redis-cache": {"cpu_usage": 30, "memory_usage": 45, "error_rate": 0.1, "p99_latency_ms": 5},
}
report = analyzer.analyze(alert_group, metrics)
print(f"Root Cause: {report.root_causes}")
print(f"Actions: {report.recommended_actions}")
模式三：自動化修復Runbook
# runbook_executor.py
# 依赖：pip install kubernetes pydantic
from pydantic import BaseModel, Field
from typing import Optional, Callable
from enum import Enum
from datetime import datetime
import subprocess
import logging
import time
logger = logging.getLogger(__name__)
class RunbookStatus(Enum):
PENDING = "pending"
RUNNING = "running"
SUCCESS = "success"
FAILED = "failed"
ROLLED_BACK = "rolled_back"
class RunbookStep(BaseModel):
"""Runbook步骤"""
name: str
action: str
params: dict = Field(default_factory=dict)
timeout: int = 300 # 秒
rollback_action: Optional[str] = None
rollback_params: dict = Field(default_factory=dict)
class RunbookExecution(BaseModel):
"""Runbook执行记录"""
runbook_id: str
incident_id: str
steps: list[RunbookStep]
status: RunbookStatus = RunbookStatus.PENDING
current_step: int = 0
started_at: Optional[datetime] = None
completed_at: Optional[datetime] = None
results: list[dict] = Field(default_factory=list)
error: Optional[str] = None
class RunbookExecutor:
"""Runbook执行器 - 安全的自动化修复"""
def __init__(self, dry_run: bool = False):
self.dry_run = dry_run
self.action_registry: dict[str, Callable] = {}
self._register_default_actions()
def _register_default_actions(self):
"""注册默认修复动作"""
self.action_registry = {
"restart_pod": self._restart_pod,
"scale_deployment": self._scale_deployment,
"clean_disk": self._clean_disk,
"restart_service": self._restart_service,
"rollback_deployment": self._rollback_deployment,
"clear_cache": self._clear_cache,
"check_health": self._check_health,
"notify_slack": self._notify_slack,
}
def execute(self, execution: RunbookExecution) -> RunbookExecution:
"""执行Runbook"""
execution.status = RunbookStatus.RUNNING
execution.started_at = datetime.now()
completed_steps = []
for i, step in enumerate(execution.steps):
execution.current_step = i
logger.info(f"Executing step {i+1}/{len(execution.steps)}: {step.name}")
try:
result = self._execute_step(step)
execution.results.append(result)
completed_steps.append((i, step))
if result.get("status") == "failed":
raise Exception(result.get("error", "Step failed"))
except Exception as e:
logger.error(f"Step {step.name} failed: {e}")
execution.status = RunbookStatus.FAILED
execution.error = str(e)
# 自动回滚已完成的步骤
self._rollback_steps(completed_steps, execution)
return execution
execution.status = RunbookStatus.SUCCESS
execution.completed_at = datetime.now()
return execution
def _execute_step(self, step: RunbookStep) -> dict:
"""执行单个步骤"""
action_fn = self.action_registry.get(step.action)
if not action_fn:
return {"status": "failed", "error": f"Unknown action: {step.action}"}
if self.dry_run:
return {
"status": "dry_run",
"action": step.action,
"params": step.params,
"message": f"[DRY RUN] Would execute: {step.action}",
}
try:
result = action_fn(**step.params)
return {"status": "success", "action": step.action, "result": result}
except Exception as e:
return {"status": "failed", "action": step.action, "error": str(e)}
def _rollback_steps(
self,
completed_steps: list[tuple[int, RunbookStep]],
execution: RunbookExecution,
):
"""回滚已完成的步骤"""
for i, step in reversed(completed_steps):
if step.rollback_action:
logger.info(f"Rolling back step {i+1}: {step.name}")
rollback_fn = self.action_registry.get(step.rollback_action)
if rollback_fn:
try:
rollback_fn(**step.rollback_params)
except Exception as e:
logger.error(f"Rollback failed for step {step.name}: {e}")
execution.status = RunbookStatus.ROLLED_BACK
# ===== 内置修复动作 =====
def _restart_pod(self, namespace: str, deployment: str, **kwargs) -> dict:
"""重启Pod"""
cmd = f"kubectl rollout restart deployment/{deployment} -n {namespace}"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60)
return {"stdout": result.stdout, "stderr": result.stderr, "returncode": result.returncode}
def _scale_deployment(self, namespace: str, deployment: str, replicas: int, **kwargs) -> dict:
"""扩缩容"""
cmd = f"kubectl scale deployment/{deployment} -n {namespace} --replicas={replicas}"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60)
return {"stdout": result.stdout, "replicas": replicas}
def _clean_disk(self, node: str, threshold: int = 80, **kwargs) -> dict:
"""清理磁盘"""
# 清理Docker镜像和退出容器
cmd = f"ssh {node} 'docker system prune -af --volumes && docker image prune -af'"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=300)
return {"stdout": result.stdout, "node": node}
def _restart_service(self, service: str, **kwargs) -> dict:
"""重启系统服务"""
cmd = f"systemctl restart {service}"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=60)
return {"stdout": result.stdout, "service": service}
def _rollback_deployment(self, namespace: str, deployment: str, revision: int = 0, **kwargs) -> dict:
"""回滚部署"""
revision_flag = f"--to-revision={revision}" if revision else ""
cmd = f"kubectl rollout undo deployment/{deployment} -n {namespace} {revision_flag}"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=120)
return {"stdout": result.stdout}
def _clear_cache(self, namespace: str, deployment: str, **kwargs) -> dict:
"""清理应用缓存"""
cmd = f"kubectl exec -n {namespace} deployment/{deployment} -- curl -s -X POST http://localhost:8080/cache/clear"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
return {"stdout": result.stdout}
def _check_health(self, namespace: str, deployment: str, **kwargs) -> dict:
"""健康检查"""
cmd = f"kubectl get pods -n {namespace} -l app={deployment} -o jsonpath='{{.items[*].status.phase}}'"
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=30)
return {"stdout": result.stdout, "phases": result.stdout.strip().split()}
def _notify_slack(self, channel: str, message: str, **kwargs) -> dict:
"""发送Slack通知"""
# 简化实现
logger.info(f"[Slack] #{channel}: {message}")
return {"channel": channel, "message": message}
# 使用示例
if __name__ == "__main__":
executor = RunbookExecutor(dry_run=True)
# 定义Runbook
execution = RunbookExecution(
runbook_id="RB-001",
incident_id="INC-20260621001",
steps=[
RunbookStep(
name="检查服务健康状态",
action="check_health",
params={"namespace": "prod", "deployment": "api-gateway"},
),
RunbookStep(
name="重启Pod",
action="restart_pod",
params={"namespace": "prod", "deployment": "api-gateway"},
rollback_action="rollback_deployment",
rollback_params={"namespace": "prod", "deployment": "api-gateway"},
),
RunbookStep(
name="扩容实例",
action="scale_deployment",
params={"namespace": "prod", "deployment": "api-gateway", "replicas": 5},
rollback_action="scale_deployment",
rollback_params={"namespace": "prod", "deployment": "api-gateway", "replicas": 3},
),
RunbookStep(
name="通知运维团队",
action="notify_slack",
params={"channel": "sre-alerts", "message": "api-gateway已自动修复"},
),
],
)
result = executor.execute(execution)
print(f"Status: {result.status.value}")
for r in result.results:
print(f" - {r}")
模式四：自愈系統設計
# self_healing_system.py
# 依赖：pip install fastapi kubernetes prometheus-client
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
import logging
import asyncio
import uuid
logger = logging.getLogger(__name__)
app = FastAPI(title="Self-Healing System", version="1.0.0")
class HealingPolicy(BaseModel):
"""自愈策略"""
id: str
name: str
description: str
trigger_condition: dict # 触发条件
runbook_id: str # 关联的Runbook
cooldown_minutes: int = 10 # 冷却时间
max_retries: int = 3 # 最大重试次数
require_approval: bool = False # 是否需要人工审批
severity_threshold: str = "critical" # 最低触发严重级别
class HealingRecord(BaseModel):
"""自愈记录"""
id: str
policy_id: str
incident_id: str
status: str # triggered, healing, healed, failed
triggered_at: datetime
healed_at: Optional[datetime] = None
actions_taken: list[str] = []
error: Optional[str] = None
class SelfHealingEngine:
"""自愈引擎 - 自动检测、决策、修复"""
def __init__(self):
self.policies: dict[str, HealingPolicy] = {}
self.records: list[HealingRecord] = []
self.last_healing: dict[str, datetime] = {} # 冷却控制
self.retry_counts: dict[str, int] = defaultdict(int)
# 注册默认自愈策略
self._register_default_policies()
def _register_default_policies(self):
"""注册默认自愈策略"""
default_policies = [
HealingPolicy(
id="hp-pod-crash",
name="Pod崩溃自愈",
description="Pod CrashLoopBackOff时自动重启",
trigger_condition={"alert": "PodCrashLooping", "duration": "5m"},
runbook_id="RB-POD-CRASH",
cooldown_minutes=10,
max_retries=3,
),
HealingPolicy(
id="hp-high-cpu",
name="高CPU自愈",
description="CPU持续高负载时自动扩容",
trigger_condition={"alert": "HighCPU", "threshold": "90%", "duration": "10m"},
runbook_id="RB-HIGH-CPU",
cooldown_minutes=30,
max_retries=2,
),
HealingPolicy(
id="hp-disk-full",
name="磁盘满自愈",
description="磁盘使用率超过阈值时自动清理",
trigger_condition={"alert": "DiskFull", "threshold": "85%"},
runbook_id="RB-DISK-FULL",
cooldown_minutes=60,
max_retries=1,
),
HealingPolicy(
id="hp-oom-killed",
name="OOM自愈",
description="容器OOM Killed时调整资源限制",
trigger_condition={"alert": "OOMKilled"},
runbook_id="RB-OOM",
cooldown_minutes=15,
max_retries=2,
require_approval=True, # OOM调整需要审批
),
]
for policy in default_policies:
self.policies[policy.id] = policy
async def evaluate_and_heal(self, alert_group: dict) -> Optional[HealingRecord]:
"""评估告警并触发自愈"""
# 1. 匹配自愈策略
matched_policy = self._match_policy(alert_group)
if not matched_policy:
logger.info("No matching healing policy found")
return None
# 2. 检查冷却时间
if not self._check_cooldown(matched_policy.id):
logger.info(f"Policy {matched_policy.id} is in cooldown")
return None
# 3. 检查重试次数
if self.retry_counts[matched_policy.id] >= matched_policy.max_retries:
logger.warning(f"Policy {matched_policy.id} exceeded max retries")
return None
# 4. 检查是否需要审批
if matched_policy.require_approval:
logger.info(f"Policy {matched_policy.id} requires manual approval")
# 发送审批请求，等待人工确认
return None
# 5. 执行自愈
record = HealingRecord(
id=str(uuid.uuid4())[:8],
policy_id=matched_policy.id,
incident_id=alert_group.get("incident_id", "unknown"),
status="triggered",
triggered_at=datetime.now(),
)
try:
record.status = "healing"
result = await self._execute_healing(matched_policy, alert_group)
record.actions_taken = result.get("actions", [])
record.status = "healed"
record.healed_at = datetime.now()
self.retry_counts[matched_policy.id] = 0 # 重置重试计数
except Exception as e:
record.status = "failed"
record.error = str(e)
self.retry_counts[matched_policy.id] += 1
self.records.append(record)
self.last_healing[matched_policy.id] = datetime.now()
return record
def _match_policy(self, alert_group: dict) -> Optional[HealingPolicy]:
"""匹配自愈策略"""
alerts = alert_group.get("alerts", [])
for alert in alerts:
alert_name = alert.get("name", "")
for policy in self.policies.values():
if policy.trigger_condition.get("alert") == alert_name:
return policy
return None
def _check_cooldown(self, policy_id: str) -> bool:
"""检查冷却时间"""
if policy_id not in self.last_healing:
return True
policy = self.policies[policy_id]
elapsed = (datetime.now() - self.last_healing[policy_id]).total_seconds() / 60
return elapsed >= policy.cooldown_minutes
async def _execute_healing(self, policy: HealingPolicy, alert_group: dict) -> dict:
"""执行自愈动作"""
from runbook_executor import RunbookExecutor, RunbookExecution, RunbookStep
executor = RunbookExecutor(dry_run=False)
# 根据策略类型构建Runbook步骤
steps = self._build_healing_steps(policy, alert_group)
execution = RunbookExecution(
runbook_id=policy.runbook_id,
incident_id=alert_group.get("incident_id", "unknown"),
steps=steps,
)
result = executor.execute(execution)
return {
"actions": [r.get("action", "") for r in result.results],
"status": result.status.value,
}
def _build_healing_steps(self, policy: HealingPolicy, alert_group: dict) -> list:
"""构建自愈步骤"""
steps_map = {
"hp-pod-crash": [
RunbookStep(name="重启Pod", action="restart_pod",
params={"namespace": "prod", "deployment": "api-gateway"}),
RunbookStep(name="健康检查", action="check_health",
params={"namespace": "prod", "deployment": "api-gateway"}),
],
"hp-high-cpu": [
RunbookStep(name="扩容实例", action="scale_deployment",
params={"namespace": "prod", "deployment": "api-gateway", "replicas": 5}),
RunbookStep(name="通知团队", action="notify_slack",
params={"channel": "sre-alerts", "message": "api-gateway已自动扩容至5副本"}),
],
"hp-disk-full": [
RunbookStep(name="清理磁盘", action="clean_disk",
params={"node": "node-1"}),
],
}
return steps_map.get(policy.id, [])
from collections import defaultdict
# 全局自愈引擎
healing_engine = SelfHealingEngine()
@app.post("/heal")
async def trigger_healing(alert_group: dict, background_tasks: BackgroundTasks):
"""触发自愈流程"""
background_tasks.add_task(healing_engine.evaluate_and_heal, alert_group)
return {"status": "healing_triggered"}
@app.get("/healing/records")
async def get_healing_records():
"""获取自愈记录"""
return {"records": [r.model_dump() for r in healing_engine.records]}
@app.get("/healing/policies")
async def get_healing_policies():
"""获取自愈策略"""
return {"policies": [p.model_dump() for p in healing_engine.policies.values()]}
模式五：生產級AIOps平臺
# aiops_platform.py - 生产级AIOps平台集成
# 依赖：pip install fastapi prometheus-client redis
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
import asyncio
import json
import redis
import logging
logger = logging.getLogger(__name__)
app = FastAPI(title="AIOps Platform", version="2.0.0")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# Redis用于实时数据流
redis_client = redis.Redis(host="localhost", port=6379, db=0)
class PlatformMetrics(BaseModel):
"""平台指标"""
active_incidents: int = 0
alerts_last_hour: int = 0
alerts_suppressed: int = 0
auto_healed: int = 0
mttr_minutes: float = 0.0
mtta_minutes: float = 0.0 # 平均确认时间
class ConnectionManager:
"""WebSocket连接管理器"""
def __init__(self):
self.active_connections: list[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def broadcast(self, message: dict):
for connection in self.active_connections:
try:
await connection.send_json(message)
except Exception:
self.active_connections.remove(connection)
manager = ConnectionManager()
@app.websocket("/ws/alerts")
async def websocket_alerts(websocket: WebSocket):
"""实时告警推送"""
await manager.connect(websocket)
try:
while True:
# 从Redis消费告警流
alert_data = redis_client.xread({"alert_stream": "$"}, count=1, block=1000)
if alert_data:
for stream, messages in alert_data:
for msg_id, msg_data in messages:
await websocket.send_json({
"type": "alert",
"data": msg_data,
"timestamp": datetime.now().isoformat(),
})
except WebSocketDisconnect:
manager.disconnect(websocket)
@app.get("/api/dashboard")
async def get_dashboard():
"""获取仪表盘数据"""
return PlatformMetrics(
active_incidents=int(redis_client.get("active_incidents") or 0),
alerts_last_hour=int(redis_client.get("alerts_last_hour") or 0),
alerts_suppressed=int(redis_client.get("alerts_suppressed") or 0),
auto_healed=int(redis_client.get("auto_healed") or 0),
mttr_minutes=float(redis_client.get("mttr_minutes") or 0),
mtta_minutes=float(redis_client.get("mtta_minutes") or 0),
)
@app.get("/api/incidents")
async def get_incidents(status: Optional[str] = None, limit: int = 20):
"""获取事件列表"""
incidents = []
keys = redis_client.keys("incident:*")
for key in keys[:limit]:
data = redis_client.hgetall(key)
if status and data.get(b"status", b"").decode() != status:
continue
incidents.append({k.decode(): v.decode() for k, v in data.items()})
return {"incidents": incidents}
@app.post("/api/incidents/{incident_id}/acknowledge")
async def acknowledge_incident(incident_id: str, ack_info: dict):
"""确认事件"""
redis_client.hset(
f"incident:{incident_id}",
mapping={
"status": "acknowledged",
"acknowledged_by": ack_info.get("user", "unknown"),
"acknowledged_at": datetime.now().isoformat(),
}
)
await manager.broadcast({
"type": "incident_acknowledged",
"incident_id": incident_id,
})
return {"status": "acknowledged"}
避坑指南
坑1：告警降噪過度導致漏報
# ❌ 错误：去重窗口过长，真实告警被抑制
dedup_window = timedelta(hours=1) # 太长！
# ✅ 正确：根据严重级别设置不同窗口
dedup_windows = {
"critical": timedelta(minutes=1), # Critical只去重1分钟
"warning": timedelta(minutes=10), # Warning去重10分钟
"info": timedelta(minutes=30), # Info去重30分钟
}
坑2：LLM根因分析的幻覺
# ❌ 错误：完全依赖LLM输出，不验证
root_cause = llm.analyze(alerts) # 可能产生幻觉
# ✅ 正确：LLM输出+规则验证
llm_result = llm.analyze(alerts)
if llm_result.root_cause not in known_services:
logger.warning("LLM suggested unknown root cause, falling back to rules")
llm_result = rule_based_analyzer.analyze(alerts)
坑3：Runbook執行缺乏安全邊界
# ❌ 错误：无限制的Runbook执行
executor.execute(runbook) # 可能无限重试或执行危险操作
# ✅ 正确：添加安全边界
class SafeRunbookExecutor(RunbookExecutor):
DANGEROUS_ACTIONS = {"delete_pod", "drop_database", "format_disk"}
def _execute_step(self, step):
if step.action in self.DANGEROUS_ACTIONS:
raise Exception(f"Dangerous action blocked: {step.action}")
return super()._execute_step(step)
坑4：自愈系統形成正反饋循環
# ❌ 错误：自愈动作触发新的告警，再次触发自愈
# 扩容→资源不足告警→再扩容→...
# ✅ 正确：冷却时间 + 全局自愈频率限制
MAX_HEALING_PER_HOUR = 5 # 每小时最多自愈5次
def check_global_rate():
count = redis_client.incr("healing_counter")
if count == 1:
redis_client.expire("healing_counter", 3600)
return count <= MAX_HEALING_PER_HOUR
坑5：忽略自愈操作的審計追蹤
# ❌ 错误：自愈操作没有审计日志
executor.execute(runbook) # 谁触发的？什么时候？结果如何？
# ✅ 正确：所有自愈操作记录审计日志
import structlog
audit_logger = structlog.get_logger("audit")
def audited_execute(runbook):
audit_logger.info("healing_started", runbook_id=runbook.runbook_id, trigger="auto")
result = executor.execute(runbook)
audit_logger.info("healing_completed", status=result.status, actions=result.results)
return result
報錯排查表
| 報錯信息 | 原因 | 解決方案 |
|---|---|---|
Connection refused to Prometheus |
Prometheus未啟動或地址錯誤 | 檢查Prometheus服務狀態和URL配置 |
Kubernetes API timeout |
kubectl配置錯誤或集群不可達 | 檢查kubeconfig和集群連接 |
LLM API rate limit exceeded |
請求頻率超限 | 添加請求限流或使用本地模型 |
Redis connection lost |
Redis服務異常 | 檢查Redis連接配置，添加重連邏輯 |
Runbook step timeout |
修復操作耗時過長 | 增加timeout或優化修復腳本 |
Alert dedup window too aggressive |
去重窗口過短 | 根據告警類型調整dedup_window |
Self-healing loop detected |
自愈操作觸發新告警 | 增加冷卻時間和全局頻率限制 |
DBSCAN clustering failed |
告警數據量不足 | 降低min_samples或使用其他聚類算法 |
WebSocket connection dropped |
客戶端斷開 | 添加心跳和自動重連機制 |
Permission denied on kubectl |
ServiceAccount權限不足 | 添加RBAC權限或使用專用運維賬號 |
進階優化
1. 基於時序異常檢測的預測性告警
from sklearn.ensemble import IsolationForest
import numpy as np
class PredictiveAlerter:
"""预测性告警 - 在故障发生前预警"""
def __init__(self):
self.models: dict[str, IsolationForest] = {}
def train(self, service: str, metrics: np.ndarray):
model = IsolationForest(contamination=0.05, random_state=42)
model.fit(metrics)
self.models[service] = model
def predict(self, service: str, current_metrics: np.ndarray) -> bool:
if service not in self.models:
return False
prediction = self.models[service].predict(current_metrics.reshape(1, -1))
return prediction[0] == -1 # -1表示异常
2. 多集群聯邦自愈
class FederatedHealingEngine:
"""多集群联邦自愈"""
def __init__(self, clusters: dict[str, str]):
self.clusters = clusters # {cluster_name: api_endpoint}
async def heal_globally(self, alert_group: dict):
affected_clusters = self._identify_affected_clusters(alert_group)
tasks = [
self._heal_cluster(cluster, alert_group)
for cluster in affected_clusters
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
3. ChatOps集成
@app.post("/slack/interactive")
async def slack_interactive(payload: dict):
"""Slack交互式自愈确认"""
action = payload["actions"][0]
if action["action_id"] == "approve_healing":
healing_engine.approve_healing(action["value"])
elif action["action_id"] == "reject_healing":
healing_engine.reject_healing(action["value"])
return {"status": "ok"}
4. 混沌工程驗證
class ChaosValidator:
"""混沌工程验证自愈能力"""
async def inject_fault(self, fault_type: str, target: str):
"""注入故障"""
if fault_type == "pod_kill":
cmd = f"kubectl delete pod -l app={target} -n prod --grace-period=0"
elif fault_type == "cpu_stress":
cmd = f"kubectl run cpu-stress --image=progrium/stress -- --cpu 4 --timeout 60s"
# 监控自愈是否在预期时间内完成
start = time.time()
# ... 等待自愈 ...
mttr = time.time() - start
return {"fault_type": fault_type, "mttr_seconds": mttr, "healed": mttr < 300}
5. SLO驅動的自愈策略
class SLODrivenHealing:
"""基于SLO的自愈策略"""
def __init__(self, slo_target: float = 99.9):
self.slo_target = slo_target
self.error_budget = self._calculate_error_budget()
def should_trigger_healing(self, current_slo: float) -> bool:
"""当SLO接近阈值时触发自愈"""
error_budget_remaining = (current_slo - (100 - self.slo_target)) / self.slo_target
return error_budget_remaining < 0.2 # 错误预算剩余不足20%时触发
def _calculate_error_budget(self) -> float:
"""计算30天错误预算（分钟）"""
total_minutes = 30 * 24 * 60
allowed_downtime = total_minutes * (1 - self.slo_target / 100)
return allowed_downtime
對比分析
| 特性 | 傳統SRE | AI輔助SRE | 全自動自愈 |
|---|---|---|---|
| 告警處理 | 人工篩選 | AI降噪+聚合 | 自動降噪+聚合 |
| 根因定位 | 人工排查 | AI輔助推理 | 自動根因分析 |
| 故障修復 | 手動執行 | Runbook輔助 | 自動執行+回滾 |
| MTTR | 30-120分鐘 | 10-30分鐘 | 1-5分鐘 |
| 誤操作風險 | 中 | 低 | 需安全邊界 |
| 7x24覆蓋 | 需值班 | 需值班 | ✅ 全自動 |
| 實施複雜度 | 低 | 中 | 高 |
| 適用階段 | 初創期 | 成長期 | 成熟期 |
總結
AI驅動的SRE正在從"輔助工具"進化為"自主系統"：
- 告警降噪：去重+聚合+聚類三管齊下，將200條告警壓縮為5個有意義的告警組
- 根因分析：知識圖譜+LLM推理，將根因定位時間從30分鐘縮短到3分鐘
- Runbook自動化：標準化修復流程，支持回滾和審計
- 自愈系統：冷卻+限流+審批三重安全邊界，實現安全的自動修復
- AIOps平臺：實時儀表盤+WebSocket推送+ChatOps集成
選擇建議：從告警降噪開始，逐步引入AI根因分析，最後實現自愈系統。不要一步到位——自愈系統需要充分的安全邊界和審計機制。
在線工具推薦
- /zh-TW/json/format - JSON格式化工具，調試告警和API數據
- /zh-TW/dev/curl-to-code - cURL轉代碼，快速生成Kubernetes API調用
- /zh-TW/encode/hash - 哈希計算工具，生成告警指紋
- /zh-TW/text/diff - 文本對比工具，對比Runbook版本差異
本站提供瀏覽器本地工具，免註冊即可試用 →
#AI SRE#智能告警#根因分析#自愈系统#AIOps#2026#DevOps运维