CPP-Log-Importance-Scoring/ML-implementation
GitHub: CPP-Log-Importance-Scoring/ML-implementation
基于机器学习的网络日志重要性评分与异常检测系统，通过无监督异常检测、图关联分析和跨运行因果链，从海量 syslog 中自动识别事件并定位根本原因。
Stars: 0 | Forks: 5
# HPE CPP3 - 日志重要性评分与跨信号关联
本项目作为 **HPE CPP3** 学生计划的一部分构建。该系统摄取原始日志，检测异常，将相关行分组为**事件**，按重要性对它们进行排名，识别**根本原因**，将跨天的事件链接为**因果链**，并通过面向分析师的 Streamlit 仪表板展示所有内容。
## 目录
- [问题](#the-problem)
- [何为“好”（需求）](#what-good-means-requirements)
- [系统架构](#system-architecture)
- [处理流水线](#processing-pipeline)
- [核心评分模型](#the-core-scoring-model)
- [从日志行到事件](#from-lines-to-incidents)
- [跨信号关联与因果链](#cross-signal-correlation--causal-chains)
- [性能指标](#performance-metrics)
- [快速开始](#quickstart)
- [模块概述](#module-overview)
- [数据契约](#data-contracts)
- [配置](#configuration)
- [仪表板](#the-dashboard)
- [测试](#testing)
- [薄弱环节与未来计划](#where-its-weak--whats-next)
- [技术栈](#technology-stack)
- [团队](#team)
## 问题
**网络系统中的日志过载。** 交换机和设备持续不断地生成海量的 syslog 日志行，人工分诊无法扩展：
- 📊 **海量数据** —— 交换机和设备持续不断生成日志。
- 🔇 **大多数是噪声** —— 大多数日志行是常规、低价值的心跳信号。
- 🔍 **大海捞针** —— 真正的故障占比不到 3%，且被淹没在数据流中。
- 🏷️ **严重性标签在两个方向上都有误导** —— 标记为 `CRITICAL` 的良性日志（例如“ASIC temp 50C — NORMAL”心跳信号），以及标记为 `INFO` 的真实故障（例如 OOM 级联故障）。
- 🔗 **事件是跨时间和主机的相关日志的*集合*** —— 而不是单行日志。
- ⛓️ 日志和信号之间**缺乏关联**会减慢根本原因分析；今天的故障可能就是明天宕机的*前兆*。
## 何为“好”（需求）
| 功能性 | 非功能性 |
|---|---|
| 摄取原始 syslog（批量 + 分析师上传） | **无监督** —— 不需要标记的故障数据 |
| 检测日志流中的异常 | **可解释** —— 每个分数都能分解为具名信号 |
| 将相关日志行分组为**事件** | **稳健** —— 冷启动安全，在 DB/LLM 宕机时仍能生存 |
| 按重要性排序 + 识别**根本原因** | **可复现** —— 固定随机种子，幂等的运行 ID |
| 跨天链接事件（**因果链**） | **高适应性** —— 通过配置而非代码接入新日志源 |
| 带有按需摘要的分析师仪表板 | **在无故障日零误报** |
## 系统架构
系统以 **Docker Compose** 堆栈形式交付。流水线将运行范围内的状态写入 Postgres + Elasticsearch；仪表板从两者读取数据；LLM 摘要**按需从仪表板生成**，绝不在流水线中生成。
```
graph LR
subgraph Sources["Log Sources"]
SW[HPE CX Switches
syslog] UP[Analyst uploads
log files] end subgraph Stack["Docker Compose stack"] PIPE[Batch Pipeline
pipeline.py orchestrator] DB[(PostgreSQL
incidents · scores · history)] ES[(Elasticsearch
full-text log search)] DASH[Streamlit Dashboard
feed · detail · host health · search] end LLM[Groq LLM
llama-3.3-70b
on-demand summaries] SW --> PIPE UP --> DASH DASH --> PIPE PIPE --> DB PIPE --> ES DB --> DASH ES --> DASH DASH -. on demand .-> LLM LLM -. summary .-> DASH ``` ## 处理流水线 **线性数据流** —— 每个阶段读取前一个阶段的输出（磁盘上的 Parquet 文件，即“数据契约”），对其进行丰富，并写入自己的输出。这使得每个阶段都可以通过 `--from-step` 独立运行、测试和重启。 ``` graph TD RAW[Raw syslog / synthetic logs] --> P1 P1[1 · Parsing & Template Extraction
Drain-style templates + per-host sessionization] P1 --> S1[(sessionized_logs.parquet)] S1 --> P2[2 · Feature Engineering
statistical · temporal · severity · metric-trend] P2 --> S2[(features_df.parquet)] S2 --> P3[3 · ML Anomaly Detection
IsolationForest + Z-score hybrid · cold-start safe] P3 --> S3[(anomaly_df.parquet)] S1 --> P4[4 · Graph Correlation Engine
co-occurrence graph · PageRank centrality · sequences] P4 --> S4[(graph_scores_df.parquet)] S2 --> P5 S3 --> P5 S4 --> P5 P5[5 · Importance Scoring
fuse signals · labels · incident clustering · root cause] P5 --> S5[(scored_logs_df.parquet
root_causes_df.parquet)] S5 --> P55[5.5 · Cross-Run Correlation
Jaccard fingerprint match · causal chains · precursor boost] P55 --> S6[(incident_history.parquet)] P55 --> P6[6 · Storage & Query Layer
Postgres upsert + Elasticsearch index] P6 --> DBX[(PostgreSQL)] P6 --> ESX[(Elasticsearch)] DBX --> DASHX[Dashboard & LLM Summaries] ESX --> DASHX ``` ### 阶段职责 | # | 阶段 | 职责 | 关键输出 | |---|-------|----------------|-------------| | 1 | **解析** | Drain 式的 `template_id` 提取、token 规范化、`severity` 分配、基于主机的 `session_id` 分组（受间隔/持续时间/大小限制）。 | `template_id`, `session_id`, `severity` | | 2 | **特征提取** | 行为特征：频率、突发性、滚动 z-score 基线 (Welford)、到达间隔率、时间增量、计数器接近度、遥测下降率/利用率 + 滚动 OLS **斜率**（渐进漂移信号）。 | `frequency_score`, `burstiness_score`, `zscore_base`, `metric_slope_*` | | 3 | **ML 异常检测** | `StandardScaler → IsolationForest`（100 棵树，种子 42）与统计 z-score 混合，由**冷启动置信度**因子加权。 | `combined_score`, `is_anomaly`, `model_confidence` | | 4 | **图关联** | 模板的加权共现图；将 PageRank / 度 / 介数中心性作为结构重要性信号；检测循环出现的有序序列。 | `centrality_score`, `degree`, `in_sequence` | | 5 | **重要性评分** | 将三个信号融合为一个可解释的 `final_score`，映射到标签，**聚类**为事件，对**根本原因**候选者进行排序。 | `final_score`, `label`, `correlation_id`, `is_root_cause` | | 5.5 | **跨运行关联** | 对每个事件进行指纹识别，通过 Jaccard 相似度与 72 小时历史记录进行匹配，分配**链 ID**，并在出现后续严重事件时**提升**前兆事件的分数。 | `chain_id`, `precursor_incident_id`, `chain_confidence` | | 6 | **存储** | 将运行范围内的数据 upsert 到 Postgres（不断积累）+ 索引到 Elasticsearch 以进行全文搜索。 | Postgres 表 + ES 索引 | ## 核心评分模型 重要性是**三个独立信号的显式加权和** —— 而不是单一的黑盒分数。分析师看到的每个数字都可以追溯到某个具名信号。 ``` final_score = 0.50 · anomaly(ML) (behavioural anomaly — unsupervised IsolationForest) + 0.25 · centrality(graph) (structural importance — co-occurrence PageRank) + 0.25 · severity(rule) (declared severity — rule) ``` **为什么分成了三个独立的项？** 将严重性作为其自身可调的项（而不是将其输入到 IsolationForest 中）可以避免两个陷阱： 将严重性标签泄漏到无监督模型中，而该模型随后又*根据*严重性派生的真实情况进行验证，以及 对严重性进行重复计算。上层有两个修正层： - **严重性可信度门控** —— 如果某行日志的*消息*声明其状态正常（“…NORMAL”，“stable”，“within bounds”），则不能仅仅因为其*标签*显示为 `CRITICAL` 就获得严重性奖励。它只会**移除**不配获得的奖励。 - **严重性故障提升** —— 反之亦然：如果某行日志的*内容*包含明确的故障指示（“out of memory”，“split-brain”，“threshold exceeded”），则会被提升为可信的高严重性，以便它可以作为事件的种子，即使其标签为 `INFO`。 ## 从日志行到事件 每行日志的原始异常分数是噪声；分析师采取行动的对象是**事件**。 ``` graph TD A[Scored log rows] --> B{Interesting?
anomalous / labelled / high-severity} B -- no --> N[Background noise
never bridges incidents] B -- yes --> C[Seed] C --> D[Group seeds within a time window
gaps measured BETWEEN seeds] D --> E{Formation} E -- "density floor ≥10 seeds" --> F[Incident] E -- "severity formation
sparse severe cascade, two-tier gap" --> F F --> G{Escalation gate
severity-corroborated?} G -- yes --> H[Surfaced by default
is_escalated = true] G -- no --> I[Stored, not surfaced
recall preserved] H --> J[Root cause:
IDF specificity-weighted ranking] ``` - **基于异常种子的事件窗口化** —— 只有“有趣”的行才会成为*种子*；同一个时间窗口内的种子构成一个事件。连续的背景噪声无法桥接事件，因为间隔是在*种子之间*而不是所有日志之间测量的。 - **密度下限 + 严重性构建** —— 密集突发的 `medium` 行构成一个事件（基于密度），稀疏但严重的级联也构成一个事件（基于严重性），使用双层间隔，使得严重的级联保持完整，同时中等程度的链条保持紧凑。 - **升级门控** —— 所有事件都会被存储，但只有严重性得到证实的事件才会被标记为 `is_escalated` 并默认展示。它是一个**排名标志，而不是过滤器** → 真实事件在无故障日与噪声分离，*而不会*降低召回率。 - **根本原因** —— 基于 IDF 特异性加权排序，因此罕见的故障模板会胜过无处不在的心跳信号。 ## 跨信号关联与因果链 **单次运行内 —— 图关联** - 构建日志模板的加权共现图（时间窗口内的事件）。 - **PageRank 中心性** = 输入到 `final_score` 的结构重要性信号。 - 检测跨会话循环出现的有序序列，例如 `IF_DOWN → BGP_RESET → OSPF_LOST`。 **跨运行 —— 因果链** - 通过模板集对每个事件进行指纹识别；通过 **Jaccard 相似度 (≥ 0.3)** 与 72 小时历史记录进行匹配。 - 将匹配的事件跨天链接为**因果链** —— *“今天的故障是昨天的前兆。”* - 当发现后续的严重事件时，**提升前兆事件的分数**（追溯提升）。 - **运行范围内的存储键**（`run_id`, `sequence_number`），以便连续的批次不断积累，且永不冲突。`run_id` = 批次中最早事件的 `YYYYMMDD` —— 由数据决定，因此重新运行同一批次是幂等的。 ## 性能指标 **360° 留出评估** —— 跨越 10 天的 31,931 条日志（5 天异常 + 5 天无故障），**模型冻结**（真正的泛化测试）： | 指标 | 结果 | |---|---| | **升级后的事件精确率**（异常日） | **1.00** — 5/5 为真实事件 | | **在 19,622 条无故障日志上的错误升级** | **0** ✅ | | **事件信号召回率** | **0.857** | | k 处的排名召回率 | 0.76 | | 信号与噪声的分数分离度 | 0.37 | | 突发场景检测率 | 3/5 | 复现方式： ``` python scripts/run_360_eval.py ``` ## 快速开始 ``` # 启动基础设施（Postgres, Elasticsearch） docker compose up -d # 设置环境 cp .env.example .env # fill in credentials (incl. GROQ_API_KEY for summaries) python3 -m venv venv source venv/bin/activate pip install -r requirements.txt # 运行完整 pipeline（dry-run — 跳过 Postgres 写入） python pipeline.py --dry-run # 运行 Postgres + Elasticsearch 写入（需要 docker compose up） python pipeline.py # 从特定步骤重新开始（读取先前步骤已有的 parquets） python pipeline.py --dry-run --from-step scoring # 启动 dashboard streamlit run dashboard/app.py # or: docker compose up -d dashboard ``` 如果您有真实的 syslog 文件，请使用 `--log-file` 传递： ``` python pipeline.py --dry-run --log-file data/raw/cx_switches.log ``` 如果没有日志文件，流水线会自动生成合成的 CX 交换机数据。 ## 模块概述 | 文件夹 | 输出文件 | 描述 | |--------|-------------|-------------| | `ingestion/` | `data/raw/*.log` | 原始日志摄取（批量加载器或 Fluent Bit 流） | | `parsing/` | `data/processed/sessionized_logs.parquet` | Drain 式解析 + 基于主机的会话化 | | `features/` | `data/processed/features_df.parquet` | 统计 + 时间 + 严重性 + 指标趋势特征 | | `ml/` | `data/processed/anomaly_df.parquet` | IsolationForest + z-score 混合异常检测 | | `correlation/` | `data/processed/graph_scores_df.parquet` | 共现图，中心性评分，序列检测 | | `scoring/` | `data/processed/scored_logs_df.parquet` | 最终重要性评分、标签、事件聚类、根本原因 | | `storage/` | Postgres / Elasticsearch | 持久化 + 全文索引 | | `dashboard/` | Streamlit UI | 事件流、详情、主机健康、日志搜索、上传与分析 | | `evaluation/` | 指标报告 | 仅限内部 Oracle 的留出评估测试工具 | | `common/` | — | 共享配置、日志记录器、工具、schema 定义 | | `pipeline.py` | 以上所有 | 端到端编排器（`--dry-run`, `--from-step`） | 所有模块**均已完成**。请参阅 [`docs/LLD.md`](docs/LLD.md) 了解各模块的内部实现细节。 ## 数据契约 ### `sessionized_logs.parquet`（规范 —— 所有下游模块均从此处读取） | 列 | 类型 | 描述 | |--------|------|-------------| | `log_id` | str | 每行唯一标识符，例如 `log_000001` | | `raw_text` | str | 原始未解析的日志行 | | `timestamp` | datetime64 | UTC 日期时间 | | `source` | str | 主机名 / 设备 | | `session_id` | str | 会话组，例如 `session_0001` | | `template_id` | str | Drain 模板别名，例如 `IF_DOWN` | | `severity` | str | `CRITICAL` / `ERROR` / `WARN` / `INFO` | | `is_anomaly` | bool | 由异常检测器设置；从解析阶段出来时始终为 `False` | | `anomaly_label` | str | 仅当 `is_anomaly=True` 时非空 | ### `anomaly_df.parquet` `log_id` · `isolation_score` · `zscore_norm` · `combined_score` · `is_anomaly` · `model_confidence` ### `graph_scores_df.parquet` `log_id` · `centrality_score` (PageRank，主要图信号) · `degree` · `betweenness` `cluster_id` · `in_sequence` · `correlated_log_ids` ### `scored_logs_df.parquet` `log_id` · `final_score` · `label` · `correlation_id` (事件) · `is_root_cause` · `root_cause_confidence` ### 存储模式 (PostgreSQL) ``` erDiagram logs ||--o| features : "run_id, sequence_number" logs ||--o| anomalies : "run_id, sequence_number" logs ||--o| scores : "run_id, sequence_number" scores }o--|| incidents : "correlation_id = incident_id" incidents ||--o| summaries : "incident_id" incident_history ||--o{ incident_history : "precursor / chain" ``` ## 配置 所有可调参数都集中在 `common/config.py` 中；特定于数据集的词汇表（服务别名、噪声标签）位于 `config/dataset_profiles/` 下的 YAML 配置文件中 —— **接入新的日志源无需修改 Python 代码**。 可用配置文件：`generic.yaml`, `hpe_synthetic.yaml`, `kubernetes_linux.yaml`。 选定的常量： | 常量 | 默认值 | 描述 | |---|---|---| | `SCORING_ML_WEIGHT` | `0.50` | `final_score` 中 ML 异常信号的权重 | | `SCORING_GRAPH_WEIGHT` | `0.25` | 图中心性的权重 | | `SCORING_SEVERITY_WEIGHT` | `0.25` | 声明的严重性的权重 | | `IF_RANDOM_STATE` | `42` | IsolationForest 种子（用于复现） | | `CORRELATION_TIME_WINDOW_SECONDS` | `60` | 共现窗口宽度 | | `PAGERANK_ALPHA` | `0.85` | PageRank 阻尼系数 | | `SEQUENCE_MIN_LENGTH` / `SEQUENCE_MIN_SUPPORT` | `3` / `3` | 循环序列检测边界 | | `INCIDENT_MIN_SIZE` | `3` | 每个事件的最少种子数（消除单日志事件） | **环境变量 / 密钥**通过 `common/config.py` 延迟访问并快速失败 —— 绝不直接使用 `os.getenv()` 或 `dotenv`。缺失的变量会以 `[ENV ERROR] Missing required variable: NAME` 正常退出（无冗长的回溯信息）。密钥绝不会被打印。 ## 仪表板 一个完全从 Postgres + Elasticsearch 读取数据的 Streamlit 应用（绝不从流水线的内存状态读取）。打开 **[http://localhost:8501](http://localhost:8501)**。 | 页面 | 用途 | |------|---------| | **事件流** | 按严重性排序的升级事件。*无故障日什么也不显示 —— 这正是目的所在。* | | **事件详情** | 关联图、根本原因候选者 + 置信度、因果链，以及**按需生成的 Groq LLM 摘要**（首次打开时生成，然后缓存在 Postgres 中）。 | | **主机健康** | 各主机的活动和异常率。 | | **日志搜索** | 由 Elasticsearch 支持的全文搜索，带有主机/标签/时间过滤器和“跳转到事件”功能。 | | **上传与分析** | 分析师上传 syslog 文件 → 触发流水线运行 → 仅展示严重结果。 | LLM 摘要**从仪表板按需生成**，因此 API 开销取决于分析师实际打开的内容，而不是数据量。 ## 测试 各模块的 `tests/` 包在 `pytest` 下运行： ``` pytest # everything pytest correlation/tests/test_correlation.py -v # 70 correlation unit tests pytest ml/tests/test_anomaly.py pytest scoring/tests/test_scoring.py -v ``` ## 薄弱环节与未来计划 - **严重性系统性地分级偏低** —— 真实的 `CRITICAL` 经常处于 `medium` 级别（受限于 ML 权重上限）。检测是正确的；只是标签低了一个层级。 - **渐进漂移**（内存泄漏、热蠕变）比突发情况更难捕获 —— 检出率为 3/5。 - **仅含无故障日的基准胜过混合数据的训练** —— 能提供更清晰的决策边界。 **路线图** 1. 跨天的无故障基准 → 稳定、可分离的分数。 2. 分数稳定后重新调整 critical 级别的阈值。 3. 用于慢斜率故障的专用漂移检测模块。 ## 技术栈 | 层 | 技术 | |-------|-----------| | 语言 | Python 3 | | ML | scikit-learn (IsolationForest, StandardScaler), NumPy | | 日志解析 | Drain3 式模板提取 | | 图计算 | NetworkX | | 数据 | pandas + Parquet | | 存储 | PostgreSQL 15, Elasticsearch 8.11 | | 仪表板 | Streamlit + vis-network | | LLM | Groq API (`llama-3.3-70b-versatile`) | | 摄取 | 批量加载器 + Fluent Bit 配置 | | 编排 | `pipeline.py` + Docker Compose | ## 团队 Shreeraksha M · Sumukha Rao H · Ujwal Hegde · Sharva Dhanvi V · Vishon Dsouza
syslog] UP[Analyst uploads
log files] end subgraph Stack["Docker Compose stack"] PIPE[Batch Pipeline
pipeline.py orchestrator] DB[(PostgreSQL
incidents · scores · history)] ES[(Elasticsearch
full-text log search)] DASH[Streamlit Dashboard
feed · detail · host health · search] end LLM[Groq LLM
llama-3.3-70b
on-demand summaries] SW --> PIPE UP --> DASH DASH --> PIPE PIPE --> DB PIPE --> ES DB --> DASH ES --> DASH DASH -. on demand .-> LLM LLM -. summary .-> DASH ``` ## 处理流水线 **线性数据流** —— 每个阶段读取前一个阶段的输出（磁盘上的 Parquet 文件，即“数据契约”），对其进行丰富，并写入自己的输出。这使得每个阶段都可以通过 `--from-step` 独立运行、测试和重启。 ``` graph TD RAW[Raw syslog / synthetic logs] --> P1 P1[1 · Parsing & Template Extraction
Drain-style templates + per-host sessionization] P1 --> S1[(sessionized_logs.parquet)] S1 --> P2[2 · Feature Engineering
statistical · temporal · severity · metric-trend] P2 --> S2[(features_df.parquet)] S2 --> P3[3 · ML Anomaly Detection
IsolationForest + Z-score hybrid · cold-start safe] P3 --> S3[(anomaly_df.parquet)] S1 --> P4[4 · Graph Correlation Engine
co-occurrence graph · PageRank centrality · sequences] P4 --> S4[(graph_scores_df.parquet)] S2 --> P5 S3 --> P5 S4 --> P5 P5[5 · Importance Scoring
fuse signals · labels · incident clustering · root cause] P5 --> S5[(scored_logs_df.parquet
root_causes_df.parquet)] S5 --> P55[5.5 · Cross-Run Correlation
Jaccard fingerprint match · causal chains · precursor boost] P55 --> S6[(incident_history.parquet)] P55 --> P6[6 · Storage & Query Layer
Postgres upsert + Elasticsearch index] P6 --> DBX[(PostgreSQL)] P6 --> ESX[(Elasticsearch)] DBX --> DASHX[Dashboard & LLM Summaries] ESX --> DASHX ``` ### 阶段职责 | # | 阶段 | 职责 | 关键输出 | |---|-------|----------------|-------------| | 1 | **解析** | Drain 式的 `template_id` 提取、token 规范化、`severity` 分配、基于主机的 `session_id` 分组（受间隔/持续时间/大小限制）。 | `template_id`, `session_id`, `severity` | | 2 | **特征提取** | 行为特征：频率、突发性、滚动 z-score 基线 (Welford)、到达间隔率、时间增量、计数器接近度、遥测下降率/利用率 + 滚动 OLS **斜率**（渐进漂移信号）。 | `frequency_score`, `burstiness_score`, `zscore_base`, `metric_slope_*` | | 3 | **ML 异常检测** | `StandardScaler → IsolationForest`（100 棵树，种子 42）与统计 z-score 混合，由**冷启动置信度**因子加权。 | `combined_score`, `is_anomaly`, `model_confidence` | | 4 | **图关联** | 模板的加权共现图；将 PageRank / 度 / 介数中心性作为结构重要性信号；检测循环出现的有序序列。 | `centrality_score`, `degree`, `in_sequence` | | 5 | **重要性评分** | 将三个信号融合为一个可解释的 `final_score`，映射到标签，**聚类**为事件，对**根本原因**候选者进行排序。 | `final_score`, `label`, `correlation_id`, `is_root_cause` | | 5.5 | **跨运行关联** | 对每个事件进行指纹识别，通过 Jaccard 相似度与 72 小时历史记录进行匹配，分配**链 ID**，并在出现后续严重事件时**提升**前兆事件的分数。 | `chain_id`, `precursor_incident_id`, `chain_confidence` | | 6 | **存储** | 将运行范围内的数据 upsert 到 Postgres（不断积累）+ 索引到 Elasticsearch 以进行全文搜索。 | Postgres 表 + ES 索引 | ## 核心评分模型 重要性是**三个独立信号的显式加权和** —— 而不是单一的黑盒分数。分析师看到的每个数字都可以追溯到某个具名信号。 ``` final_score = 0.50 · anomaly(ML) (behavioural anomaly — unsupervised IsolationForest) + 0.25 · centrality(graph) (structural importance — co-occurrence PageRank) + 0.25 · severity(rule) (declared severity — rule) ``` **为什么分成了三个独立的项？** 将严重性作为其自身可调的项（而不是将其输入到 IsolationForest 中）可以避免两个陷阱： 将严重性标签泄漏到无监督模型中，而该模型随后又*根据*严重性派生的真实情况进行验证，以及 对严重性进行重复计算。上层有两个修正层： - **严重性可信度门控** —— 如果某行日志的*消息*声明其状态正常（“…NORMAL”，“stable”，“within bounds”），则不能仅仅因为其*标签*显示为 `CRITICAL` 就获得严重性奖励。它只会**移除**不配获得的奖励。 - **严重性故障提升** —— 反之亦然：如果某行日志的*内容*包含明确的故障指示（“out of memory”，“split-brain”，“threshold exceeded”），则会被提升为可信的高严重性，以便它可以作为事件的种子，即使其标签为 `INFO`。 ## 从日志行到事件 每行日志的原始异常分数是噪声；分析师采取行动的对象是**事件**。 ``` graph TD A[Scored log rows] --> B{Interesting?
anomalous / labelled / high-severity} B -- no --> N[Background noise
never bridges incidents] B -- yes --> C[Seed] C --> D[Group seeds within a time window
gaps measured BETWEEN seeds] D --> E{Formation} E -- "density floor ≥10 seeds" --> F[Incident] E -- "severity formation
sparse severe cascade, two-tier gap" --> F F --> G{Escalation gate
severity-corroborated?} G -- yes --> H[Surfaced by default
is_escalated = true] G -- no --> I[Stored, not surfaced
recall preserved] H --> J[Root cause:
IDF specificity-weighted ranking] ``` - **基于异常种子的事件窗口化** —— 只有“有趣”的行才会成为*种子*；同一个时间窗口内的种子构成一个事件。连续的背景噪声无法桥接事件，因为间隔是在*种子之间*而不是所有日志之间测量的。 - **密度下限 + 严重性构建** —— 密集突发的 `medium` 行构成一个事件（基于密度），稀疏但严重的级联也构成一个事件（基于严重性），使用双层间隔，使得严重的级联保持完整，同时中等程度的链条保持紧凑。 - **升级门控** —— 所有事件都会被存储，但只有严重性得到证实的事件才会被标记为 `is_escalated` 并默认展示。它是一个**排名标志，而不是过滤器** → 真实事件在无故障日与噪声分离，*而不会*降低召回率。 - **根本原因** —— 基于 IDF 特异性加权排序，因此罕见的故障模板会胜过无处不在的心跳信号。 ## 跨信号关联与因果链 **单次运行内 —— 图关联** - 构建日志模板的加权共现图（时间窗口内的事件）。 - **PageRank 中心性** = 输入到 `final_score` 的结构重要性信号。 - 检测跨会话循环出现的有序序列，例如 `IF_DOWN → BGP_RESET → OSPF_LOST`。 **跨运行 —— 因果链** - 通过模板集对每个事件进行指纹识别；通过 **Jaccard 相似度 (≥ 0.3)** 与 72 小时历史记录进行匹配。 - 将匹配的事件跨天链接为**因果链** —— *“今天的故障是昨天的前兆。”* - 当发现后续的严重事件时，**提升前兆事件的分数**（追溯提升）。 - **运行范围内的存储键**（`run_id`, `sequence_number`），以便连续的批次不断积累，且永不冲突。`run_id` = 批次中最早事件的 `YYYYMMDD` —— 由数据决定，因此重新运行同一批次是幂等的。 ## 性能指标 **360° 留出评估** —— 跨越 10 天的 31,931 条日志（5 天异常 + 5 天无故障），**模型冻结**（真正的泛化测试）： | 指标 | 结果 | |---|---| | **升级后的事件精确率**（异常日） | **1.00** — 5/5 为真实事件 | | **在 19,622 条无故障日志上的错误升级** | **0** ✅ | | **事件信号召回率** | **0.857** | | k 处的排名召回率 | 0.76 | | 信号与噪声的分数分离度 | 0.37 | | 突发场景检测率 | 3/5 | 复现方式： ``` python scripts/run_360_eval.py ``` ## 快速开始 ``` # 启动基础设施（Postgres, Elasticsearch） docker compose up -d # 设置环境 cp .env.example .env # fill in credentials (incl. GROQ_API_KEY for summaries) python3 -m venv venv source venv/bin/activate pip install -r requirements.txt # 运行完整 pipeline（dry-run — 跳过 Postgres 写入） python pipeline.py --dry-run # 运行 Postgres + Elasticsearch 写入（需要 docker compose up） python pipeline.py # 从特定步骤重新开始（读取先前步骤已有的 parquets） python pipeline.py --dry-run --from-step scoring # 启动 dashboard streamlit run dashboard/app.py # or: docker compose up -d dashboard ``` 如果您有真实的 syslog 文件，请使用 `--log-file` 传递： ``` python pipeline.py --dry-run --log-file data/raw/cx_switches.log ``` 如果没有日志文件，流水线会自动生成合成的 CX 交换机数据。 ## 模块概述 | 文件夹 | 输出文件 | 描述 | |--------|-------------|-------------| | `ingestion/` | `data/raw/*.log` | 原始日志摄取（批量加载器或 Fluent Bit 流） | | `parsing/` | `data/processed/sessionized_logs.parquet` | Drain 式解析 + 基于主机的会话化 | | `features/` | `data/processed/features_df.parquet` | 统计 + 时间 + 严重性 + 指标趋势特征 | | `ml/` | `data/processed/anomaly_df.parquet` | IsolationForest + z-score 混合异常检测 | | `correlation/` | `data/processed/graph_scores_df.parquet` | 共现图，中心性评分，序列检测 | | `scoring/` | `data/processed/scored_logs_df.parquet` | 最终重要性评分、标签、事件聚类、根本原因 | | `storage/` | Postgres / Elasticsearch | 持久化 + 全文索引 | | `dashboard/` | Streamlit UI | 事件流、详情、主机健康、日志搜索、上传与分析 | | `evaluation/` | 指标报告 | 仅限内部 Oracle 的留出评估测试工具 | | `common/` | — | 共享配置、日志记录器、工具、schema 定义 | | `pipeline.py` | 以上所有 | 端到端编排器（`--dry-run`, `--from-step`） | 所有模块**均已完成**。请参阅 [`docs/LLD.md`](docs/LLD.md) 了解各模块的内部实现细节。 ## 数据契约 ### `sessionized_logs.parquet`（规范 —— 所有下游模块均从此处读取） | 列 | 类型 | 描述 | |--------|------|-------------| | `log_id` | str | 每行唯一标识符，例如 `log_000001` | | `raw_text` | str | 原始未解析的日志行 | | `timestamp` | datetime64 | UTC 日期时间 | | `source` | str | 主机名 / 设备 | | `session_id` | str | 会话组，例如 `session_0001` | | `template_id` | str | Drain 模板别名，例如 `IF_DOWN` | | `severity` | str | `CRITICAL` / `ERROR` / `WARN` / `INFO` | | `is_anomaly` | bool | 由异常检测器设置；从解析阶段出来时始终为 `False` | | `anomaly_label` | str | 仅当 `is_anomaly=True` 时非空 | ### `anomaly_df.parquet` `log_id` · `isolation_score` · `zscore_norm` · `combined_score` · `is_anomaly` · `model_confidence` ### `graph_scores_df.parquet` `log_id` · `centrality_score` (PageRank，主要图信号) · `degree` · `betweenness` `cluster_id` · `in_sequence` · `correlated_log_ids` ### `scored_logs_df.parquet` `log_id` · `final_score` · `label` · `correlation_id` (事件) · `is_root_cause` · `root_cause_confidence` ### 存储模式 (PostgreSQL) ``` erDiagram logs ||--o| features : "run_id, sequence_number" logs ||--o| anomalies : "run_id, sequence_number" logs ||--o| scores : "run_id, sequence_number" scores }o--|| incidents : "correlation_id = incident_id" incidents ||--o| summaries : "incident_id" incident_history ||--o{ incident_history : "precursor / chain" ``` ## 配置 所有可调参数都集中在 `common/config.py` 中；特定于数据集的词汇表（服务别名、噪声标签）位于 `config/dataset_profiles/` 下的 YAML 配置文件中 —— **接入新的日志源无需修改 Python 代码**。 可用配置文件：`generic.yaml`, `hpe_synthetic.yaml`, `kubernetes_linux.yaml`。 选定的常量： | 常量 | 默认值 | 描述 | |---|---|---| | `SCORING_ML_WEIGHT` | `0.50` | `final_score` 中 ML 异常信号的权重 | | `SCORING_GRAPH_WEIGHT` | `0.25` | 图中心性的权重 | | `SCORING_SEVERITY_WEIGHT` | `0.25` | 声明的严重性的权重 | | `IF_RANDOM_STATE` | `42` | IsolationForest 种子（用于复现） | | `CORRELATION_TIME_WINDOW_SECONDS` | `60` | 共现窗口宽度 | | `PAGERANK_ALPHA` | `0.85` | PageRank 阻尼系数 | | `SEQUENCE_MIN_LENGTH` / `SEQUENCE_MIN_SUPPORT` | `3` / `3` | 循环序列检测边界 | | `INCIDENT_MIN_SIZE` | `3` | 每个事件的最少种子数（消除单日志事件） | **环境变量 / 密钥**通过 `common/config.py` 延迟访问并快速失败 —— 绝不直接使用 `os.getenv()` 或 `dotenv`。缺失的变量会以 `[ENV ERROR] Missing required variable: NAME` 正常退出（无冗长的回溯信息）。密钥绝不会被打印。 ## 仪表板 一个完全从 Postgres + Elasticsearch 读取数据的 Streamlit 应用（绝不从流水线的内存状态读取）。打开 **[http://localhost:8501](http://localhost:8501)**。 | 页面 | 用途 | |------|---------| | **事件流** | 按严重性排序的升级事件。*无故障日什么也不显示 —— 这正是目的所在。* | | **事件详情** | 关联图、根本原因候选者 + 置信度、因果链，以及**按需生成的 Groq LLM 摘要**（首次打开时生成，然后缓存在 Postgres 中）。 | | **主机健康** | 各主机的活动和异常率。 | | **日志搜索** | 由 Elasticsearch 支持的全文搜索，带有主机/标签/时间过滤器和“跳转到事件”功能。 | | **上传与分析** | 分析师上传 syslog 文件 → 触发流水线运行 → 仅展示严重结果。 | LLM 摘要**从仪表板按需生成**，因此 API 开销取决于分析师实际打开的内容，而不是数据量。 ## 测试 各模块的 `tests/` 包在 `pytest` 下运行： ``` pytest # everything pytest correlation/tests/test_correlation.py -v # 70 correlation unit tests pytest ml/tests/test_anomaly.py pytest scoring/tests/test_scoring.py -v ``` ## 薄弱环节与未来计划 - **严重性系统性地分级偏低** —— 真实的 `CRITICAL` 经常处于 `medium` 级别（受限于 ML 权重上限）。检测是正确的；只是标签低了一个层级。 - **渐进漂移**（内存泄漏、热蠕变）比突发情况更难捕获 —— 检出率为 3/5。 - **仅含无故障日的基准胜过混合数据的训练** —— 能提供更清晰的决策边界。 **路线图** 1. 跨天的无故障基准 → 稳定、可分离的分数。 2. 分数稳定后重新调整 critical 级别的阈值。 3. 用于慢斜率故障的专用漂移检测模块。 ## 技术栈 | 层 | 技术 | |-------|-----------| | 语言 | Python 3 | | ML | scikit-learn (IsolationForest, StandardScaler), NumPy | | 日志解析 | Drain3 式模板提取 | | 图计算 | NetworkX | | 数据 | pandas + Parquet | | 存储 | PostgreSQL 15, Elasticsearch 8.11 | | 仪表板 | Streamlit + vis-network | | LLM | Groq API (`llama-3.3-70b-versatile`) | | 摄取 | 批量加载器 + Fluent Bit 配置 | | 编排 | `pipeline.py` + Docker Compose | ## 团队 Shreeraksha M · Sumukha Rao H · Ujwal Hegde · Sharva Dhanvi V · Vishon Dsouza
标签：Apex, Kubernetes, Streamlit, 异常检测, 机器学习, 根因分析, 测试用例, 版权保护, 访问控制, 运维监控, 逆向工具