diff --git a/src/fault_log_analyzer/__main__.py b/src/fault_log_analyzer/__main__.py new file mode 100644 index 0000000..9c39e5d --- /dev/null +++ b/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()) diff --git a/src/fault_log_analyzer/api.py b/src/fault_log_analyzer/api.py new file mode 100644 index 0000000..ecd3ab6 --- /dev/null +++ b/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() diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_parser.py b/tests/test_parser.py new file mode 100644 index 0000000..0754794 --- /dev/null +++ b/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("", tpl) + self.assertIn("", tpl) + + def test_replace_uuid(self): + tpl = templatize("request 123e4567-e89b-12d3-a456-426614174000 failed") + self.assertIn("", 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()