AI駆動SRE：インテリジェントアラート根本原因分析と自己修復システム完全ガイド 2026
DevOps运维
AI駆動SRE：インテリジェントアラート根本原因分析と自己修復システム完全ガイド 2026
午前3時、あなたのスマホに200件のアラートが殺到しても、実際に対応すべきは1件だけかもしれません。これはSREエンジニアの日常的な悪夢です。従来のアラートシステムは静的しきい値に基づいており、症状と根本原因を区別できず、自動修復はなおさら不可能です。2026年、AIはSREを再構築しています：アラートノイズ削減から根本原因特定まで、自動修復から自己修復システムまで、AIOpsは概念から本番環境へと移行しました。
主要概念早見表
| 概念 | 説明 | 適用シナリオ |
|---|---|---|
| アラートノイズ削減 | 冗長アラートの統合/抑制/ミュート | アラートストーム |
| アラート集約 | 関連性に基づき複数アラートを統合 | 根本原因特定 |
| 根本原因分析(RCA) | 障害の根本原因を特定 | 障害調査 |
| Runbook | 標準化された障害修復ワークフロー | 自動修復 |
| 自己修復システム | 人の介入なしに自動修復 | 24時間運用 |
| AIOps | AI駆動のIT運用 | インテリジェント運用 |
| 異常検知 | MLベースの異常識別 | 能動的発見 |
| カオスエンジニアリング | 能動的障害注入でレジリエンスを検証 | 信頼性検証 |
5つの主要ペインポイント
- アラートストーム：マイクロサービスアーキテクチャでは、1つのサービス障害がカスケードアラートを引き起こし、5分以内に数百件のアラートが生成され、真の根本原因が埋もれる
- 根本原因特定の困難さ：平均MTTR（平均復旧時間）のうち、70%の時間が根本原因の特定に費やされ、修復そのものではない
- 反復的な修復作業：運用タスクの80%は反復的（サービス再起動、スケールアウト、ディスククリーンアップ）だが、毎回手動介入が必要
- 知識の断絶：シニアエンジニアの退職後、障害対応の経験が失われ、新人が迅速にキャッチアップできない
- 受動的対応：従来の監視は障害発生後にのみアラートを出せ、予測や予防ができない
ステップバイステップ：5つのコアパターン
パターン1：アラートノイズ削減と集約
実行環境：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()}")
パターン2：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}")
パターン3：自動修復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}")
パターン4：自己修復システム設計
# 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()]}
パターン5：本番級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分 |
| 誤操作リスク | 中 | 低 | 安全境界が必要 |
| 24時間対応 | オンコール必要 | オンコール必要 | 完全自動 |
| 実装複雑さ | 低 | 中 | 高 |
| 適用段階 | スタートアップ | 成長期 | 成熟期 |
まとめ
AI駆動SREは「支援ツール」から「自律システム」へと進化しています：
- アラートノイズ削減：重複排除+集約+クラスタリングの三本柱で、200件のアラートを5つの意味のあるアラートグループに圧縮
- 根本原因分析：ナレッジグラフ+LLM推論で、根本原因特定時間を30分から3分に短縮
- Runbook自動化：標準化された修復ワークフロー、ロールバックと監査をサポート
- 自己修復システム：クールダウン+レート制限+承認の三重安全境界で、安全な自動修復を実現
- AIOpsプラットフォーム：リアルタイムダッシュボード+WebSocketプッシュ+ChatOps統合
推奨事項：アラートノイズ削減から始め、段階的にAI根本原因分析を導入し、最後に自己修復システムを実装してください。一度に全てを導入しないでください——自己修復システムには十分な安全境界と監査メカニズムが必要です。
オンラインツール推奨
- /ja/json/format - JSONフォーマッター、アラートとAPIデータのデバッグに
- /ja/dev/curl-to-code - cURLからコード変換、Kubernetes API呼び出しを素早く生成
- /ja/encode/hash - ハッシュ計算ツール、アラートフィンガープリントの生成に
- /ja/text/diff - テキスト差分ツール、Runbookバージョン差異の比較に
ブラウザローカルツールを無料で試す →
#AI SRE#智能告警#根因分析#自愈系统#AIOps#2026#DevOps运维