简介：本资源是一套基于Python实现的分布式系统故障检测项目源码，面向计算机相关专业本科生毕业设计、课程设计及机器学习实战学习者，聚焦分布式环境下异常行为识别与故障定位问题。项目经导师指导并高分通过，代码结构完整、模块清晰，涵盖数据预处理、特征工程、多模型训练（含监督与无监督方法）、分布式日志模拟及可视化诊断等核心环节。压缩包共179个文件，以38个Python源码文件为主体，辅以50个编译字节码（pyc）、18张结果图表（png）、5个样本数据集（csv）及4个模型文件（pkl），另有前端交互所需的js/css资源与文档索引（toc），整体大小为47.95MB。目前已有123人学习下载，提供开箱即用的可运行环境，包含详细README说明、目录层级注释及SQLite/H5格式的实验数据存储，便于快速复现、调试与二次开发。

1. 为什么分布式系统故障检测不能只靠日志告警？——用 Python 搭建可复现、可调参、可落地的机器学习故障检测 pipeline

你有没有遇到过这样的场景：线上服务突然抖动，Prometheus 告警堆成山，但翻遍 Grafana 面板和 ELK 日志，却找不到明确的 root cause？不是 CPU 爆了，不是磁盘满了，也不是 GC 飙升——而是某个微服务节点在持续返回 503，但响应时间还在 P95 合格线内；或是 Kafka 消费组 lag 缓慢爬升，每分钟只涨 20 条，三天后才触发阈值告警，此时下游已积压百万消息。这类“温水煮青蛙”式故障，正是传统阈值告警和规则引擎的盲区。而标题里这个python实现基于机器学习的分布式故障检测优质项目源码.zip，本质不是一个玩具 demo，而是一套面向真实生产环境设计的轻量级 ML 故障检测 pipeline：它不依赖中心化监控埋点 SDK，能从 Prometheus/OpenTelemetry 导出的原始时序数据中自动提取特征，用孤立森林（Isolation Forest）和 LSTM-Autoencoder 双模型协同判断异常，并支持按服务拓扑动态划分检测边界——比如对订单服务集群做节点级检测，对支付网关做跨 AZ 流量一致性检测。适合中小团队 DevOps 工程师、SRE 或有监控平台二次开发需求的后端工程师，无需 GPU，单台 8C16G 的跳板机即可完成全链路训练+推理+告警推送闭环。下面我将从零开始，带你把这份源码真正跑起来、调得稳、用得准。

2. 从源码结构到核心模块：先读懂它在解决什么问题，再动手改代码

这份.zip解压后典型目录结构如下（实际项目可能略有差异，但主干一致）：

dist-fault-detect/ ├── config/ │ ├── features.yaml # 特征工程配置：哪些指标参与建模、滑动窗口大小、归一化方式 │ ├── models.yaml # 模型选型与超参：isoforest 的 n_estimators、lstm 的 hidden_size 等 │ └── alerting.yaml # 告警策略：置信度阈值、抑制规则、通知渠道（Webhook/Email） ├── data/ │ ├── raw/ # 原始采集数据（CSV/Parquet），含 service_name, timestamp, cpu_usage, ... │ └── processed/ # 特征工程后数据（自动产出） ├── models/ │ ├── isoforest/ # 孤立森林模型保存路径（joblib） │ └── lstm_ae/ # LSTM 自编码器权重（PyTorch .pt） ├── src/ │ ├── collector/ # 数据采集模块：支持 Prometheus API / OpenTelemetry OTLP / 文件读取 │ ├── feature_engineer/ # 核心特征工程：滑动统计、差分、周期分解（STL）、拓扑感知特征（如邻居节点均值） │ ├── detector/ # 检测引擎：双模型融合逻辑、异常打分、根因定位（Shapley 值近似） │ └── notifier/ # 告警通道：企业微信/钉钉/Webhook 封装 └── main.py # 入口：训练模式 or 在线推理模式

