88 lines
3.3 KiB
Python
88 lines
3.3 KiB
Python
import _bootstrap # noqa: F401
|
|
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
from fault_log_analyzer.models import Event, FaultLog
|
|
from fault_log_analyzer.root_cause import RootCauseAnalyzer, RootCauseRule
|
|
|
|
|
|
def make_log(message="No space left on device", fid="fl-1", host="h-1", trace_id=""):
|
|
return FaultLog(
|
|
fault_log_id=fid,
|
|
host_id=host,
|
|
fingerprint="fp",
|
|
level="ERROR",
|
|
message=message,
|
|
occurred_at=datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc),
|
|
trace_id=trace_id,
|
|
)
|
|
|
|
|
|
def make_event(event_id="e-1", host="h-1", metric="disk_used_percent", value=97.2, minute=1):
|
|
return Event(
|
|
event_id=event_id,
|
|
host_id=host,
|
|
metric=metric,
|
|
value=value,
|
|
fired_at=datetime(2025, 1, 1, 10, minute, 0, tzinfo=timezone.utc),
|
|
)
|
|
|
|
|
|
class TestRootCauseRule(unittest.TestCase):
|
|
def test_match_log(self):
|
|
rule = RootCauseRule("disk_full", r"no space left on device")
|
|
self.assertTrue(rule.match_log("No space left on device"))
|
|
self.assertFalse(rule.match_log("connection refused"))
|
|
|
|
def test_match_metric_threshold(self):
|
|
rule = RootCauseRule("disk_full", metric_pattern=r"disk.*used", metric_threshold=90.0)
|
|
self.assertTrue(rule.match_metric("disk_used_percent", 95.0))
|
|
self.assertFalse(rule.match_metric("disk_used_percent", 80.0))
|
|
self.assertFalse(rule.match_metric("cpu_usage", 95.0))
|
|
|
|
|
|
class TestRootCauseAnalyzer(unittest.TestCase):
|
|
def test_disk_full_with_metric_event(self):
|
|
rc = RootCauseAnalyzer().analyze(
|
|
make_log("No space left on device"),
|
|
[make_event()],
|
|
)
|
|
self.assertEqual(rc.cause_type, "disk_full")
|
|
self.assertGreaterEqual(rc.confidence, 0.95)
|
|
self.assertTrue(any(e.type == "event" for e in rc.evidence))
|
|
self.assertTrue(any(e.type == "log" for e in rc.evidence))
|
|
|
|
def test_rule_without_matching_event(self):
|
|
rc = RootCauseAnalyzer().analyze(
|
|
make_log("No space left on device"),
|
|
[make_event(metric="cpu_usage", value=80.0)],
|
|
)
|
|
self.assertEqual(rc.cause_type, "disk_full")
|
|
# 日志规则命中但指标事件不匹配，置信度为基础值
|
|
self.assertAlmostEqual(rc.confidence, 0.95)
|
|
|
|
def test_unknown_with_related_event(self):
|
|
rc = RootCauseAnalyzer().analyze(
|
|
make_log("mysterious failure"),
|
|
[make_event(metric="cpu_usage", value=50.0)],
|
|
)
|
|
self.assertEqual(rc.cause_type, "related_cpu_usage")
|
|
self.assertEqual(rc.confidence, 0.4)
|
|
|
|
def test_unknown_with_trace_only(self):
|
|
rc = RootCauseAnalyzer().analyze(make_log("mysterious failure", trace_id="t-1"), [])
|
|
self.assertEqual(rc.cause_type, "trace_linked")
|
|
self.assertEqual(rc.confidence, 0.3)
|
|
|
|
def test_event_out_of_window_ignored(self):
|
|
old_event = make_event(event_id="e-old", minute=20) # 超出 ±5 分钟窗口
|
|
rc = RootCauseAnalyzer().analyze(make_log("No space left on device"), [old_event])
|
|
# 日志规则仍命中，但没有指标事件证据增强
|
|
self.assertEqual(rc.cause_type, "disk_full")
|
|
self.assertFalse(any(e.type == "event" for e in rc.evidence))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|