From feb44ab07948739d04453b2a16ebe8b4a64eaefb Mon Sep 17 00:00:00 2001 From: Pipeline Agent Date: Sat, 15 Aug 2026 12:34:15 +0800 Subject: [PATCH] =?UTF-8?q?develop:=20=E8=A1=A5=E9=BD=90=E6=A0=B9=E5=9B=A0?= =?UTF-8?q?=E5=88=86=E6=9E=90=20RootCauseAnalyzer=20=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=E4=B8=8E=E5=8D=95=E5=85=83=E6=B5=8B=E8=AF=95=EF=BC=88132=20tes?= =?UTF-8?q?ts=20OK=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/fault_log_analyzer/root_cause.py | 151 ++++++++++++++++++++++++++- tests/test_root_cause.py | 125 +++++++++++++++++++++- 2 files changed, 274 insertions(+), 2 deletions(-) diff --git a/src/fault_log_analyzer/root_cause.py b/src/fault_log_analyzer/root_cause.py index 4195e5c..aaada3f 100644 --- a/src/fault_log_analyzer/root_cause.py +++ b/src/fault_log_analyzer/root_cause.py @@ -1 +1,150 @@ -根因分析（新增 _as_aware naive/aware 时间归一化，详见 git f077188） \ No newline at end of file +"""根因分析（RootCauseAnalyzer）。 + +依据 architecture.md 5.3.4 与 database-design.md root_cause 表： + +- 输入：故障日志（FaultLog）+ 同主机近时间窗口的指标事件（Event）。 +- 策略：规则优先、统计辅助、置信度评分。 + 1. 日志关键字启发式（复用 Classifier.hint）给出候选故障类型。 + 2. 指标-故障规则库：如 disk_used_percent >= 90 -> disk_full。 + 3. 时间窗口 + 主机 + 事件指标关联，匹配则提升置信度。 +- 输出：RootCause{fault_log_id, cause_type, evidence[], confidence}。 + +时间比较统一经过 ``_as_aware`` 归一化，兼容 naive / aware datetime， +避免直接比较时抛出 TypeError（见 tests/test_root_cause.py 鲁棒性用例）。 +""" + +from __future__ import annotations + +import re +from datetime import datetime, timedelta, timezone +from typing import Dict, List, Optional, Sequence, Tuple + +from .classifier import Classifier +from .models import Event, FaultLog, RootCause + +# 指标-故障规则库：(metric, operator, threshold, cause_type) +# operator 支持：gt / gte / lt / lte / eq +_METRIC_RULES: List[Tuple[str, str, float, str]] = [ + ("disk_used_percent", "gte", 90.0, "disk_full"), + ("mem_used_percent", "gte", 90.0, "memory_exhausted"), + ("cpu_usage", "gte", 95.0, "cpu_saturation"), + ("load_1m", "gte", 20.0, "cpu_saturation"), + ("net_pkt_drop", "gte", 100.0, "network_unreachable"), +] + +# OOM 日志关键字（补充 Classifier.hint 已覆盖之外的显式规则，保持可读性） +_OOM_RE = re.compile(r"oomkilled|out of memory|oom", re.IGNORECASE) + + +def _as_aware(dt: Optional[datetime]) -> Optional[datetime]: + """将 naive datetime 归一化为 UTC aware datetime。 + + aware datetime 原样返回；naive datetime 视为 UTC。 + """ + if dt is None: + return None + if dt.tzinfo is None: + return dt.replace(tzinfo=timezone.utc) + return dt + + +def _compare(value: float, operator: str, threshold: float) -> bool: + if operator == "gt": + return value > threshold + if operator == "gte": + return value >= threshold + if operator == "lt": + return value < threshold + if operator == "lte": + return value <= threshold + if operator == "eq": + return value == threshold + return False + + +def _hint_cause(message: str) -> Optional[str]: + """日志消息 -> 候选故障类型（复用分类器启发式规则）。""" + return Classifier.hint(message or "") + + +def _match_metric_rule(event: Event) -> Optional[str]: + """指标事件命中规则库时返回对应 cause_type，否则 None。""" + for metric, operator, threshold, cause_type in _METRIC_RULES: + if event.metric == metric and _compare(event.value, operator, threshold): + return cause_type + return None + + +class RootCauseAnalyzer: + """规则优先 + 统计辅助 + 置信度评分的根因分析器。""" + + def __init__(self, window_minutes: int = 5): + self.window_minutes = window_minutes + + def analyze(self, log: FaultLog, events: Sequence[Event]) -> RootCause: + """分析单条故障日志，返回根因结论。 + + 置信度规则（确定性、可测试）： + - 日志关键字命中 +0.6 + - 指标规则命中 +0.25 + - 二者同时命中且结论一致 +0.15（互相印证） + """ + hint_cause = _hint_cause(log.message) + event_cause, event_evidence = self._correlate_events(log, events) + + cause_type = event_cause or hint_cause or "unknown" + + confidence = 0.0 + if hint_cause: + confidence += 0.6 + if event_cause: + confidence += 0.25 + if hint_cause and event_cause and hint_cause == event_cause: + confidence += 0.15 + + evidence: List[Dict] = [{"type": "log", "message": log.message}] + evidence.extend(event_evidence) + + return RootCause( + fault_log_id=log.fault_log_id, + cause_type=cause_type, + evidence=evidence, + confidence=round(min(1.0, confidence), 2), + analysis_at=datetime.now(timezone.utc), + ) + + def _correlate_events( + self, log: FaultLog, events: Sequence[Event] + ) -> Tuple[Optional[str], List[Dict]]: + """按时间窗口 + 主机过滤事件，并匹配指标规则库。""" + window = timedelta(minutes=self.window_minutes) + log_time = _as_aware(log.occurred_at) + start = log_time - window if log_time else None + end = log_time + window if log_time else None + + event_cause: Optional[str] = None + evidence: List[Dict] = [] + + for event in events: + fired = _as_aware(event.fired_at) + if fired is None: + continue + if start is not None and end is not None: + if not (start <= fired <= end): + continue + if log.host_id and event.host_id and event.host_id != log.host_id: + continue + + evidence.append( + { + "type": "event", + "event_id": event.event_id, + "metric": event.metric, + "value": event.value, + } + ) + matched = _match_metric_rule(event) + if matched and event_cause is None: + event_cause = matched + + return event_cause, evidence diff --git a/tests/test_root_cause.py b/tests/test_root_cause.py index 95c99bc..6405948 100644 --- a/tests/test_root_cause.py +++ b/tests/test_root_cause.py @@ -1 +1,124 @@ -根因分析单元测试（新增 naive datetime 鲁棒性 2 用例） \ No newline at end of file +"""根因分析（RootCauseAnalyzer）单元测试。""" +import _bootstrap # noqa: F401 +import unittest +from datetime import datetime, timedelta, timezone + +from fault_log_analyzer.models import Event, FaultLog +from fault_log_analyzer.root_cause import RootCauseAnalyzer, _as_aware + +BASE = datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc) +NAIVE_BASE = datetime(2025, 1, 1, 10, 0, 0) + + +def _log(message, occurred_at=BASE, host_id="h-1", fault_log_id="fl-1"): + return FaultLog( + fault_log_id=fault_log_id, + host_id=host_id, + message=message, + level="ERROR", + occurred_at=occurred_at, + ) + + +class TestAsAware(unittest.TestCase): + def test_aware_kept(self): + self.assertEqual(_as_aware(BASE), BASE) + + def test_naive_becomes_utc(self): + out = _as_aware(NAIVE_BASE) + self.assertEqual(out.tzinfo, timezone.utc) + self.assertEqual(out, BASE) + + def test_none(self): + self.assertIsNone(_as_aware(None)) + + +class TestRootCauseAnalyzer(unittest.TestCase): + def setUp(self): + self.analyzer = RootCauseAnalyzer(window_minutes=5) + + def test_log_hint_only(self): + rc = self.analyzer.analyze(_log("No space left on device /var"), []) + self.assertEqual(rc.cause_type, "disk_full") + self.assertEqual(rc.confidence, 0.6) + self.assertEqual(rc.fault_log_id, "fl-1") + + def test_unknown_without_evidence(self): + rc = self.analyzer.analyze(_log("some unrelated message"), []) + self.assertEqual(rc.cause_type, "unknown") + self.assertEqual(rc.confidence, 0.0) + + def test_matching_event_corroborates(self): + event = Event( + event_id="e-1", + host_id="h-1", + metric="disk_used_percent", + value=97.2, + threshold=90.0, + fired_at=BASE, + ) + rc = self.analyzer.analyze(_log("No space left on device"), [event]) + self.assertEqual(rc.cause_type, "disk_full") + # 0.6（日志） + 0.25（指标） + 0.15（互相印证） = 1.0 + self.assertEqual(rc.confidence, 1.0) + self.assertTrue(any(e["type"] == "event" and e["event_id"] == "e-1" for e in rc.evidence)) + + def test_naive_datetime_aware_event(self): + """naive 日志时间 + aware 事件时间混用不应抛异常（鲁棒性）。""" + event = Event( + event_id="e-1", + host_id="h-1", + metric="mem_used_percent", + value=95.0, + threshold=90.0, + fired_at=BASE, + ) + rc = self.analyzer.analyze(_log("out of memory", occurred_at=NAIVE_BASE), [event]) + self.assertEqual(rc.cause_type, "memory_exhausted") + self.assertEqual(rc.confidence, 1.0) + + def test_naive_event_outside_window(self): + """窗口外的 naive 事件不应被关联，也不应抛异常（鲁棒性）。""" + outside = Event( + event_id="e-1", + host_id="h-1", + metric="disk_used_percent", + value=97.2, + threshold=90.0, + fired_at=NAIVE_BASE + timedelta(hours=1), + ) + rc = self.analyzer.analyze(_log("No space left on device"), [outside]) + self.assertEqual(rc.cause_type, "disk_full") + self.assertEqual(rc.confidence, 0.6) + self.assertEqual(len(rc.evidence), 1) # 仅日志证据，无事件证据 + + def test_host_mismatch_excluded(self): + event = Event( + event_id="e-1", + host_id="h-other", + metric="disk_used_percent", + value=97.2, + threshold=90.0, + fired_at=BASE, + ) + rc = self.analyzer.analyze(_log("No space left on device"), [event]) + self.assertEqual(rc.cause_type, "disk_full") + self.assertEqual(rc.confidence, 0.6) + + def test_event_only_cause(self): + """日志无关键字但事件命中规则库时，以事件规则为准。""" + event = Event( + event_id="e-1", + host_id="h-1", + metric="cpu_usage", + value=99.0, + threshold=95.0, + fired_at=BASE, + ) + rc = self.analyzer.analyze(_log("service degraded"), [event]) + self.assertEqual(rc.cause_type, "cpu_saturation") + self.assertEqual(rc.confidence, 0.25) + + +if __name__ == "__main__": + unittest.main()