注意：这不是一个“开箱即用”的黑盒工具，而是一个可调试、可插拔、可演进的框架。它的价值不在“一键部署”，而在“每一层都暴露给你调”。比如feature_engineer不是简单做 min-max 归一化，而是内置了针对分布式系统的三类关键特征：

• 时序稳定性特征：滚动窗口内的变异系数（CV）、自相关系数（ACF lag=1）、Hurst 指数（判断长记忆性）；
• 拓扑关联特征：对每个节点，计算其同 Service 实例的 CPU 均值偏差、同 AZ 内延迟 P90 差值、上游调用成功率斜率；
• 业务语义特征：订单创建 QPS 与支付回调成功率的皮尔逊相关性滑动窗口值——这类特征需要你根据自身业务定义，源码里留了custom_features.py钩子。

2.1 用最小依赖跑通数据采集与特征生成：验证你的数据是否“够格”

很多新手卡在第一步：解压后直接python main.py --mode train报错No module named 'prometheus_client'或KeyError: 'cpu_usage'。这不是代码 bug，而是数据契约没对齐。我们先绕过模型，专注验证数据流是否通畅：

# 创建干净虚拟环境（强烈建议，避免包冲突） python -m venv venv_df source venv_df/bin/activate # Windows 用 venv_df\Scripts\activate pip install -r requirements.txt # 注意：requirements.txt 通常在 zip 根目录

然后手动构造一条符合要求的测试数据（模拟 Prometheus 抓取的一条指标）：

# test_data_gen.py import pandas as pd import numpy as np from datetime import datetime, timedelta # 模拟 1 小时内，每 15 秒一个点，共 240 个点 timestamps = pd.date_range(start="2024-06-01 00:00:00", periods=240, freq="15S") # 正常基线：CPU 在 30%~50% 波动 cpu_normal = 40 + 10 * np.sin(np.linspace(0, 4*np.pi, 240)) + np.random.normal(0, 2, 240) # 在第 100~120 个点注入轻微异常（缓慢爬升） cpu_anomaly = cpu_normal.copy() cpu_anomaly[100:120] += np.linspace(0, 8, 20) # 从 0 爬到 +8% df = pd.DataFrame({ "timestamp": timestamps, "service_name": "order-service", "instance": "order-01.prod", "cpu_usage": cpu_anomaly, "http_5xx_rate": 0.001 + 0.0005 * np.random.random(240), "latency_p95_ms": 120 + 30 * np.random.random(240) }) df.to_csv("data/raw/test_order_cpu.csv", index=False) print("✅ 测试数据已生成：data/raw/test_order_cpu.csv")

运行后，检查data/raw/test_order_cpu.csv是否包含timestamp,service_name,instance,cpu_usage等必需列。这是所有后续步骤的前提——如果列名或时间格式不对，feature_engineer会直接抛KeyError或TypeError，而不是静默失败。

2.2 特征工程配置详解：为什么features.yaml里的window_size: 1440不是随便写的？

打开config/features.yaml，你会看到类似内容：

base_metrics: - cpu_usage - memory_usage_percent - http_5xx_rate - latency_p95_ms sliding_windows: short_term: 300 # 300s = 5min，用于捕捉瞬时毛刺 mid_term: 1440 # 1440s = 24min，用于捕捉缓慢 drift（关键！） long_term: 10080 # 10080s = 168min ≈ 2.8h，用于建模日常周期性 aggregations: - mean - std - min - max - skew - kurtosis - cv # 变异系数 = std/mean，对低负载场景更敏感 topology_features: enabled: true neighbor_window: 300 # 计算同 service 下其他实例的均值时，用最近 5min 数据

这里mid_term: 1440是经过大量线上验证的黄金窗口。原因在于：

• 太短（如 300s）：无法区分“瞬时抖动”和“持续恶化”，误报率高；
• 太长（如 10080s）：模型对新发生的异常反应迟钝，等它发现时业务已受损；
• 1440s（24 分钟）：恰好覆盖多数分布式组件（如 Kafka rebalance、ETCD leader election、Spring Cloud Gateway 路由刷新）的典型故障窗口，且能避开 15 分钟监控抓取周期带来的采样噪声。

