AIOps 智能运维：从"告警风暴"到"根因定位"的工程实践

from collections import defaultdict
import statistics

# ---- 1. 动态基线异常检测：EWMA + 残差判定 ----
def detect_anomaly(series: list[float], alpha=0.3, k=3.0) -> list[int]:
    """返回异常点下标。用 EWMA 做基线，残差超过 k 倍标准差即判异常"""
    baseline, residuals, anomalies = series[0], [], []
    for i, x in enumerate(series):
        baseline = alpha * x + (1 - alpha) * baseline   # 指数加权基线
        residuals.append(x - baseline)
    # 用中位数绝对偏差(MAD)抗离群，比标准差稳健
    med = statistics.median(residuals)
    mad = statistics.median([abs(r - med) for r in residuals]) or 1e-6
    for i, r in enumerate(residuals):
        if abs(r - med) > k * 1.4826 * mad:              # 1.4826: MAD→σ 换算
            anomalies.append(i)
    return anomalies

# ---- 2. 告警聚合：同一拓扑子树内的告警合并为一个事件 ----
def aggregate_alarms(alarms: list[dict], parent: dict[str, str]) -> list[dict]:
    """alarms: [{service, ts, metric}], parent: 服务->上游服务"""
    def root_of(svc):
        while svc in parent:
            svc = parent[svc]
        return svc
    groups = defaultdict(list)
    for a in alarms:
        groups[(root_of(a["service"]), a["ts"] // 60)].append(a)  # 按根+分钟窗口聚合
    return [{"services": list({a["service"] for a in g}), "alarms": g}
            for g in groups.values()]

# ---- 3. 根因排序：异常最早出现 + 拓扑越深越可疑 ----
def rank_root_cause(group: dict, parent: dict[str, str], depth: dict[str, int]):
    score = {}
    first_ts = min(a["ts"] for a in group["alarms"])
    for a in group["alarms"]:
        svc = a["service"]
        # 越早异常加分；在依赖图越深（越靠下游）加分
        score[svc] = (1.0 / (1 + (a["ts"] - first_ts))) + 0.5 * depth.get(svc, 0)
    return sorted(score.items(), key=lambda kv: -kv[1])

# 使用示例
series = [100, 102, 99, 101, 400, 380, 120, 101]          # 某指标序列
print("异常点:", detect_anomaly(series))
parent = {"gateway": "web", "web": "api", "api": "db"}    # 服务依赖：gateway→web→api→db
depth  = {"gateway": 0, "web": 1, "api": 2, "db": 3}
alarms = [{"service": s, "ts": 1000 + i * 5, "metric": "latency"}
          for i, s in enumerate(["db", "api", "web", "gateway"])]  # db 最先异常
for svc, sc in rank_root_cause(aggregate_alarms(alarms, parent)[0], parent, depth):
    print(f"{svc}: {sc:.2f}")