yumoqing/fault-log-analyzer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

develop: 补齐 threshold-event-detector 规范交付件并清理交换文件

Browse Source
This commit is contained in:
Pipeline Agent 2026-08-15 11:22:39 +08:00
parent 70614776be
commit c62453ef70
9 changed files with 308 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/fault_log_analyzer/api.py
View File

@ -6,9 +6,7 @@ host-monitor 网关统一转发到本服务（或替换为 FastAPI 等实现）
from __future__ import annotations
import json
from datetime import datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Optional
from urllib.parse import urlparse, parse_qs
from .models import FaultFilterRule, FaultType, from_iso
@ -189,7 +187,8 @@ class FaultLogApiServer:
handler = create_handler(fault_log_repo, root_cause_repo, fault_type_repo, filter_rule_repo)
self._httpd = ThreadingHTTPServer((host, port), handler)
self.host = host
self.port = port
# port=0 时使用操作系统分配的实际端口
self.port = self._httpd.server_address[1]
def start(self, block: bool = False) -> None:
if block:

6
src/fault_log_analyzer/models.py
View File

@ -20,12 +20,16 @@ def to_iso(dt: Optional[datetime]) -> Optional[str]:
def from_iso(value: Optional[str]) -> Optional[datetime]:
"""解析 ISO 时间字符串，非法输入返回 None（不抛异常）。"""
if value is None or value == "":
return None
normalized = value.strip()
if normalized.endswith("Z"):
normalized = normalized[:-1] + "+00:00"
dt = datetime.fromisoformat(normalized)
try:
dt = datetime.fromisoformat(normalized)
except (ValueError, TypeError):
return None
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)

20
src/fault_log_analyzer/parser.py
View File

@ -3,7 +3,7 @@ from __future__ import annotations
import json
import re
from datetime import datetime
from datetime import datetime, timezone
from typing import Any, Optional
from .models import LogEntry, from_iso, utcnow
@ -42,22 +42,30 @@ def _find_value(fields: dict[str, Any], keys: tuple[str, ...]) -> Optional[str]:
return None
def _as_utc(dt: datetime) -> datetime:
"""统一转为带时区的 UTC 时间，避免 naive/aware 混用。"""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
def parse_timestamp(value: Any) -> Optional[datetime]:
"""解析多种时间戳表达，失败返回 None。"""
if value is None:
return None
if isinstance(value, (int, float)):
# 秒/毫秒/纳秒时间戳
# 秒/毫秒/纳秒时间戳。阈值按数量级划分：
# 秒 ~1e9、毫秒 ~1e12、纳秒 ~1e18。
if value > 1e17: # 纳秒
value = value / 1e9
elif value > 1e14: # 毫秒
elif value > 1e11: # 毫秒
value = value / 1e3
try:
return datetime.utcfromtimestamp(value)
return datetime.fromtimestamp(value, tz=timezone.utc)
except (ValueError, OSError, OverflowError):
return None
if isinstance(value, datetime):
return value
return _as_utc(value)
text = str(value).strip()
if not text:
return None
@ -66,7 +74,7 @@ def parse_timestamp(value: Any) -> Optional[datetime]:
return parsed
for fmt in _TIMESTAMP_FORMATS:
try:
return datetime.strptime(text, fmt)
return _as_utc(datetime.strptime(text, fmt))
except ValueError:
continue
return None

14
tests/_bootstrap.py Normal file
View File

@ -0,0 +1,14 @@
"""测试引导模块。
将 ``src`` 目录加入 ``sys.path``，使 ``python -m unittest discover -s tests -v``
在未安装包的情况下也能直接运行（src-layout 项目无需额外设置 PYTHONPATH）。
每个测试模块顶部 ``import _bootstrap`` 即可。
"""
from __future__ import annotations
import os
import sys
_SRC = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)

42
tests/test_config.py Normal file
View File