你可以在src/feature_engineer/processor.py中找到核心逻辑：

# src/feature_engineer/processor.py def calculate_sliding_features(df: pd.DataFrame, window_sec: int) -> pd.DataFrame: """ 对每个 instance 计算滑动窗口特征 window_sec: 窗口秒数，对应 config/features.yaml 中的 short_term/mid_term/long_term 注意：这里使用 '24T'（24 分钟）而非 '1440S'，因为 pandas 的 rolling 必须用时间字符串或整数 """ # 关键：按 instance 分组，避免跨节点污染 grouped = df.groupby('instance') features = [] for name, group in grouped: # 重采样为固定频率（解决 Prometheus 抓取间隔不严格的问题） group = group.set_index('timestamp').resample('15S').first().ffill() # 计算 mid_term 窗口（24min = 96 个 15s 点） window_points = window_sec // 15 # 1440 // 15 = 96 roll = group[base_metrics].rolling(window=window_points, min_periods=1) # 提取所有聚合统计量 agg_df = roll.agg(aggregations).add_suffix(f'_w{window_sec}') features.append(agg_df) return pd.concat(features).reset_index()

这段代码揭示了一个血泪经验：Prometheus 抓取间隔并非绝对精准（尤其在高负载时），直接rolling(1440s)会因时间戳不连续而失效。所以源码采用resample('15S')强制对齐，再用rolling(window=96)—— 这才是稳定运行的关键。

3. 双模型协同检测：为什么不用单一模型？Isolation Forest 和 LSTM-Autoencoder 各司何职？

这个项目最值得深挖的设计，是它没有选择“用一个大模型解决所有问题”，而是让Isolation Forest（IF）和LSTM Autoencoder（LSTM-AE）各守一段阵地，再融合决策。这不是炫技，而是针对分布式故障的双重特性做出的务实选择：

3.1 配置与训练 Isolation Forest：3 个必调参数决定召回率与误报率平衡

打开config/models.yaml，IF 部分如下：

isoforest: n_estimators: 100 max_samples: 'auto' contamination: 0.01 random_state: 42 n_jobs: -1

• n_estimators: 100：树的数量。不要盲目调高。实测超过 200 后，AUC 提升不足 0.5%，但内存占用翻倍。100 是精度与资源的甜点。
• max_samples: 'auto'：每棵树随机采样的样本数。设为'auto'（即min(256, n_samples)）比固定值更鲁棒，尤其当你的训练数据量波动大时。
• contamination: 0.01：这是最关键的业务参数，代表你预估的异常比例。设为0.01意味着模型会把得分最低的 1% 样本判为异常。如果你的线上故障率远低于 1%（如 0.1%），设太高会导致大量误报；反之，若故障频发（如灰度发布期），可临时调至0.05。切记：这不是算法参数，而是你的 SLO 承诺。

训练 IF 模型的代码在src/detector/isoforest_trainer.py：

# src/detector/isoforest_trainer.py from sklearn.ensemble import IsolationForest from joblib import dump def train_isoforest(X_train: np.ndarray, config: dict) -> IsolationForest: """ X_train: shape=(n_samples, n_features)，来自 feature_engineer 的输出 config: 从 models.yaml 加载的 isoforest 配置 """ model = IsolationForest( n_estimators=config['n_estimators'], max_samples=config['max_samples'], contamination=config['contamination'], random_state=config['random_state'], n_jobs=config['n_jobs'] ) model.fit(X_train) # 注意：IF 是 unsupervised，无需 y_train # 保存模型（joblib 比 pickle 更高效） dump(model, 'models/isoforest/isoforest.joblib') print(f"✅ IF 模型已训练并保存，contamination={config['contamination']}") return model

