SRE piloté par l'IA : Guide complet d'analyse de cause racine intelligente des alertes et de systèmes d'auto-réparation 2026
DevOps
SRE piloté par l'IA : Guide complet d'analyse de cause racine intelligente des alertes et de systèmes d'auto-réparation 2026
À 3 heures du matin, votre téléphone est bombardé de 200 alertes, mais seulement 1 d'entre elles nécessite réellement votre attention. C'est le cauchemar quotidien des ingénieurs SRE. Les systèmes d'alerte traditionnels basés sur des seuils statiques ne peuvent pas distinguer les symptômes des causes racines, et encore moins s'auto-remédier. En 2026, l'IA refaçonne le SRE : de la réduction du bruit des alertes à la localisation de la cause racine, de la remédiation automatisée aux systèmes d'auto-réparation, l'AIOps est passé du concept à la production.
Concepts clés en un coup d'œil
| Concept | Description | Cas d'utilisation |
|---|---|---|
| Réduction du bruit des alertes | Fusionner/supprimer/réduire les alertes redondantes | Tempêtes d'alertes |
| Agrégation d'alertes | Corréler et fusionner plusieurs alertes | Localisation de la cause racine |
| Analyse de Cause Racine (RCA) | Identifier la cause racine des défaillances | Investigation d'incidents |
| Runbook | Workflow de remédiation d'incident standardisé | Remédiation automatisée |
| Système d'auto-réparation | Auto-remédiation sans intervention humaine | Opérations 24/7 |
| AIOps | Opérations IT pilotées par l'IA | Opérations intelligentes |
| Détection d'anomalies | Identification d'anomalies basée sur le ML | Découverte proactive |
| Chaos Engineering | Injection de pannes proactive pour la validation de résilience | Vérification de fiabilité |
Cinq points de douleur clés
- Tempêtes d'alertes : Dans les architectures de microservices, une défaillance de service déclenche des alertes en cascade, générant des centaines d'alertes en 5 minutes, noyant la cause racine réelle
- Localisation de la cause racine difficile : En moyenne, 70 % du MTTR (Temps Moyen de Récupération) est consacré à localiser la cause racine plutôt qu'à la corriger
- Remédiation répétitive : 80 % des tâches opérationnelles sont répétitives (redémarrage de services, mise à l'échelle, nettoyage de disques), nécessitant pourtant une intervention manuelle à chaque fois
- Lacunes de connaissances : Lorsque les ingénieurs seniors partent, l'expérience de gestion des incidents est perdue, et les nouveaux membres de l'équipe ne peuvent pas être intégrés rapidement
- Réponse réactive : La surveillance traditionnelle ne peut alerter qu'après la survenue de défaillances, incapable de les prédire ou de les prévenir
Étape par étape : 5 modèles fondamentaux
Modèle 1 : Réduction du bruit des alertes et agrégation
Environnement d'exécution : 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()}")
Modèle 2 : Analyse de cause racine par IA
Analyse de cause racine basée sur les graphes de connaissances et l'inférence causale :
# 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}")
Modèle 3 : Runbook de remédiation automatisée
# 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}")
Modèle 4 : Conception de système d'auto-réparation
# 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()]}
Modèle 5 : Plateforme AIOps de niveau production
# 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"}
Guide des pièges
Piège 1 : Réduction du bruit des alertes trop agressive entraînant des alertes manquées
# ❌ 错误：去重窗口过长，真实告警被抑制
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分钟
}
Piège 2 : Hallucinations de l'analyse de cause racine par 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)
Piège 3 : Exécution de Runbook sans limites de sécurité
# ❌ 错误：无限制的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)
Piège 4 : Le système d'auto-réparation crée des boucles de rétroaction positive
# ❌ 错误：自愈动作触发新的告警，再次触发自愈
# 扩容→资源不足告警→再扩容→...
# ✅ 正确：冷却时间 + 全局自愈频率限制
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
Piège 5 : Ignorer les pistes d'audit pour les opérations d'auto-réparation
# ❌ 错误：自愈操作没有审计日志
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
Tableau de dépannage des erreurs
| Message d'erreur | Cause | Solution |
|---|---|---|
Connection refused to Prometheus |
Prometheus non démarré ou adresse incorrecte | Vérifier le statut du service Prometheus et la configuration URL |
Kubernetes API timeout |
kubectl mal configuré ou cluster inaccessible | Vérifier kubeconfig et la connectivité du cluster |
LLM API rate limit exceeded |
Taux de requêtes dépassé | Ajouter une limitation de requêtes ou utiliser des modèles locaux |
Redis connection lost |
Service Redis anormal | Vérifier la configuration de connexion Redis, ajouter une logique de reconnexion |
Runbook step timeout |
Opération de remédiation trop longue | Augmenter le délai d'attente ou optimiser les scripts de remédiation |
Alert dedup window too aggressive |
Fenêtre de déduplication trop courte | Ajuster dedup_window en fonction du type d'alerte |
Self-healing loop detected |
L'auto-réparation déclenche de nouvelles alertes | Ajouter un refroidissement et une limitation de débit globale |
DBSCAN clustering failed |
Données d'alerte insuffisantes | Réduire min_samples ou utiliser d'autres algorithmes de clustering |
WebSocket connection dropped |
Client déconnecté | Ajouter un mécanisme de battement de cœur et de reconnexion automatique |
Permission denied on kubectl |
Permissions ServiceAccount insuffisantes | Ajouter des permissions RBAC ou utiliser un compte ops dédié |
Optimisations avancées
1. Alertes prédictives basées sur la détection d'anomalies de séries temporelles
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. Auto-réparation fédérée multi-cluster
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. Intégration 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. Validation par Chaos Engineering
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. Stratégie d'auto-réparation pilotée par les 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
Analyse comparative
| Fonctionnalité | SRE Traditionnel | SRE Assisté par IA | Auto-Réparation Entièrement Automatisée |
|---|---|---|---|
| Gestion des alertes | Filtrage manuel | Réduction du bruit IA + agrégation | Réduction du bruit auto + agrégation |
| Localisation de la cause racine | Investigation manuelle | Raisonnement assisté par IA | Analyse de cause racine automatique |
| Remédiation d'incidents | Exécution manuelle | Assistée par Runbook | Exécution auto + retour en arrière |
| MTTR | 30-120 min | 10-30 min | 1-5 min |
| Risque de mauvaise opération | Moyen | Faible | Nécessite des limites de sécurité |
| Couverture 24/7 | Nécessite une astreinte | Nécessite une astreinte | Entièrement automatisée |
| Complexité de mise en œuvre | Faible | Moyenne | Élevée |
| Stade applicable | Démarrage | Croissance | Maturité |
Résumé
Le SRE piloté par l'IA évolue d'un « outil d'assistance » vers un « système autonome » :
- Réduction du bruit des alertes : Déduplication + agrégation + clustering fonctionnant ensemble, compressant 200 alertes en 5 groupes d'alertes significatifs
- Analyse de cause racine : Graphe de connaissances + raisonnement LLM, réduisant le temps de localisation de la cause racine de 30 minutes à 3 minutes
- Automatisation des Runbooks : Workflows de remédiation standardisés avec support de retour en arrière et d'audit
- Système d'auto-réparation : Refroidissement + limitation de débit + approbation comme triple limite de sécurité pour une auto-remédiation sécurisée
- Plateforme AIOps : Tableau de bord en temps réel + push WebSocket + intégration ChatOps
Recommandation : Commencez par la réduction du bruit des alertes, introduisez progressivement l'analyse de cause racine par IA, et enfin mettez en œuvre les systèmes d'auto-réparation. Ne vous lancez pas à fond d'emblée — les systèmes d'auto-réparation nécessitent des limites de sécurité adéquates et des mécanismes d'audit.
Outils en ligne recommandés
- /fr/json/format - Formateur JSON pour le débogage des alertes et des données API
- /fr/dev/curl-to-code - Convertisseur cURL en code pour générer rapidement des appels API Kubernetes
- /fr/encode/hash - Calculateur de hachage pour générer des empreintes d'alertes
- /fr/text/diff - Outil de comparaison de texte pour comparer les différences de versions de Runbook
Essayez ces outils exécutés localement dans le navigateur — aucune inscription requise →
#AI SRE#智能告警#根因分析#自愈系统#AIOps#2026#DevOps运维