develop: 开发 fault-log-analyzer 分析模块

src/fault_log_analyzer/__main__.py

@@ -0,0 +1,76 @@
+"""CLI 入口：启动 REST API（默认内存存储，便于本地验证）。"""
+from __future__ import annotations
+
+import argparse
+
+from .api import FaultLogApiServer
+from .config import Config
+from .storage import (
+    InMemoryEventRepository,
+    InMemoryFaultLogRepository,
+    InMemoryFaultTypeRepository,
+    InMemoryFilterRuleRepository,
+    InMemoryRootCauseRepository,
+)
+
+
+def build_server(cfg: Config) -> FaultLogApiServer:
+    if cfg.storage == "memory":
+        fault_log_repo = InMemoryFaultLogRepository()
+        root_cause_repo = InMemoryRootCauseRepository()
+        fault_type_repo = InMemoryFaultTypeRepository()
+        filter_rule_repo = InMemoryFilterRuleRepository()
+    else:  # pragma: no cover - 需要真实后端依赖
+        from .integrations import (
+            MySQLFaultLogRepository,
+            MySQLFaultTypeRepository,
+            MySQLFilterRuleRepository,
+            MySQLRootCauseRepository,
+        )
+
+        fault_log_repo = MySQLFaultLogRepository(
+            cfg.mysql_host, cfg.mysql_port, cfg.mysql_user, cfg.mysql_password, cfg.mysql_db
+        )
+        root_cause_repo = MySQLRootCauseRepository(
+            cfg.mysql_host, cfg.mysql_port, cfg.mysql_user, cfg.mysql_password, cfg.mysql_db
+        )
+        fault_type_repo = MySQLFaultTypeRepository(
+            cfg.mysql_host, cfg.mysql_port, cfg.mysql_user, cfg.mysql_password, cfg.mysql_db
+        )
+        filter_rule_repo = MySQLFilterRuleRepository(
+            cfg.mysql_host, cfg.mysql_port, cfg.mysql_user, cfg.mysql_password, cfg.mysql_db
+        )
+
+    return FaultLogApiServer(
+        fault_log_repo=fault_log_repo,
+        root_cause_repo=root_cause_repo,
+        fault_type_repo=fault_type_repo,
+        filter_rule_repo=filter_rule_repo,
+        host=cfg.api_host,
+        port=cfg.api_port,
+    )
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="fault-log-analyzer service")
+    parser.add_argument("--storage", default=None, help="memory 或 mysql（默认从环境变量读取）")
+    parser.add_argument("--api-host", default=None)
+    parser.add_argument("--api-port", type=int, default=None)
+    args = parser.parse_args(argv)
+
+    cfg = Config.from_env()
+    if args.storage:
+        cfg.storage = args.storage
+    if args.api_host:
+        cfg.api_host = args.api_host
+    if args.api_port:
+        cfg.api_port = args.api_port
+
+    server = build_server(cfg)
+    print(f"fault-log-analyzer listening on http://{cfg.api_host}:{cfg.api_port}")
+    server.start(block=True)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/fault_log_analyzer/api.py