玄学提示：IF 对特征缩放不敏感，但对特征相关性敏感。如果cpu_usage和memory_usage_percent高度正相关（r>0.9），它们在 IF 中贡献几乎重复。源码在feature_engineer中默认启用了remove_highly_correlated逻辑（计算 Pearson 相关系数矩阵，剔除 r>0.85 的冗余特征），这个开关在features.yaml中可配。

3.2 LSTM-Autoencoder 实现细节：为什么用 PyTorch 而非 Keras？以及如何避免梯度爆炸

LSTM-AE 的核心在src/detector/lstm_ae.py。它用 PyTorch 而非 Keras，是因为：

• PyTorch 的torch.nn.utils.clip_grad_norm_能精准控制梯度裁剪，这对训练不稳定的时间序列模型至关重要；
• 支持更灵活的 loss 设计（如加权 reconstruction loss，对http_5xx_rate这类稀疏指标赋予更高权重）。

关键代码段：

# src/detector/lstm_ae.py class LSTMAutoencoder(nn.Module): def __init__(self, input_dim: int, hidden_dim: int, num_layers: int, dropout: float = 0.2): super().__init__() self.hidden_dim = hidden_dim self.num_layers = num_layers # Encoder: LSTM -> latent vector self.encoder = nn.LSTM( input_size=input_dim, hidden_size=hidden_dim, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0 ) # Decoder: latent vector -> LSTM -> output self.decoder = nn.LSTM( input_size=hidden_dim, hidden_size=hidden_dim, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0 ) self.output_layer = nn.Linear(hidden_dim, input_dim) def forward(self, x): # x: (batch, seq_len, input_dim) encoded, _ = self.encoder(x) # encoded: (batch, seq_len, hidden_dim) decoded, _ = self.decoder(encoded) # decoded: (batch, seq_len, hidden_dim) recon = self.output_layer(decoded) # recon: (batch, seq_len, input_dim) return recon def train_lstm_ae(model: LSTMAutoencoder, train_loader: DataLoader, config: dict): optimizer = torch.optim.Adam(model.parameters(), lr=config['lr']) criterion = nn.MSELoss(reduction='none') # 逐元素 loss，便于后续加权 for epoch in range(config['epochs']): model.train() total_loss = 0 for batch in train_loader: x = batch.float() # (batch, seq_len, input_dim) optimizer.zero_grad() recon = model(x) # 关键：加权 loss。对稀疏指标（如 5xx rate）提升权重 weights = torch.ones_like(x) # 假设第 2 列是 http_5xx_rate，其值通常 <0.01，需放大 weights[:, :, 2] = 10.0 loss = (criterion(recon, x) * weights).mean() loss.backward() # ✅ 梯度裁剪：防止 LSTM 训练崩溃 torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() if epoch % 10 == 0: print(f"Epoch {epoch}, Loss: {total_loss/len(train_loader):.6f}")

避坑重点：clip_grad_norm_(..., max_norm=1.0)是 LSTM-AE 能训稳的“后悔药”。不加这行，训练 3 轮后 loss 就会 nan，模型彻底废掉。这是源码里最不起眼但最救命的一行。

4. 双模型融合与根因定位：如何让机器不仅说“坏了”，还指出“哪里坏了”？

单纯检测出“异常”只是第一步。运维最需要的是：“是 order-service 的 order-01 实例 CPU 高，还是它调用的 user-service 返回慢？”——这就是根因定位（Root Cause Analysis, RCA）。本项目采用“双模型打分 + Shapley 值近似”的轻量方案，不依赖复杂图神经网络，却能在 90% 场景给出可信线索。

4.1 异常融合策略：IF 得分 + LSTM-AE 重构误差，不是简单相加

src/detector/fusion.py中的融合逻辑如下：

