智能告警降噪：先合并事件，再通知人-拓冰建站

智能告警降噪：先合并事件，再通知人

一、告警噪声的本质是"同一件事件被重复通知"

很多团队的告警系统，在出现一个故障时，会同时触发几十条告警：数据库慢查询、应用延迟升高、HTTP 500 错误增多、消息队列堆积、用户体验下降……这些告警其实都是同一件事件的不同表现，但告警系统不知道它们是相关的，所以会逐条通知值班人员。

告警降噪的第一原则：能合并的告警必须合并，人只应该收到"事件"而不是"信号"。

我们在生产中要求：同一个服务在 10 分钟内的所有告警，必须合并成一条通知，附带完整的上下文（影响了什么、可能的原因、建议的处理步骤）。

二、告警合并的策略：时间窗口 + 拓扑关系

flowchart TD A[原始告警信号] --> B[时间窗口合并] B --> C{同一服务?} C -->|是| D[合并到同一个事件] C -->|否| E[检查拓扑关系] E --> F{上下游依赖?} F -->|是| G[只通知下游根因告警] F -->|否| H[分别创建事件] D --> I[事件通知] G --> I H --> I I --> J[值班人员收到合并后的通知] style A fill:#f9f,stroke:#333 style D fill:#bfb,stroke:#333 style G fill:#bbf,stroke:#333 style J fill:#fbb,stroke:#333

告警合并的核心是两个维度：时间窗口和拓扑关系。

时间窗口合并：同一服务在指定时间窗口（比如 10 分钟）内的所有告警，合并成一个事件。

# 告警合并器（简化版） from datetime import datetime, timedelta from typing import List, Dict from dataclasses import dataclass, field @dataclass class Alert: alert_id: str service: str alertname: str severity: str started_at: datetime tags: Dict[str, str] description: str @dataclass class Incident: incident_id: str service: str severity: str started_at: datetime alerts: List[Alert] = field(default_factory=list) status: str = "open" # open / acknowledged / resolved def add_alert(self, alert: Alert): self.alerts.append(alert) # 更新严重等级（取最高的） severity_order = {"critical": 0, "warning": 1, "info": 2} if severity_order[alert.severity] < severity_order[self.severity]: self.severity = alert.severity def summary(self) -> str: """生成事件的摘要""" alert_names = [a.alertname for a in self.alerts] return f"服务 {self.service} 在 {self.started_at} 发生异常: {', '.join(set(alert_names))}" class AlertMerger: def __init__(self, time_window_minutes: int = 10): self.time_window = timedelta(minutes=time_window_minutes) self.open_incidents: Dict[str, Incident] = {} # service -> Incident def process_alert(self, alert: Alert) -> Incident: """处理一条告警，返回它属于哪个事件""" # 检查是否能合并到已有的事件 if alert.service in self.open_incidents: incident = self.open_incidents[alert.service] # 检查时间窗口 if datetime.now() - incident.started_at < self.time_window: incident.add_alert(alert) return incident # 不能合并，创建新事件 incident = Incident( incident_id=f"inc-{datetime.now().strftime('%Y%m%d%H%M%S')}", service=alert.service, severity=alert.severity, started_at=alert.started_at ) incident.add_alert(alert) self.open_incidents[alert.service] = incident return incident

拓扑关系合并：如果服务 A 的告警是因为服务 B（下游）故障导致的，那应该只通知服务 B 的告警。

这需要一个服务依赖拓扑。我们在每个服务的配置里标注了它的下游依赖，告警系统会根据拓扑自动抑制上游服务的告警。

# 服务拓扑配置示例 service: order-service dependencies: - service: user-service type: grpc critical: true - service: payment-service type: http critical: true - service: redis-cache type: redis critical: false

当payment-service故障时，告警系统会检查所有依赖它的服务（比如order-service），然后只发送payment-service的告警，抑制order-service的告警（因为后者的告警是次生告警）。

三、智能降噪：用 AI 识别告警模式

时间窗口合并和拓扑合并是"确定性规则"，能解决 80% 的告警噪声。但还有些噪声是"不确定性"的，比如：

网络抖动导致的偶发超时（可能 10 分钟后自动恢复）
开发在调试时触发的测试告警
下游服务的计划内维护

这些场景，用规则很难覆盖，但AI 可以识别模式。

我们的做法是：把所有告警记录（包括误报）喂给一个分类模型，让它学习"什么样的告警模式通常是误报"。

# 告警模式识别模型（简化版） from sklearn.ensemble import RandomForestClassifier import pandas as pd # 特征工程：把告警转换成模型能理解的特征 def extract_alert_features(alerts: List[Alert]) -> pd.DataFrame: """提取告警的特征""" features = [] for alert in alerts: feature = { "service": alert.service, "alertname": alert.alertname, "severity": alert.severity, "hour_of_day": alert.started_at.hour, "day_of_week": alert.started_at.weekday(), "duration_minutes": (alert.ended_at - alert.started_at).total_seconds() / 60, "similar_alerts_count": count_similar_alerts(alert), "is_deployment_related": check_deployment(alert), "is_downstream_alert": check_downstream(alert), } features.append(feature) return pd.DataFrame(features) # 训练模型（用历史告警数据，标注哪些是误报） model = RandomForestClassifier(n_estimators=100) train_data = load_historical_alerts() X = extract_alert_features(train_data["alerts"]) y = train_data["is_false_positive"] model.fit(X, y) # 预测新告警是否是误报 def predict_false_positive(alert: Alert) -> float: """预测告警是误报的概率""" features = extract_alert_features([alert]) probability = model.predict_proba(features)[0][1] # 返回是正样本的概率 return probability

落地时的关键：AI 模型的预测结果不要直接用来"屏蔽告警"，而是用来"调整告警的优先级"。比如，模型预测某条告警有 80% 概率是误报，那就把它从 P1 降级到 P3，而不是完全不通知。

四、降噪效果的量化评估

告警降噪做得好不好，不能只凭感觉，要量化评估。我们关注三个指标：

1. 告警压缩比：合并前告警数 / 合并后事件数

目标：≥ 5:1
计算方法：比如原来每天 100 条告警，合并后变成 20 个事件，压缩比就是 5:1

2. 误报率：被人工标记为"误报"的事件数 / 总事件数

目标：≤ 10%
优化方法：分析误报的共同特征，补充到合并规则或 AI 模型中

3. 平均确认时间（MTTA）：从事件创建到被人工确认的时间

目标：P0 < 2min, P1 < 10min
优化方法：改进事件通知的内容质量（让值班人员能快速判断严重性）

# 降噪效果评估报告（每周自动生成） def generate_noise_reduction_report(start: datetime, end: datetime) -> dict: """生成告警降噪效果报告""" alerts = get_alerts_in_timerange(start, end) incidents = get_incidents_in_timerange(start, end) compression_ratio = len(alerts) / len(incidents) if incidents else 0 false_positive_rate = count_false_positives(incidents) / len(incidents) mtta = calculate_mtta(incidents) report = { "period": f"{start.date()} ~ {end.date()}", "total_alerts": len(alerts), "total_incidents": len(incidents), "compression_ratio": f"{compression_ratio:.1f}:1", "false_positive_rate": f"{false_positive_rate:.1%}", "mtta_p0": f"{mtta['P0']:.1f}min", "mtta_p1": f"{mtta['P1']:.1f}min", "recommendations": generate_recommendations(compression_ratio, false_positive_rate, mtta) } return report