@ -0,0 +1,42 @@
import _bootstrap # noqa: F401
import unittest
from unittest.mock import patch
from fault_log_analyzer.config import Config
class TestConfig(unittest.TestCase):
def test_defaults(self):
with patch.dict("os.environ", {}, clear=True):
cfg = Config.from_env()
self.assertEqual(cfg.kafka_logs_raw_topic, "logs.raw")
self.assertEqual(cfg.kafka_logs_fault_topic, "logs.fault")
self.assertEqual(cfg.cluster_eps, 0.75)
self.assertEqual(cfg.cluster_min_samples, 5)
self.assertEqual(cfg.storage, "memory")
self.assertEqual(cfg.api_port, 8080)
def test_env_overrides(self):
env = {
"KAFKA_LOGS_RAW_TOPIC": "my.logs",
"CLUSTER_EPS": "0.5",
"CLUSTER_MIN_SAMPLES": "3",
"API_PORT": "9090",
"STORAGE": "mysql",
"ES_HOSTS": "http://es1:9200, http://es2:9200",
"CLUSTER_USE_SKLEARN": "true",
}
with patch.dict("os.environ", env, clear=True):
cfg = Config.from_env()
self.assertEqual(cfg.kafka_logs_raw_topic, "my.logs")
self.assertEqual(cfg.cluster_eps, 0.5)
self.assertEqual(cfg.cluster_min_samples, 3)
self.assertEqual(cfg.api_port, 9090)
self.assertEqual(cfg.storage, "mysql")
self.assertEqual(cfg.es_hosts, ["http://es1:9200", "http://es2:9200"])
self.assertTrue(cfg.cluster_use_sklearn)
if __name__ == "__main__":
unittest.main()

58
tests/test_filters.py Normal file
View File

@ -0,0 +1,58 @@
import _bootstrap # noqa: F401
import unittest
from datetime import datetime, timezone
from fault_log_analyzer.filters import CaptureFilter
from fault_log_analyzer.models import FaultFilterRule, LogEntry
def entry(level="ERROR", message="something", host_id="h-1"):
return LogEntry(
timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc),
level=level,
message=message,
host_id=host_id,
)
class TestCaptureFilter(unittest.TestCase):
def test_default_levels(self):
f = CaptureFilter()
self.assertTrue(f.is_fault(entry(level="ERROR")))
self.assertTrue(f.is_fault(entry(level="FATAL")))
self.assertTrue(f.is_fault(entry(level="CRITICAL")))
self.assertFalse(f.is_fault(entry(level="WARN")))
self.assertFalse(f.is_fault(entry(level="INFO")))
def test_rule_pattern_hit(self):
f = CaptureFilter([FaultFilterRule(name="disk", level="ERROR", pattern="disk full")])
self.assertTrue(f.is_fault(entry(message="No space, disk full on device")))
self.assertFalse(f.is_fault(entry(message="connection refused")))
def test_rule_exclude_pattern(self):
f = CaptureFilter(
[FaultFilterRule(name="disk", level="ERROR", pattern="disk", exclude_pattern="ignore")]
)
self.assertTrue(f.is_fault(entry(message="disk full")))
self.assertFalse(f.is_fault(entry(message="disk ignore")))
def test_rule_level_mismatch(self):
f = CaptureFilter([FaultFilterRule(name="disk", level="FATAL", pattern="disk")])
self.assertFalse(f.is_fault(entry(level="ERROR", message="disk full")))
def test_configured_rules_do_not_fallback(self):
# 配置了规则但均未命中时不走默认级别兜底
f = CaptureFilter([FaultFilterRule(name="disk", level="ERROR", pattern="disk full")])
self.assertFalse(f.is_fault(entry(level="FATAL", message="connection refused")))
def test_disabled_rule_falls_back_to_default(self):
f = CaptureFilter(
[FaultFilterRule(name="disk", level="ERROR", pattern="disk", enabled=False)]
)
self.assertTrue(f.is_fault(entry(level="ERROR", message="anything")))
self.assertFalse(f.is_fault(entry(level="INFO", message="anything")))
if __name__ == "__main__":
unittest.main()