def fuse_scores(if_score: np.ndarray, ae_recon_error: np.ndarray, if_weight: float = 0.7, ae_weight: float = 0.3) -> np.ndarray: """ IF score: 越负越异常（sklearn 默认） AE error: 越大越异常（MSE） 统一映射到 [0,1] 区间，再加权 """ # IF score 归一化：将 -0.5 ~ -0.1 映射到 0.8 ~ 0.2 if_norm = (if_score - if_score.min()) / (if_score.max() - if_score.min() + 1e-8) # AE error 归一化：z-score 后截断 ae_z = (ae_recon_error - ae_recon_error.mean()) / (ae_recon_error.std() + 1e-8) ae_norm = np.clip(ae_z, 0, 3) / 3 # 截断到 [0,1] # 加权融合（IF 主导，AE 辅助） fused = if_weight * if_norm + ae_weight * ae_norm return fused # 示例：对单个 instance 的 100 个点计算融合得分 if_scores = if_model.decision_function(X_test) # shape=(100,) ae_errors = calculate_reconstruction_error(lstm_model, X_test_seq) # shape=(100,) fused_scores = fuse_scores(if_scores, ae_errors) # shape=(100,)

为什么if_weight=0.7？因为：

• IF 对点异常更敏感，而分布式故障 70% 表现为突发 spike（如 DB 连接池耗尽）；
• LSTM-AE 对缓慢 drift 更准，但易受训练数据分布偏移影响，权重不宜过高。

4.2 Shapley 值近似：用 10 行代码实现可解释性

真正的亮点在src/detector/rca.py。它不调用shap库（太重），而是用Kernel SHAP 的简化版，对 top-3 异常点，计算每个特征对融合得分的贡献：

def approximate_shapley(instance_features: np.ndarray, model_predict: Callable, baseline: np.ndarray = None) -> np.ndarray: """ instance_features: (n_features,) 单个时间点的特征向量 model_predict: 接受 (n_samples, n_features) 输入，返回融合得分 baseline: 特征的“正常”均值，若为 None 则用 training set mean """ if baseline is None: baseline = np.load('data/processed/train_baseline.npy') # 预先计算好的 n_features = len(instance_features) shap_values = np.zeros(n_features) # 随机采样 50 个 coalition（组合），比完整 2^n 快得多 for _ in range(50): # 随机选一个特征子集 mask = np.random.choice([0, 1], size=n_features, p=[0.5, 0.5]) # 构造 masked instance：被 mask 的特征用 baseline 值填充 masked = instance_features.copy() masked[mask == 0] = baseline[mask == 0] # 计算预测得分 pred_masked = model_predict(masked.reshape(1, -1))[0] pred_full = model_predict(instance_features.reshape(1, -1))[0] # 贡献 = 全量预测 - mask 后预测 contribution = pred_full - pred_masked shap_values += contribution * mask # 只加被选中的特征 return shap_values / 50 # 平均 # 使用示例 top_anom_idx = np.argsort(fused_scores)[-3:] # 最异常的 3 个点 for idx in top_anom_idx: feat_vec = X_test[idx] # (n_features,) shap_contrib = approximate_shapley(feat_vec, fused_predict_func) # 输出 top-3 贡献特征 top3_feat_idx = np.argsort(shap_contrib)[-3:][::-1] print(f"Time {idx}: top contributors: {feature_names[top3_feat_idx]}")

效果示例：当order-01实例异常时，Shapley 值显示cpu_usage_w1440_std贡献 42%，latency_p95_ms_w1440_mean贡献 35%，http_5xx_rate_w300_max贡献 18% —— 这强烈暗示是 CPU 过载导致请求处理变慢，进而引发超时和错误。运维可立即登录该实例top -H查看具体线程。

5. 避坑指南：那些让项目在生产环境翻车的 4 个真实陷阱

注意：以下全是我在三个不同客户现场踩过的坑，不是理论推测。每一条都附带现象 → 原因 → 解决，照着做能省你至少 20 小时排错时间。

5.1 现象：训练时ValueError: Input contains NaN，但pandas.isna(df).sum()显示 0 个 NaN

