From c62453ef703a5e3e8493981da1ac702ad95b4e8e Mon Sep 17 00:00:00 2001 From: Pipeline Agent Date: Sat, 15 Aug 2026 11:22:39 +0800 Subject: [PATCH] =?UTF-8?q?develop:=20=E8=A1=A5=E9=BD=90=20threshold-event?= =?UTF-8?q?-detector=20=E8=A7=84=E8=8C=83=E4=BA=A4=E4=BB=98=E4=BB=B6?= =?UTF-8?q?=E5=B9=B6=E6=B8=85=E7=90=86=E4=BA=A4=E6=8D=A2=E6=96=87=E4=BB=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/fault_log_analyzer/api.py | 5 +- src/fault_log_analyzer/models.py | 6 ++- src/fault_log_analyzer/parser.py | 20 +++++--- tests/_bootstrap.py | 14 ++++++ tests/test_config.py | 42 ++++++++++++++++ tests/test_filters.py | 58 +++++++++++++++++++++ tests/test_fingerprint.py | 62 +++++++++++++++++++++++ tests/test_models.py | 86 ++++++++++++++++++++++++++++++++ tests/test_parser.py | 26 +++++++++- 9 files changed, 308 insertions(+), 11 deletions(-) create mode 100644 tests/_bootstrap.py create mode 100644 tests/test_config.py create mode 100644 tests/test_filters.py create mode 100644 tests/test_fingerprint.py create mode 100644 tests/test_models.py diff --git a/src/fault_log_analyzer/api.py b/src/fault_log_analyzer/api.py index ecd3ab6..845e8ef 100644 --- a/src/fault_log_analyzer/api.py +++ b/src/fault_log_analyzer/api.py @@ -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: diff --git a/src/fault_log_analyzer/models.py b/src/fault_log_analyzer/models.py index 1888efb..4ab92c8 100644 --- a/src/fault_log_analyzer/models.py +++ b/src/fault_log_analyzer/models.py @@ -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) diff --git a/src/fault_log_analyzer/parser.py b/src/fault_log_analyzer/parser.py index 0f12ea9..cb05da9 100644 --- a/src/fault_log_analyzer/parser.py +++ b/src/fault_log_analyzer/parser.py @@ -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 diff --git a/tests/_bootstrap.py b/tests/_bootstrap.py new file mode 100644 index 0000000..4f38a49 --- /dev/null +++ b/tests/_bootstrap.py @@ -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) diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..63ce515 --- /dev/null +++ b/tests/test_config.py @@ -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() diff --git a/tests/test_filters.py b/tests/test_filters.py new file mode 100644 index 0000000..f43b755 --- /dev/null +++ b/tests/test_filters.py @@ -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() diff --git a/tests/test_fingerprint.py b/tests/test_fingerprint.py new file mode 100644 index 0000000..4310597 --- /dev/null +++ b/tests/test_fingerprint.py @@ -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() diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..20b1395 --- /dev/null +++ b/tests/test_models.py @@ -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() diff --git a/tests/test_parser.py b/tests/test_parser.py index 0754794..0a070d6 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -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):