@@ -0,0 +1,205 @@
+"""REST API：故障日志查询、根因分析、故障类型/过滤规则管理。
+
+使用标准库 ``http.server`` 实现，零第三方依赖，可直接运行；生产环境可由
+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
+from .storage import (
+    FaultLogRepository,
+    FaultTypeRepository,
+    FilterRuleRepository,
+    RootCauseRepository,
+)
+
+BASE_PATH = "/api/v1"
+
+
+class _JsonHandler(BaseHTTPRequestHandler):
+    server_version = "FaultLogAnalyzer/0.1"
+
+    # 由 create_handler 注入
+    fault_log_repo: FaultLogRepository
+    root_cause_repo: RootCauseRepository
+    fault_type_repo: FaultTypeRepository
+    filter_rule_repo: FilterRuleRepository
+
+    # -- 基础工具 ---------------------------------------------------------
+    def _send_json(self, code: int, payload: dict, status: int = 200) -> None:
+        body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def _ok(self, data) -> None:
+        self._send_json(200, {"code": 0, "message": "ok", "data": data})
+
+    def _error(self, code: int, message: str, status: int = 400) -> None:
+        self._send_json(status, {"code": code, "message": message, "data": None})
+
+    def _read_body(self) -> dict:
+        length = int(self.headers.get("Content-Length", "0") or "0")
+        if length <= 0:
+            return {}
+        try:
+            return json.loads(self.rfile.read(length).decode("utf-8"))
+        except (json.JSONDecodeError, UnicodeDecodeError):
+            return {}
+
+    def _query(self) -> dict:
+        parsed = urlparse(self.path)
+        return {k: v[0] for k, v in parse_qs(parsed.query).items()}
+
+    # -- 路由 -------------------------------------------------------------
+    def do_GET(self) -> None:  # noqa: N802
+        path = urlparse(self.path).path
+        if path == "/healthz":
+            return self._ok({"status": "ok"})
+        if path == "/readyz":
+            return self._ok({"status": "ok", "deps": ["memory"]})
+        if path == f"{BASE_PATH}/fault-logs":
+            return self._list_fault_logs()
+        if path.startswith(f"{BASE_PATH}/fault-logs/"):
+            rest = path[len(f"{BASE_PATH}/fault-logs/") :]
+            if rest.endswith("/root-cause"):
+                return self._get_root_cause(rest[: -len("/root-cause")])
+            return self._get_fault_log(rest)
+        if path == f"{BASE_PATH}/fault-types":
+            return self._list_fault_types()
+        if path == f"{BASE_PATH}/fault-filters":
+            return self._list_fault_filters()
+        return self._error(40401, "not found", 404)
+
+    def do_POST(self) -> None:  # noqa: N802
+        path = urlparse(self.path).path
+        if path == f"{BASE_PATH}/fault-types":
+            return self._create_fault_type()
+        if path == f"{BASE_PATH}/fault-filters":
+            return self._create_fault_filter()
+        return self._error(40401, "not found", 404)
+
+    # -- 故障日志 ---------------------------------------------------------
+    def _list_fault_logs(self) -> None:
+        q = self._query()
+        try:
+            page = max(1, int(q.get("page", "1")))
+            page_size = min(200, max(1, int(q.get("page_size", "20"))))
+        except ValueError:
+            return self._error(40001, "invalid param: page/page_size")
+        start = from_iso(q.get("from"))
+        end = from_iso(q.get("to"))
+        total, items = self.fault_log_repo.list(
+            host_id=q.get("host_id", ""),
+            fault_type=q.get("fault_type", ""),
+            level=q.get("level", ""),
+            keyword=q.get("keyword", ""),
+            start=start,
+            end=end,
+            page=page,
+            page_size=page_size,
+        )
+        self._ok({"total": total, "items": [i.to_dict() for i in items]})
+
+    def _get_fault_log(self, fault_log_id: str) -> None:
+        item = self.fault_log_repo.get(fault_log_id)
+        if item is None:
+            return self._error(40402, "fault log not found", 404)
+        self._ok(item.to_dict())
+
+    def _get_root_cause(self, fault_log_id: str) -> None:
+        item = self.root_cause_repo.get(fault_log_id)
+        if item is None:
+            return self._error(40403, "root cause not found", 404)
+        self._ok(item.to_dict())
+
+    # -- 故障类型 ---------------------------------------------------------
+    def _list_fault_types(self) -> None:
+        self._ok({"items": [ft.to_dict() for ft in self.fault_type_repo.list()]})
+
+    def _create_fault_type(self) -> None:
+        body = self._read_body()
+        try:
+            ft = FaultType.from_dict(body)
+            if not ft.fault_type or not ft.name:
+                return self._error(40002, "fault_type and name are required")
+            self.fault_type_repo.add(ft)
+        except (ValueError, KeyError):
+            return self._error(40001, "invalid param")
+        self._ok(ft.to_dict())
+
+    # -- 过滤规则 ---------------------------------------------------------
+    def _list_fault_filters(self) -> None:
+        self._ok({"items": [r.to_dict() for r in self.filter_rule_repo.list()]})
+
+    def _create_fault_filter(self) -> None:
+        body = self._read_body()
+        try:
+            rule = FaultFilterRule.from_dict(body)
+            if not rule.name or not rule.pattern:
+                return self._error(40002, "name and pattern are required")
+            self.filter_rule_repo.add(rule)
+        except (ValueError, KeyError):
+            return self._error(40001, "invalid param")
+        self._ok(rule.to_dict())
+
+    def log_message(self, format: str, *args) -> None:  # noqa: A002
+        # 静默访问日志，避免污染测试输出
+        pass
+
+
+def create_handler(
+    fault_log_repo: FaultLogRepository,
+    root_cause_repo: RootCauseRepository,
+    fault_type_repo: FaultTypeRepository,
+    filter_rule_repo: FilterRuleRepository,
+):
+    """创建绑定存储的请求处理器类。"""
+
+    class Handler(_JsonHandler):
+        pass
+
+    Handler.fault_log_repo = fault_log_repo
+    Handler.root_cause_repo = root_cause_repo
+    Handler.fault_type_repo = fault_type_repo
+    Handler.filter_rule_repo = filter_rule_repo
+    return Handler
+
+
+class FaultLogApiServer:
+    """可编程启停的 HTTP 服务。"""
+
+    def __init__(
+        self,
+        fault_log_repo: FaultLogRepository,
+        root_cause_repo: RootCauseRepository,
+        fault_type_repo: FaultTypeRepository,
+        filter_rule_repo: FilterRuleRepository,
+        host: str = "127.0.0.1",
+        port: int = 8080,
+    ):
+        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
+
+    def start(self, block: bool = False) -> None:
+        if block:
+            self._httpd.serve_forever()
+        else:
+            import threading
+
+            t = threading.Thread(target=self._httpd.serve_forever, daemon=True)
+            t.start()
+
+    def stop(self) -> None:
+        self._httpd.shutdown()
+        self._httpd.server_close()

tests/test_parser.py

@@ -0,0 +1,75 @@
+import unittest
+from datetime import datetime, timezone
+
+from fault_log_analyzer.parser import parse_log, parse_timestamp, templatize, tokenize
+
+
+class TestParseTimestamp(unittest.TestCase):
+    def test_iso_with_z(self):
+        dt = parse_timestamp("2025-01-01T10:18:00Z")
+        self.assertEqual(dt.year, 2025)
+        self.assertEqual(dt.minute, 18)
+
+    def test_epoch_seconds(self):
+        dt = parse_timestamp(1735726680)  # 2025-01-01T10:18:00Z
+        self.assertEqual(dt.year, 2025)
+
+    def test_epoch_millis(self):
+        dt = parse_timestamp(1735726680000)
+        self.assertEqual(dt.year, 2025)
+
+    def test_common_format(self):
+        dt = parse_timestamp("2025-01-01 10:18:00")
+        self.assertEqual(dt.year, 2025)
+        self.assertEqual(dt.hour, 10)
+
+    def test_invalid_returns_none(self):
+        self.assertIsNone(parse_timestamp("not-a-date"))
+
+
+class TestParseLog(unittest.TestCase):
+    def test_structured_json(self):
+        raw = {
+            "timestamp": "2025-01-01T10:18:00Z",
+            "level": "ERROR",
+            "message": "No space left on device",
+            "host_id": "h-001",
+            "service": "app",
+            "trace_id": "tr-1",
+        }
+        entry = parse_log(raw)
+        self.assertEqual(entry.level, "ERROR")
+        self.assertEqual(entry.host_id, "h-001")
+        self.assertEqual(entry.trace_id, "tr-1")
+
+    def test_level_extracted_from_message(self):
+        entry = parse_log({"message": "ERROR: something bad"})
+        self.assertEqual(entry.level, "ERROR")
+
+    def test_json_message_expands_fields(self):
+        raw = {"message": '{"message": "disk full", "level": "FATAL", "host_id": "h-2"}'}
+        entry = parse_log(raw)
+        self.assertEqual(entry.message, "disk full")
+        self.assertEqual(entry.level, "FATAL")
+        self.assertEqual(entry.host_id, "h-2")
+
+
+class TestTemplatize(unittest.TestCase):
+    def test_replace_variables(self):
+        msg = "Connection from 10.0.0.11 port 8080 failed after 3 retries"
+        tpl = templatize(msg)
+        self.assertIn("<IP>", tpl)
+        self.assertIn("<NUM>", tpl)
+
+    def test_replace_uuid(self):
+        tpl = templatize("request 123e4567-e89b-12d3-a456-426614174000 failed")
+        self.assertIn("<UUID>", tpl)
+
+    def test_tokenize(self):
+        tokens = tokenize("No space left on device")
+        self.assertIn("space", tokens)
+        self.assertIn("device", tokens)
+
+
+if __name__ == "__main__":
+    unittest.main()