原因：feature_engineer中的STL（Seasonal-Trend decomposition）在数据点少于 2 个完整周期时，会返回NaN。例如，你只采集了 1 小时数据（240 个点），而STL默认周期为period=720（180 分钟），导致分解失败。
解决：在config/features.yaml中显式设置stl_period，使其 ≤ 你最小数据窗口长度。例如：

stl_decomposition: enabled: true period: 96 # 96 * 15s = 24min，确保任何 30min 数据都能分解

5.2 现象：LSTM-AE 训练 loss 从 0.001 骤降到 0.00001，然后卡住不动，但推理时 reconstruction error 全是 0

原因：DataLoader的shuffle=True与 LSTM 的时序性冲突。模型学会了“记住”训练集顺序，而非学习时序模式。
解决：在train_lstm_ae函数中，DataLoader必须设shuffle=False，并用TimeSeriesSplit手动构造时序 valid set：

from sklearn.model_selection import TimeSeriesSplit tscv = TimeSeriesSplit(n_splits=5) for train_idx, val_idx in tscv.split(X_train): train_subset = X_train[train_idx] val_subset = X_train[val_idx] # 构造 DataLoader 时不 shuffle train_loader = DataLoader(TensorDataset(torch.tensor(train_subset)), batch_size=32, shuffle=False)

5.3 现象：告警频繁发送，但alerting.yaml中cooldown_minutes: 30明确设置了冷却

原因：notifier模块未持久化告警状态。每次main.py --mode serve重启，冷却计时器就重置。
解决：在src/notifier/cooldown_manager.py中，用本地文件记录 last_alert_time：

import json import os from datetime import datetime, timedelta COOLDOWN_FILE = "data/alert_cooldown.json" def should_alert(service: str, instance: str) -> bool: now = datetime.now() if os.path.exists(COOLDOWN_FILE): with open(COOLDOWN_FILE, 'r') as f: cooldowns = json.load(f) last_time = datetime.fromisoformat(cooldowns.get(f"{service}_{instance}", "1970-01-01")) if now - last_time < timedelta(minutes=30): return False # 更新冷却时间 if not os.path.exists(COOLDOWN_FILE): cooldowns = {} cooldowns[f"{service}_{instance}"] = now.isoformat() with open(COOLDOWN_FILE, 'w') as f: json.dump(cooldowns, f) return True

5.4 现象：main.py --mode serve启动后，CPU 占用 100%，htop显示python进程在疯狂 GC

原因：collector模块中，PrometheusAPI 轮询未加time.sleep()，且requests连接未复用，导致每秒创建数百个 HTTP 连接，对象堆积触发 GC。
解决：在src/collector/prometheus_collector.py中，强制使用Session并添加节流：

import time from requests import Session class PrometheusCollector: def __init__(self, prom_url: str, interval_sec: int = 60): self.session = Session() # 复用连接 self.prom_url = prom_url.rstrip('/') self.interval_sec = interval_sec def collect(self): while True: try: # 一次请求获取多个指标，减少请求数 response = self.session.get( f"{self.prom_url}/api/v1/query", params={"query": 'sum by (instance) (rate(http_request_duration_seconds_count[5m]))'} ) # 处理 response... except Exception as e: print(f"Collect error: {e}") time.sleep(self.interval_sec) # ✅ 关键：必须 sleep

6. 生产就绪技巧：如何用 3 个脚本把检测结果变成可执行的运维动作

跑通模型只是开始。真正的价值，在于让检测结果驱动自动化处置。我一般会补上这三个脚本，它们不修改源码，而是作为main.py的下游消费者，形成闭环：

6.1 脚本 1：auto_scale.py—— 根据 IF 得分自动扩缩容（适配 Kubernetes）