62
tests/test_fingerprint.py Normal file
View File

@ -0,0 +1,62 @@
import _bootstrap # noqa: F401
import unittest
from fault_log_analyzer.fingerprint import (
MinHash,
message_fingerprint,
sha1_hex,
shingles,
)
class TestFingerprint(unittest.TestCase):
def test_sha1_hex_length(self):
self.assertEqual(len(sha1_hex("abc", 16)), 16)
def test_same_template_same_fingerprint(self):
f1 = message_fingerprint("Connection from 10.0.0.11 failed after 3 retries")
f2 = message_fingerprint("Connection from 10.0.0.12 failed after 5 retries")
self.assertEqual(f1, f2)
def test_different_message_different_fingerprint(self):
self.assertNotEqual(message_fingerprint("disk full"), message_fingerprint("connection refused"))
def test_shingles(self):
s = shingles(["a", "b", "c", "d"], k=3)
self.assertIn("a|b|c", s)
self.assertIn("b|c|d", s)
self.assertNotIn("c|d|e", s)
def test_shingles_short_sequence(self):
s = shingles(["a", "b"], k=3)
self.assertIn("a|b", s)
def test_shingles_empty(self):
self.assertEqual(shingles([], k=3), set())
class TestMinHash(unittest.TestCase):
def test_signature_length(self):
mh = MinHash(num_hashes=16)
sig = mh.signature(["a", "b", "c"])
self.assertEqual(len(sig), 16)
def test_signature_empty(self):
mh = MinHash(num_hashes=8)
sig = mh.signature([])
self.assertEqual(sig, [0] * 8)
def test_jaccard_estimate_identical(self):
mh = MinHash(num_hashes=32)
sig = mh.signature(["a", "b", "c", "d"])
self.assertEqual(mh.jaccard_estimate(sig, sig), 1.0)
def test_jaccard_estimate_length_mismatch(self):
mh = MinHash(num_hashes=8)
with self.assertRaises(ValueError):
mh.jaccard_estimate([1, 2], [1, 2, 3])
if __name__ == "__main__":
unittest.main()

86
tests/test_models.py Normal file
View File

