基于 Prometheus PromQL 的时序异常波动自动归因-编程实验室

基于 Prometheus PromQL 的时序异常波动自动归因

在生产故障发生时，监控系统往往只能呈现“发生了什么异常”（比如订单服务的 P99 响应时间从 30ms 突增到了 1200ms），却无法直接指出“为什么会发生这种异常”。值班工程师为了找出导致 P99 飙升的罪魁祸首，通常需要在 Grafana 上逐个打开数十个指标面板，手动比对 CPU 使用率、JVM GC 暂停耗时、数据库慢查询计数、RPC 客户端错误率以及下游接口的延迟曲线。

在多维时序数据极其庞大的现代云原生环境中，这种依靠人工肉眼寻找时序曲线相关性的方式不仅耗时漫长（通常需要 15~30 分钟），而且极易受工程师个人经验偏差的影响。

为了实现异常指标的秒级归因定位，我们在故障诊断 Agent 中引入了基于 PromQL 批量时序关联分析与皮尔逊相关系数（Pearson Correlation Coefficient）的自动化归因算法。

时序波动的数学归因原理

当目标核心指标 $Y(t)$（例如服务响应延迟）在时间窗口 $[T_{start}, T_{end}]$ 内发生突变时，我们的目标是在该服务关联的基础设施与依赖组件的候选指标集合 ${X_1(t), X_2(t), \dots, X_m(t)}$ 中，找出与 $Y(t)$ 波动形态最吻合、且在时间维度上具有前置或同步因果关系的特征指标。

皮尔逊相关系数 $r(X, Y)$ 用于衡量两个连续时序变量之间的线性相关程度：

$$r(X, Y) = \frac{\sum_{i=1}^n (X_i - \bar{X})(Y_i - \bar{Y})}{\sqrt{\sum_{i=1}^n (X_i - \bar{X})^2} \sqrt{\sum_{i=1}^n (Y_i - \bar{Y})^2}}$$

当 $r \ge 0.85$ 时，说明候选指标 $X$ 与目标指标 $Y$ 存在极强的同步正相关性。
结合时间序列的互相关函数（Cross-Correlation），若 $X$ 的波峰领先 $Y$ 的波峰 $k$ 个采样周期（$k > 0$），则 $X$ 极大概率是导致 $Y$ 恶化的物理根因（例如：DB 慢查询先飙升 30 秒，随后引发应用线程池打满与接口响应超时）。

Python 实现自动化 PromQL 归因诊断引擎

诊断 Agent 在收到异常告警后，自动根据服务元数据生成一组候选 PromQL 查询，拉取时序矩阵并执行高速相关性打分：

import numpy as np import pandas as pd import requests from typing import Dict, List, Any class TimeSeriesAttributionEngine: def __init__(self, prometheus_base_url: str = "http://prometheus-k8s.monitoring.svc:9090"): self.prom_url = f"{prometheus_base_url}/api/v1/query_range" def fetch_metric_series(self, promql: str, start_ts: int, end_ts: int, step: str = "15s") -> pd.Series: """从 Prometheus 拉取指定时间范围的时序数据并转为 Pandas Series""" params = { "query": promql, "start": start_ts, "end": end_ts, "step": step } resp = requests.get(self.prom_url, params=params, timeout=10) data = resp.json() if data.get("status") != "success": return pd.Series(dtype=float) results = data.get("data", {}).get("result", []) if not results: return pd.Series(dtype=float) # 默认取第一个时序序列 values = results[0].get("values", []) timestamps = [int(v[0]) for v in values] metrics = [float(v[1]) for v in values] return pd.Series(data=metrics, index=pd.to_datetime(timestamps, unit='s')) def run_root_cause_attribution( self, target_promql: str, candidate_promqls: Dict[str, str], start_ts: int, end_ts: int ) -> List[Dict[str, Any]]: """ 输入异常目标 PromQL 与候选指标字典，输出按因果相关度排序的根因列表 """ target_series = self.fetch_metric_series(target_promql, start_ts, end_ts) if target_series.empty or len(target_series) < 10: return [] # 目标指标归一化 (Z-Score) target_norm = (target_series - target_series.mean()) / (target_series.std() + 1e-9) attribution_results = [] for candidate_name, candidate_promql in candidate_promqls.items(): candidate_series = self.fetch_metric_series(candidate_promql, start_ts, end_ts) if candidate_series.empty or len(candidate_series) != len(target_series): continue candidate_norm = (candidate_series - candidate_series.mean()) / (candidate_series.std() + 1e-9) # 计算皮尔逊相关系数 correlation = np.corrcoef(target_norm, candidate_norm)[0, 1] if not np.isnan(correlation) and correlation > 0.70: attribution_results.append({ "candidate_name": candidate_name, "promql": candidate_promql, "correlation_score": round(float(correlation), 4) }) # 按相关度得分降序排序 attribution_results.sort(key=lambda x: x["correlation_score"], reverse=True) return attribution_results