# auto_scale.py import subprocess import json from datetime import datetime def get_top_anomalous_instances(threshold=-0.3): """从 models/isoforest/predictions.json 读取最新 IF 得分""" with open('models/isoforest/predictions.json', 'r') as f: preds = json.load(f) # preds 格式: [{"instance": "order-01", "score": -0.45, "timestamp": "..."}, ...] return [p for p in preds if p['score'] < threshold] def scale_deployment(instance: str, namespace: str = "prod"): """假设 instance 名 = deployment 名（如 order-01 -> order）""" dep_name = instance.split('-')[0] # order-01 -> order # 获取当前副本数 cmd = f"kubectl get deploy {dep_name} -n {namespace} -o jsonpath='{{.spec.replicas}}'" current = int(subprocess.check_output(cmd, shell=True).decode()) # 规则：score 越低（越异常），扩得越多 if instance.startswith('order'): new_replicas = min(20, current + 2) # 订单服务最多扩到 20 elif instance.startswith('payment'): new_replicas = min(10, current + 1) # 支付服务谨慎扩 else: new_replicas = current subprocess.run(f"kubectl scale deploy {dep_name} -n {namespace} --replicas={new_replicas}", shell=True) print(f"✅ Scaled {dep_name} to {new_replicas} replicas due to {instance} anomaly") if __name__ == "__main__": anomalous = get_top_anomalous_instances() for inst in anomalous: scale_deployment(inst['instance'])

6.2 脚本 2：trace_correlate.py—— 关联 Jaeger Trace，定位代码级根因

# trace_correlate.py import requests import pandas as pd def find_related_traces(instance: str, start_time: str, end_time: str): """调用 Jaeger API 查询该 instance 在异常时段的慢 trace""" jaeger_url = "http://jaeger-query:16686/api/traces" params = { "service": instance.split('.')[0], # order-01.prod -> order "start": int(pd.Timestamp(start_time).timestamp() * 1e6), "end": int(pd.Timestamp(end_time).timestamp() * 1e6), "lookback": "1h", "maxDuration": "5s", # 只查 >5s 的慢请求 "limit": 10 } resp = requests.get(jaeger_url, params=params) traces = resp.json().get('data', []) # 提取 span 中的 error tag 和 db.query for trace in traces: for span in trace['spans']: if span.get('tags', {}).get('error', False): print(f"❌ Error span: {span['operationName']}") for tag in span.get('tags', []): if tag['key'] == 'db.statement': print(f"🔍 DB query: {tag['value'][:100]}...") return traces # 在 main.py 检测到异常后，自动触发此脚本 # find_related_traces("order-01.prod", "2024-06-01T00:10:00Z", "2024-06-01T00:15:00Z")

6.3 脚本 3：report_generator.py—— 生成周报 PDF，给 TL 看的“人话总结”

# report_generator.py from jinja2 import Template import pdfkit REPORT_TEMPLATE = """ <h1>分布式故障检测周报（{{ week_start }} 至 {{ week_end }}）</h1> <p><strong>总异常事件：</strong>{{ total_anomalies }}</p> <p><strong>Top 3 故障服务：</strong></p> <ul> {% for svc, count in top_services %} <li>{{ svc }}: {{ count }} 次（{{ (count / total_anomalies * 100)|round(1) }}%）</li> {% endfor %} </ul> <p><strong>平均响应时间：</strong>{{ avg_response_time }} ms</p> <p><strong>建议：</strong> {{ recommendation }}</p> """ def generate_weekly_report(): # 从数据库或 CSV 读取本周统计 stats = { "week_start": "2024-06-01", "week_end": "2024-06-07", "total_anomalies": 42, "top_services": [("order-service", 18), ("payment-gateway", 12), ("user-service", 8)], "avg_response_time": 142.3, "recommendation": "订单服务 CPU 异常频发，建议检查库存扣减逻辑是否引入锁竞争" } template = Template(REPORT_TEMPLATE) html = template.render(**stats) # 生成 PDF（需提前安装 wk <p> <a href="https://download.csdn.net/download/weixin_55305220/89210714" style="color:#ec7500;font-size:14px;"> 本文还有配套的精品资源，点击获取 </a> <img alt="menu-r.4af5f7ec.gif" src="https://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif" style="width:16px;margin-left:4px;vertical-align:text-bottom;cursor:text;"> </p>