@ -0,0 +1,86 @@
import _bootstrap # noqa: F401
import unittest
from datetime import datetime, timezone
from fault_log_analyzer import models
class TestIsoHelpers(unittest.TestCase):
def test_to_iso_naive(self):
dt = datetime(2025, 1, 1, 10, 0, 0)
self.assertEqual(models.to_iso(dt), "2025-01-01T10:00:00Z")
def test_from_iso_z(self):
dt = models.from_iso("2025-01-01T10:00:00Z")
self.assertEqual(dt.year, 2025)
self.assertIsNotNone(dt.tzinfo)
def test_from_iso_invalid_returns_none(self):
self.assertIsNone(models.from_iso("not-a-date"))
self.assertIsNone(models.from_iso(None))
self.assertIsNone(models.from_iso(""))
class TestFaultLogRoundTrip(unittest.TestCase):
def test_roundtrip(self):
fl = models.FaultLog(
fault_log_id="fl-1",
host_id="h-1",
fingerprint="fp",
level="ERROR",
message="disk full",
occurred_at=datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc),
fault_type="disk_full",
cluster_id="c-1",
service="app",
trace_id="t-1",
count=3,
)
d = fl.to_dict()
fl2 = models.FaultLog.from_dict(d)
self.assertEqual(fl2.fault_log_id, "fl-1")
self.assertEqual(fl2.host_id, "h-1")
self.assertEqual(fl2.count, 3)
self.assertEqual(fl2.occurred_at.tzinfo is not None, True)
class TestRootCauseRoundTrip(unittest.TestCase):
def test_roundtrip_with_evidence(self):
rc = models.RootCause(
fault_log_id="fl-1",
cause_type="disk_full",
confidence=0.92,
evidence=[
models.Evidence(type="event", event_id="e-1", metric="disk_used_percent", value=97.2),
models.Evidence(type="log", message="No space left on device"),
],
)
d = rc.to_dict()
rc2 = models.RootCause.from_dict(d)
self.assertEqual(rc2.fault_log_id, "fl-1")
self.assertEqual(rc2.cause_type, "disk_full")
self.assertEqual(len(rc2.evidence), 2)
self.assertEqual(rc2.evidence[0].event_id, "e-1")
class TestOtherModelRoundTrips(unittest.TestCase):
def test_fault_type(self):
ft = models.FaultType(fault_type="disk_full", name="磁盘满", pattern="disk full", severity="critical")
self.assertEqual(models.FaultType.from_dict(ft.to_dict()).fault_type, "disk_full")
def test_filter_rule(self):
r = models.FaultFilterRule(name="disk", level="ERROR", pattern="disk", exclude_pattern="ignore")
self.assertEqual(models.FaultFilterRule.from_dict(r.to_dict()).pattern, "disk")
def test_event(self):
e = models.Event(event_id="e-1", host_id="h-1", metric="cpu_usage", value=99.0)
self.assertEqual(models.Event.from_dict(e.to_dict()).value, 99.0)
def test_log_entry(self):
le = models.LogEntry(timestamp=datetime(2025, 1, 1, tzinfo=timezone.utc), level="ERROR", message="x")
self.assertEqual(models.LogEntry.from_dict(le.to_dict()).message, "x")
if __name__ == "__main__":
unittest.main()

26
tests/test_parser.py
View File

@ -1,5 +1,6 @@
import _bootstrap # noqa: F401 (将 src 加入 sys.path)
import unittest
from datetime import datetime, timezone
from fault_log_analyzer.parser import parse_log, parse_timestamp, templatize, tokenize
@ -9,14 +10,21 @@ class TestParseTimestamp(unittest.TestCase):
dt = parse_timestamp("2025-01-01T10:18:00Z")
self.assertEqual(dt.year, 2025)
self.assertEqual(dt.minute, 18)
self.assertIsNotNone(dt.tzinfo)
def test_epoch_seconds(self):
dt = parse_timestamp(1735726680) # 2025-01-01T10:18:00Z
self.assertEqual(dt.year, 2025)
self.assertIsNotNone(dt.tzinfo)
def test_epoch_millis(self):
dt = parse_timestamp(1735726680000)
self.assertEqual(dt.year, 2025)
self.assertEqual(dt.hour, 10)
def test_epoch_nanos(self):
dt = parse_timestamp(1735726680000000000)
self.assertEqual(dt.year, 2025)
def test_common_format(self):
dt = parse_timestamp("2025-01-01 10:18:00")
@ -26,6 +34,12 @@ class TestParseTimestamp(unittest.TestCase):
def test_invalid_returns_none(self):
self.assertIsNone(parse_timestamp("not-a-date"))
def test_datetime_passthrough(self):
from datetime import datetime, timezone
dt = parse_timestamp(datetime(2025, 1, 1, tzinfo=timezone.utc))
self.assertEqual(dt.year, 2025)
class TestParseLog(unittest.TestCase):
def test_structured_json(self):
@ -53,6 +67,16 @@ class TestParseLog(unittest.TestCase):
self.assertEqual(entry.level, "FATAL")
self.assertEqual(entry.host_id, "h-2")
def test_fields_fallback(self):
raw = {
"message": "boom",
"fields": {"hostname": "web-01", "service_name": "api", "request_id": "r-9"},
}
entry = parse_log(raw)
self.assertEqual(entry.host_id, "web-01")
self.assertEqual(entry.service, "api")
self.assertEqual(entry.trace_id, "r-9")
class TestTemplatize(unittest.TestCase):
def test_replace_variables(self):