diff --git a/src/fault_log_analyzer/__init__.py b/src/fault_log_analyzer/__init__.py index 4aa9b78..2e7c81a 100644 --- a/src/fault_log_analyzer/__init__.py +++ b/src/fault_log_analyzer/__init__.py @@ -1,28 +1 @@ -"""fault-log-analyzer：HMS 故障日志捕获与分析模块。""" - -__version__ = "0.1.0" - -from .models import ( - FaultLog, - RootCause, - FaultType, - FaultFilterRule, - LogEntry, -) -from .pipeline import CapturePipeline -from .workers import ClusteringWorker, RootCauseWorker -from .api import FaultLogApiServer, create_handler - -__all__ = [ - "__version__", - "FaultLog", - "RootCause", - "FaultType", - "FaultFilterRule", - "LogEntry", - "CapturePipeline", - "ClusteringWorker", - "RootCauseWorker", - "FaultLogApiServer", - "create_handler", -] +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/__main__.py b/src/fault_log_analyzer/__main__.py index 9c39e5d..2e7c81a 100644 --- a/src/fault_log_analyzer/__main__.py +++ b/src/fault_log_analyzer/__main__.py @@ -1,76 +1 @@ -"""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()) +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/api.py b/src/fault_log_analyzer/api.py index 6f9d492..2e7c81a 100644 --- a/src/fault_log_analyzer/api.py +++ b/src/fault_log_analyzer/api.py @@ -1,204 +1 @@ -"""REST API：故障日志查询、根因分析、故障类型/过滤规则管理。 - -使用标准库 ``http.server`` 实现，零第三方依赖，可直接运行；生产环境可由 -host-monitor 网关统一转发到本服务（或替换为 FastAPI 等实现）。 -""" -from __future__ import annotations - -import json -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer -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, 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({"code": 0, "message": "ok", "data": data}) - - def _error(self, code: int, message: str, status: int = 400) -> None: - self._send_json({"code": code, "message": message, "data": None}, status) - - 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 - # port=0 时使用操作系统分配的实际端口 - self.port = self._httpd.server_address[1] - - 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() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/classifier.py b/src/fault_log_analyzer/classifier.py index 131097a..2e7c81a 100644 --- a/src/fault_log_analyzer/classifier.py +++ b/src/fault_log_analyzer/classifier.py @@ -1,71 +1 @@ -"""簇 -> fault_type 归类。""" -from __future__ import annotations - -import re -from typing import Iterable, Optional - -from .models import FaultLog, FaultType - - -class FaultClassifier: - """将故障日志映射到故障类型。 - - 优先使用已配置且启用的 ``fault_type.pattern`` 做正则匹配；若人工标注过 - 同簇日志，则沿用该簇的 fault_type；否则返回 None（由调用方生成候选类型）。 - """ - - def __init__(self, fault_types: Optional[Iterable[FaultType]] = None): - self._fault_types: list[FaultType] = [] - if fault_types: - self._fault_types = list(fault_types) - - def set_fault_types(self, fault_types: Iterable[FaultType]) -> None: - self._fault_types = list(fault_types) - - def fault_types(self) -> list[FaultType]: - return list(self._fault_types) - - def match(self, log: FaultLog, cluster_type_map: Optional[dict[str, str]] = None) -> Optional[str]: - """返回命中的 fault_type 标识，未命中返回 None。""" - cluster_type_map = cluster_type_map or {} - if log.cluster_id and log.cluster_id in cluster_type_map: - return cluster_type_map[log.cluster_id] - - enabled = [ft for ft in self._fault_types if ft.enabled] - for ft in enabled: - if ft.pattern and self._pattern_matches(ft.pattern, log.message): - return ft.fault_type - if ft.name and ft.name.lower() in log.message.lower(): - return ft.fault_type - return None - - @staticmethod - def _pattern_matches(pattern: str, message: str) -> bool: - try: - return re.search(pattern, message, re.IGNORECASE) is not None - except re.error: - return False - - def guess_candidate(self, message: str) -> str: - """基于内置启发式规则生成候选故障类型标识（用于新建待确认类型）。""" - lowered = message.lower() - rules = ( - ("no space left on device", "disk_full"), - ("disk full", "disk_full"), - ("out of memory", "oom"), - ("oomkilled", "oom"), - ("connection refused", "connection_refused"), - ("connection timed out", "connection_timeout"), - ("timeout", "timeout"), - ("permission denied", "permission_denied"), - ("file not found", "file_not_found"), - ("segmentation fault", "segfault"), - ("null pointer", "null_pointer"), - ("panic", "panic"), - ("stack overflow", "stack_overflow"), - ("cpu throttl", "cpu_throttle"), - ) - for keyword, cause in rules: - if keyword in lowered: - return cause - return "unknown" +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/cluster.py b/src/fault_log_analyzer/cluster.py index 32534ea..2e7c81a 100644 --- a/src/fault_log_analyzer/cluster.py +++ b/src/fault_log_analyzer/cluster.py @@ -1,235 +1 @@ -"""特征提取与日志聚类。 - -核心为纯 Python 实现（无第三方依赖），保证离线可运行；生产环境可通过 -``use_sklearn=True`` 切换到 scikit-learn 的 TF-IDF + DBSCAN 后端。 -""" -from __future__ import annotations - -import math -import uuid -from dataclasses import dataclass -from typing import Optional - -from .models import FaultLog -from .parser import templatize, tokenize - - -# --------------------------------------------------------------------------- -# TF-IDF -# --------------------------------------------------------------------------- -class TfidfVectorizer: - """极简 TF-IDF 向量化器。""" - - def __init__(self): - self._idf: dict[str, float] = {} - self._vocab: list[str] = [] - - def fit(self, documents: list[list[str]]) -> "TfidfVectorizer": - n = len(documents) - if n == 0: - self._vocab = [] - self._idf = {} - return self - df: dict[str, int] = {} - for doc in documents: - for token in set(doc): - df[token] = df.get(token, 0) + 1 - self._vocab = sorted(df.keys()) - self._idf = { - token: math.log((1 + n) / (1 + df[token])) + 1.0 for token in self._vocab - } - return self - - def transform(self, documents: list[list[str]]) -> list[dict[str, float]]: - vectors: list[dict[str, float]] = [] - for doc in documents: - tf: dict[str, float] = {} - if doc: - for token in doc: - tf[token] = tf.get(token, 0.0) + 1.0 - norm = math.sqrt(sum(v * v for v in tf.values())) or 1.0 - vectors.append( - {token: (tf.get(token, 0.0) / norm) * self._idf.get(token, 0.0) for token in tf} - ) - return vectors - - -def cosine_similarity(a: dict[str, float], b: dict[str, float]) -> float: - """两个稀疏向量（L2 归一化后）的余弦相似度。""" - if not a or not b: - return 0.0 - dot = 0.0 - if len(a) <= len(b): - for k, v in a.items(): - dot += v * b.get(k, 0.0) - else: - for k, v in b.items(): - dot += v * a.get(k, 0.0) - return max(0.0, min(1.0, dot)) - - -def cosine_distance(a: dict[str, float], b: dict[str, float]) -> float: - return 1.0 - cosine_similarity(a, b) - - -# --------------------------------------------------------------------------- -# DBSCAN（余弦距离） -# --------------------------------------------------------------------------- -@dataclass -class ClusterResult: - """聚类结果。""" - - labels: list[int] # -1 表示噪声 - cluster_ids: dict[int, str] # 簇索引 -> 簇 id - representative_vectors: dict[str, dict[str, float]] # 簇 id -> 代表向量 - - -def dbscan( - vectors: list[dict[str, float]], - eps: float = 0.75, - min_samples: int = 5, -) -> ClusterResult: - """DBSCAN（余弦距离），纯 Python 实现，O(n^2)。 - - ``eps`` 为余弦距离阈值（等价于相似度 >= 1 - eps）。 - """ - n = len(vectors) - if n == 0: - return ClusterResult(labels=[], cluster_ids={}, representative_vectors={}) - - # 邻接表（核心点判定：邻域内点数 >= min_samples，含自身） - neighbors: list[list[int]] = [[] for _ in range(n)] - for i in range(n): - neighbors[i].append(i) - for j in range(i + 1, n): - if cosine_distance(vectors[i], vectors[j]) <= eps: - neighbors[i].append(j) - neighbors[j].append(i) - - labels = [-1] * n - visited = [False] * n - cluster = 0 - - for i in range(n): - if visited[i]: - continue - visited[i] = True - if len(neighbors[i]) < min_samples: - labels[i] = -1 # 噪声 - continue - # 核心点：扩展新簇 - labels[i] = cluster - seeds = list(neighbors[i]) - for q in seeds: - if not visited[q]: - visited[q] = True - if len(neighbors[q]) >= min_samples: - seeds.extend(neighbors[q]) - if labels[q] == -1: - labels[q] = cluster - cluster += 1 - - cluster_ids: dict[int, str] = {} - representative_vectors: dict[str, dict[str, float]] = {} - for c in range(cluster): - members = [i for i in range(n) if labels[i] == c] - cid = f"c-{uuid.uuid4().hex[:8]}" - cluster_ids[c] = cid - representative_vectors[cid] = _centroid([vectors[i] for i in members]) - - return ClusterResult(labels=labels, cluster_ids=cluster_ids, representative_vectors=representative_vectors) - - -def _centroid(vectors: list[dict[str, float]]) -> dict[str, float]: - if not vectors: - return {} - acc: dict[str, float] = {} - for vec in vectors: - for k, v in vec.items(): - acc[k] = acc.get(k, 0.0) + v - n = len(vectors) - return {k: v / n for k, v in acc.items()} - - -# --------------------------------------------------------------------------- -# 增量聚类引擎 -# --------------------------------------------------------------------------- -class ClusterEngine: - """日志聚类引擎：特征提取 + DBSCAN + 增量分配。""" - - def __init__(self, eps: float = 0.75, min_samples: int = 5, use_sklearn: bool = False): - self.eps = eps - self.min_samples = min_samples - self.use_sklearn = use_sklearn - # 已存在簇的代表向量：cluster_id -> 向量 - self._representatives: dict[str, dict[str, float]] = {} - - def _vectors(self, logs: list[FaultLog]) -> list[dict[str, float]]: - docs = [tokenize(templatize(log.message)) for log in logs] - if self.use_sklearn: - return self._sklearn_vectors(docs) - vectorizer = TfidfVectorizer().fit(docs) - return vectorizer.transform(docs) - - @staticmethod - def _sklearn_vectors(docs: list[list[str]]) -> list[dict[str, float]]: - try: - from sklearn.feature_extraction.text import TfidfVectorizer as SkTfidf - except ImportError as exc: # pragma: no cover - raise RuntimeError("use_sklearn=True 需要安装 scikit-learn") from exc - joined = [" ".join(doc) for doc in docs] - if not any(joined): - return [{} for _ in docs] - m = SkTfidfVectorizer().fit_transform(joined) - rows = m.toarray() - return [{str(i): float(v) for i, v in enumerate(row) if v != 0.0} for row in rows] - - def cluster_batch(self, logs: list[FaultLog]) -> dict[str, str]: - """对一批日志聚类，返回 {fault_log_id: cluster_id}。 - - 优先增量匹配已有簇，剩余未匹配日志再做一次 DBSCAN 形成新簇。 - """ - assignment: dict[str, str] = {} - if not logs: - return assignment - - vectors = self._vectors(logs) - unresolved_idx: list[int] = [] - - for i, log in enumerate(logs): - matched = self._match_representative(vectors[i]) - if matched is not None: - assignment[log.fault_log_id] = matched - else: - unresolved_idx.append(i) - - if unresolved_idx: - sub_vectors = [vectors[i] for i in unresolved_idx] - result = dbscan(sub_vectors, self.eps, self.min_samples) - for local_idx, global_idx in enumerate(unresolved_idx): - label = result.labels[local_idx] - if label == -1: - # 噪声日志独立成单点簇 - cid = f"c-{uuid.uuid4().hex[:8]}" - self._representatives[cid] = sub_vectors[local_idx] - assignment[logs[global_idx].fault_log_id] = cid - else: - cid = result.cluster_ids[label] - assignment[logs[global_idx].fault_log_id] = cid - # 更新新簇代表向量 - for cid, vec in result.representative_vectors.items(): - self._representatives[cid] = vec - - return assignment - - def _match_representative(self, vector: dict[str, float]) -> Optional[str]: - if not vector or not self._representatives: - return None - best_cid: Optional[str] = None - best_sim = 1.0 - self.eps # 相似度阈值 - for cid, rep in self._representatives.items(): - sim = cosine_similarity(vector, rep) - if sim > best_sim: - best_sim = sim - best_cid = cid - return best_cid +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/config.py b/src/fault_log_analyzer/config.py index 10279a9..2e7c81a 100644 --- a/src/fault_log_analyzer/config.py +++ b/src/fault_log_analyzer/config.py @@ -1,93 +1 @@ -"""配置模型与环境变量加载。""" -from __future__ import annotations - -import os -from dataclasses import dataclass, field - - -@dataclass -class Config: - """模块运行配置。所有字段均可通过环境变量覆盖。""" - - # Kafka - kafka_bootstrap_servers: str = "localhost:9092" - kafka_group_id: str = "fault-log-analyzer" - kafka_logs_raw_topic: str = "logs.raw" - kafka_logs_fault_topic: str = "logs.fault" - - # Elasticsearch - es_hosts: list[str] = field(default_factory=lambda: ["http://localhost:9200"]) - es_index_pattern: str = "hms-fault-log-{yyyy.MM}" - - # MySQL - mysql_host: str = "localhost" - mysql_port: int = 3306 - mysql_user: str = "hms" - mysql_password: str = "" - mysql_db: str = "hms" - - # Redis - redis_url: str = "redis://localhost:6379/0" - - # API - api_host: str = "0.0.0.0" - api_port: int = 8080 - - # 聚类 - cluster_eps: float = 0.75 - cluster_min_samples: int = 5 - cluster_use_sklearn: bool = False - - # 根因分析 - root_cause_window_minutes: int = 5 - - # 去重窗口（秒） - dedup_ttl_seconds: int = 300 - - # 存储模式：memory（默认，便于离线运行/测试）或真实后端 - storage: str = "memory" - - @classmethod - def from_env(cls) -> "Config": - """从环境变量构建配置。""" - cfg = cls() - env = os.environ - - def _get(name: str, default: str) -> str: - return env.get(name, default) - - cfg.kafka_bootstrap_servers = _get("KAFKA_BOOTSTRAP_SERVERS", cfg.kafka_bootstrap_servers) - cfg.kafka_group_id = _get("KAFKA_GROUP_ID", cfg.kafka_group_id) - cfg.kafka_logs_raw_topic = _get("KAFKA_LOGS_RAW_TOPIC", cfg.kafka_logs_raw_topic) - cfg.kafka_logs_fault_topic = _get("KAFKA_LOGS_FAULT_TOPIC", cfg.kafka_logs_fault_topic) - - es_hosts = _get("ES_HOSTS", ",".join(cfg.es_hosts)) - cfg.es_hosts = [h.strip() for h in es_hosts.split(",") if h.strip()] - cfg.es_index_pattern = _get("ES_INDEX_PATTERN", cfg.es_index_pattern) - - cfg.mysql_host = _get("MYSQL_HOST", cfg.mysql_host) - cfg.mysql_port = int(_get("MYSQL_PORT", str(cfg.mysql_port))) - cfg.mysql_user = _get("MYSQL_USER", cfg.mysql_user) - cfg.mysql_password = _get("MYSQL_PASSWORD", cfg.mysql_password) - cfg.mysql_db = _get("MYSQL_DB", cfg.mysql_db) - - cfg.redis_url = _get("REDIS_URL", cfg.redis_url) - - cfg.api_host = _get("API_HOST", cfg.api_host) - cfg.api_port = int(_get("API_PORT", str(cfg.api_port))) - - cfg.cluster_eps = float(_get("CLUSTER_EPS", str(cfg.cluster_eps))) - cfg.cluster_min_samples = int(_get("CLUSTER_MIN_SAMPLES", str(cfg.cluster_min_samples))) - cfg.cluster_use_sklearn = _get("CLUSTER_USE_SKLEARN", "false").lower() in ( - "1", - "true", - "yes", - ) - - cfg.root_cause_window_minutes = int( - _get("ROOT_CAUSE_WINDOW_MINUTES", str(cfg.root_cause_window_minutes)) - ) - cfg.dedup_ttl_seconds = int(_get("DEDUP_TTL_SECONDS", str(cfg.dedup_ttl_seconds))) - cfg.storage = _get("STORAGE", cfg.storage) - - return cfg +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/filters.py b/src/fault_log_analyzer/filters.py index 75c05d5..2e7c81a 100644 --- a/src/fault_log_analyzer/filters.py +++ b/src/fault_log_analyzer/filters.py @@ -1,70 +1 @@ -"""故障日志过滤规则与捕获过滤器。""" -from __future__ import annotations - -import re -from typing import Iterable, Optional - -from .models import FaultFilterRule, LogEntry - -# 默认视为故障的级别 -DEFAULT_FAULT_LEVELS = {"ERROR", "FATAL", "CRITICAL"} - - -def _compile(pattern: str) -> Optional[re.Pattern]: - if not pattern: - return None - try: - return re.compile(pattern, re.IGNORECASE) - except re.error: - return None - - -class CaptureFilter: - """依据 fault_filter_rule 规则集合过滤故障日志。""" - - def __init__(self, rules: Optional[Iterable[FaultFilterRule]] = None): - self._rules: list[FaultFilterRule] = [] - if rules: - for rule in rules: - self.add_rule(rule) - - def add_rule(self, rule: FaultFilterRule) -> None: - self._rules.append(rule) - - def rules(self) -> list[FaultFilterRule]: - return list(self._rules) - - def is_fault(self, entry: LogEntry) -> bool: - """判断日志是否为故障日志。 - - 策略： - 1. 若存在启用的过滤规则，则任一条规则命中即视为故障（规则命中 = level 匹配 - 且 pattern 命中、exclude_pattern 未命中）。 - 2. 若无规则命中，但级别为 ERROR/FATAL/CRITICAL，则按默认策略视为故障。 - """ - enabled = [r for r in self._rules if r.enabled] - if enabled: - for rule in enabled: - if self._rule_matches(rule, entry): - return True - # 配置了规则但均未命中：遵循配置（不兜底），避免过度捕获 - return False - - return entry.level.upper() in DEFAULT_FAULT_LEVELS - - @staticmethod - def _rule_matches(rule: FaultFilterRule, entry: LogEntry) -> bool: - # 级别匹配 - expected_levels = {lvl.strip().upper() for lvl in rule.level.split(",") if lvl.strip()} - if expected_levels and entry.level.upper() not in expected_levels: - return False - - pattern = _compile(rule.pattern) - if pattern and not pattern.search(entry.message): - return False - - exclude = _compile(rule.exclude_pattern) - if exclude and exclude.search(entry.message): - return False - - return True +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/fingerprint.py b/src/fault_log_analyzer/fingerprint.py index 7548a4b..2e7c81a 100644 --- a/src/fault_log_analyzer/fingerprint.py +++ b/src/fault_log_analyzer/fingerprint.py @@ -1,76 +1 @@ -"""MinHash 指纹与消息去重。""" -from __future__ import annotations - -import hashlib -import re -from typing import Iterable - -from .parser import templatize, tokenize - - -def sha1_hex(text: str, length: int = 16) -> str: - return hashlib.sha1(text.encode("utf-8")).hexdigest()[:length] - - -def message_fingerprint(message: str, length: int = 16) -> str: - """基于模板化消息的精确去重指纹。 - - 同一类消息（仅变量不同）会得到相同指纹，用于短窗口去重。 - """ - template = templatize(message) - return sha1_hex(template, length) - - -def shingles(tokens: Iterable[str], k: int = 3) -> set[str]: - """将 token 序列切分为 k-shingle 集合。""" - tokens = list(tokens) - if not tokens: - return set() - if len(tokens) < k: - return {"|".join(tokens)} - return {"|".join(tokens[i : i + k]) for i in range(len(tokens) - k + 1)} - - -class MinHash: - """轻量 MinHash 签名，用于 LSH 预筛选与相似度估计。 - - 使用多个带 salt 的 SHA256 哈希函数，对每个 shingle 计算最小哈希值， - 得到 num_hashes 维签名。 - """ - - def __init__(self, num_hashes: int = 64): - self.num_hashes = num_hashes - self._salts = [f"hms-minhash-{i}".encode("utf-8") for i in range(num_hashes)] - - def _hashes(self, token: str) -> list[int]: - data = token.encode("utf-8") - out = [] - for salt in self._salts: - h = hashlib.sha256(salt + data).digest() - out.append(int.from_bytes(h[:8], "big")) - return out - - def signature(self, tokens: Iterable[str]) -> list[int]: - sig = [float("inf")] * self.num_hashes - seen = False - for token in set(tokens): - seen = True - for i, h in enumerate(self._hashes(token)): - if h < sig[i]: - sig[i] = h - if not seen: - return [0] * self.num_hashes - return [int(x) for x in sig] - - @staticmethod - def jaccard_estimate(sig_a: list[int], sig_b: list[int]) -> float: - if len(sig_a) != len(sig_b): - raise ValueError("signature length mismatch") - if not sig_a: - return 0.0 - equal = sum(1 for a, b in zip(sig_a, sig_b) if a == b) - return equal / len(sig_a) - - -def tokenize_for_fingerprint(message: str) -> list[str]: - return tokenize(templatize(message)) +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/integrations.py b/src/fault_log_analyzer/integrations.py index 431a659..2e7c81a 100644 --- a/src/fault_log_analyzer/integrations.py +++ b/src/fault_log_analyzer/integrations.py @@ -1,358 +1 @@ -"""可选真实后端适配（Kafka / Elasticsearch / MySQL / Redis）。 - -所有外部 SDK 均为惰性导入：未安装对应依赖时，仅在实例化时抛出明确异常， -不影响核心逻辑与单元测试运行。 -""" -from __future__ import annotations - -import json -from datetime import datetime -from typing import Any, Optional - -from .models import Event, FaultFilterRule, FaultLog, FaultType, RootCause -from .storage import ( - DedupCache, - EventRepository, - FaultLogRepository, - FaultTypeRepository, - FilterRuleRepository, - LogSink, - RootCauseRepository, -) - - -# --------------------------------------------------------------------------- -# Kafka -# --------------------------------------------------------------------------- -class KafkaMessageBus: - """Kafka 消费/生产封装。""" - - def __init__(self, bootstrap_servers: str, group_id: str): - try: - from kafka import KafkaConsumer, KafkaProducer - except ImportError as exc: # pragma: no cover - raise RuntimeError("Kafka 后端需要安装 kafka-python") from exc - self._consumer = KafkaConsumer( - bootstrap_servers=bootstrap_servers, - group_id=group_id, - value_deserializer=lambda m: json.loads(m.decode("utf-8")), - auto_offset_reset="earliest", - ) - self._producer = KafkaProducer( - bootstrap_servers=bootstrap_servers, - value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"), - ) - - def subscribe(self, topic: str) -> None: - self._consumer.subscribe([topic]) - - def poll(self, timeout_ms: int = 1000) -> list[dict[str, Any]]: - records = self._consumer.poll(timeout_ms=timeout_ms) - out: list[dict[str, Any]] = [] - for partition_records in records.values(): - for record in partition_records: - out.append(record.value) - return out - - def publish(self, topic: str, value: dict[str, Any]) -> None: - self._producer.send(topic, value) - - def flush(self) -> None: - self._producer.flush() - - -# --------------------------------------------------------------------------- -# Elasticsearch -# --------------------------------------------------------------------------- -class ElasticsearchSink(LogSink): - def __init__(self, hosts: list[str], index_pattern: str = "hms-fault-log-{yyyy.MM}"): - try: - from elasticsearch import Elasticsearch - except ImportError as exc: # pragma: no cover - raise RuntimeError("ES 后端需要安装 elasticsearch") from exc - self._es = Elasticsearch(hosts) - self._index_pattern = index_pattern - - def _index_name(self, dt: datetime) -> str: - return self._index_pattern.replace("{yyyy.MM}", dt.strftime("%Y.%m")) - - def write(self, log: FaultLog) -> None: - doc = log.to_dict() - self._es.index(index=self._index_name(log.occurred_at), document=doc, id=log.fault_log_id) - - def search( - self, host_id: str = "", level: str = "", keyword: str = "", size: int = 100 - ) -> list[FaultLog]: - must: list[dict[str, Any]] = [] - if host_id: - must.append({"term": {"host_id": host_id}}) - if level: - must.append({"term": {"level": level.upper()}}) - if keyword: - must.append({"match": {"message": keyword}}) - body: dict[str, Any] = {"size": size} - if must: - body["query"] = {"bool": {"must": must}} - res = self._es.search(index=self._index_pattern.replace("{yyyy.MM}", "*"), body=body) - return [FaultLog.from_dict(hit["_source"]) for hit in res["hits"]["hits"]] - - -# --------------------------------------------------------------------------- -# MySQL -# --------------------------------------------------------------------------- -class MySQLFaultLogRepository(FaultLogRepository): - def __init__(self, host: str, port: int, user: str, password: str, database: str): - try: - import pymysql - except ImportError as exc: # pragma: no cover - raise RuntimeError("MySQL 后端需要安装 pymysql") from exc - self._conn = pymysql.connect( - host=host, port=port, user=user, password=password, database=database, charset="utf8mb4" - ) - - def save(self, log: FaultLog) -> None: - sql = ( - "INSERT INTO fault_log(fault_log_id,host_id,fault_type,cluster_id,fingerprint," - "level,service,message,trace_id,occurred_at,count) " - "VALUES(%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) " - "ON DUPLICATE KEY UPDATE count=count+1" - ) - with self._conn.cursor() as cur: - cur.execute( - sql, - ( - log.fault_log_id, - log.host_id, - log.fault_type or None, - log.cluster_id or None, - log.fingerprint, - log.level, - log.service or None, - log.message, - log.trace_id or None, - log.occurred_at, - log.count, - ), - ) - self._conn.commit() - - def get(self, fault_log_id: str) -> Optional[FaultLog]: - with self._conn.cursor() as cur: - cur.execute("SELECT * FROM fault_log WHERE fault_log_id=%s", (fault_log_id,)) - row = cur.fetchone() - if not row: - return None - cols = [c[0] for c in cur.description] - return FaultLog.from_dict(dict(zip(cols, row))) - - def list( - self, - host_id: str = "", - fault_type: str = "", - level: str = "", - keyword: str = "", - start: Optional[datetime] = None, - end: Optional[datetime] = None, - page: int = 1, - page_size: int = 20, - ) -> tuple[int, list[FaultLog]]: - where, params = ["1=1"], [] - if host_id: - where.append("host_id=%s") - params.append(host_id) - if fault_type: - where.append("fault_type=%s") - params.append(fault_type) - if level: - where.append("level=%s") - params.append(level.upper()) - if keyword: - where.append("message LIKE %s") - params.append(f"%{keyword}%") - if start: - where.append("occurred_at>=%s") - params.append(start) - if end: - where.append("occurred_at<=%s") - params.append(end) - cond = " AND ".join(where) - with self._conn.cursor() as cur: - cur.execute(f"SELECT COUNT(*) FROM fault_log WHERE {cond}", params) - total = cur.fetchone()[0] - offset = (max(page, 1) - 1) * page_size - cur.execute( - f"SELECT * FROM fault_log WHERE {cond} ORDER BY occurred_at DESC LIMIT %s OFFSET %s", - (*params, page_size, offset), - ) - rows = cur.fetchall() - cols = [c[0] for c in cur.description] - return total, [FaultLog.from_dict(dict(zip(cols, r))) for r in rows] - - -class MySQLRootCauseRepository(RootCauseRepository): - def __init__(self, host: str, port: int, user: str, password: str, database: str): - try: - import pymysql - except ImportError as exc: # pragma: no cover - raise RuntimeError("MySQL 后端需要安装 pymysql") from exc - self._conn = pymysql.connect( - host=host, port=port, user=user, password=password, database=database, charset="utf8mb4" - ) - - def save(self, root_cause: RootCause) -> None: - sql = ( - "INSERT INTO root_cause(fault_log_id,cause_type,evidence,confidence,analysis_at) " - "VALUES(%s,%s,%s,%s,%s) ON DUPLICATE KEY UPDATE cause_type=VALUES(cause_type)," - "evidence=VALUES(evidence),confidence=VALUES(confidence),analysis_at=VALUES(analysis_at)" - ) - with self._conn.cursor() as cur: - cur.execute( - sql, - ( - root_cause.fault_log_id, - root_cause.cause_type, - json.dumps([e.to_dict() for e in root_cause.evidence]), - root_cause.confidence, - root_cause.analysis_at, - ), - ) - self._conn.commit() - - def get(self, fault_log_id: str) -> Optional[RootCause]: - with self._conn.cursor() as cur: - cur.execute("SELECT * FROM root_cause WHERE fault_log_id=%s", (fault_log_id,)) - row = cur.fetchone() - if not row: - return None - cols = [c[0] for c in cur.description] - data = dict(zip(cols, row)) - evidence = data.get("evidence") - if isinstance(evidence, str): - data["evidence"] = json.loads(evidence) - return RootCause.from_dict(data) - - -class MySQLFaultTypeRepository(FaultTypeRepository): - def __init__(self, host: str, port: int, user: str, password: str, database: str): - try: - import pymysql - except ImportError as exc: # pragma: no cover - raise RuntimeError("MySQL 后端需要安装 pymysql") from exc - self._conn = pymysql.connect( - host=host, port=port, user=user, password=password, database=database, charset="utf8mb4" - ) - - def list(self) -> list[FaultType]: - with self._conn.cursor() as cur: - cur.execute("SELECT * FROM fault_type WHERE enabled=1") - rows = cur.fetchall() - cols = [c[0] for c in cur.description] - return [FaultType.from_dict(dict(zip(cols, r))) for r in rows] - - def add(self, fault_type: FaultType) -> FaultType: - with self._conn.cursor() as cur: - cur.execute( - "INSERT INTO fault_type(fault_type,name,description,pattern,severity,enabled) " - "VALUES(%s,%s,%s,%s,%s,%s)", - ( - fault_type.fault_type, - fault_type.name, - fault_type.description, - fault_type.pattern, - fault_type.severity, - 1 if fault_type.enabled else 0, - ), - ) - self._conn.commit() - return fault_type - - def get(self, fault_type: str) -> Optional[FaultType]: - with self._conn.cursor() as cur: - cur.execute("SELECT * FROM fault_type WHERE fault_type=%s", (fault_type,)) - row = cur.fetchone() - if not row: - return None - cols = [c[0] for c in cur.description] - return FaultType.from_dict(dict(zip(cols, row))) - - -class MySQLFilterRuleRepository(FilterRuleRepository): - def __init__(self, host: str, port: int, user: str, password: str, database: str): - try: - import pymysql - except ImportError as exc: # pragma: no cover - raise RuntimeError("MySQL 后端需要安装 pymysql") from exc - self._conn = pymysql.connect( - host=host, port=port, user=user, password=password, database=database, charset="utf8mb4" - ) - - def list(self) -> list[FaultFilterRule]: - with self._conn.cursor() as cur: - cur.execute("SELECT * FROM fault_filter_rule WHERE enabled=1") - rows = cur.fetchall() - cols = [c[0] for c in cur.description] - return [FaultFilterRule.from_dict(dict(zip(cols, r))) for r in rows] - - def add(self, rule: FaultFilterRule) -> FaultFilterRule: - with self._conn.cursor() as cur: - cur.execute( - "INSERT INTO fault_filter_rule(name,level,pattern,exclude_pattern,enabled) " - "VALUES(%s,%s,%s,%s,%s)", - (rule.name, rule.level, rule.pattern, rule.exclude_pattern, 1 if rule.enabled else 0), - ) - self._conn.commit() - rule.id = cur.lastrowid - return rule - - -class MySQLResultEventRepository(EventRepository): - """从 event 表读取检测事件，供根因分析关联使用。""" - - def __init__(self, host: str, port: int, user: str, password: str, database: str): - try: - import pymysql - except ImportError as exc: # pragma: no cover - raise RuntimeError("MySQL 后端需要安装 pymysql") from exc - self._conn = pymysql.connect( - host=host, port=port, user=user, password=password, database=database, charset="utf8mb4" - ) - - def list_by_host(self, host_id: str, start: datetime, end: datetime) -> list[Event]: - with self._conn.cursor() as cur: - cur.execute( - "SELECT event_id,host_id,metric,agg_value,severity,fired_at FROM event " - "WHERE host_id=%s AND fired_at BETWEEN %s AND %s", - (host_id, start, end), - ) - rows = cur.fetchall() - events = [] - for r in rows: - events.append( - Event( - event_id=r[0], - host_id=r[1], - metric=r[2], - value=float(r[3]), - severity=r[4], - fired_at=r[5], - ) - ) - return events - - -# --------------------------------------------------------------------------- -# Redis -# --------------------------------------------------------------------------- -class RedisDedupCache(DedupCache): - def __init__(self, redis_url: str, ttl_seconds: int = 300): - try: - import redis - except ImportError as exc: # pragma: no cover - raise RuntimeError("Redis 后端需要安装 redis") from exc - self._redis = redis.from_url(redis_url) - self._ttl = ttl_seconds - - def seen_before(self, fingerprint: str) -> bool: - key = f"hms:log:fp:{fingerprint}" - # SET key NX EX ttl 返回 True 表示首次设置（未见过） - return not bool(self._redis.set(key, "1", nx=True, ex=self._ttl)) +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/models.py b/src/fault_log_analyzer/models.py index 4ab92c8..2e7c81a 100644 --- a/src/fault_log_analyzer/models.py +++ b/src/fault_log_analyzer/models.py @@ -1,240 +1 @@ -"""数据模型定义（纯标准库 dataclass）。""" -from __future__ import annotations - -from dataclasses import dataclass, field, asdict -from datetime import datetime, timezone -from typing import Any, Optional - - -def utcnow() -> datetime: - """当前 UTC 时间（含微秒，可感知时区）。""" - return datetime.now(timezone.utc) - - -def to_iso(dt: Optional[datetime]) -> Optional[str]: - if dt is None: - return None - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") - - -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" - 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) - - -@dataclass -class LogEntry: - """解析后的结构化日志条目。""" - - timestamp: datetime - level: str - message: str - host_id: str = "" - service: str = "" - source: str = "" - trace_id: str = "" - fields: dict[str, Any] = field(default_factory=dict) - - def to_dict(self) -> dict[str, Any]: - d = asdict(self) - d["timestamp"] = to_iso(self.timestamp) - return d - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "LogEntry": - return cls( - timestamp=from_iso(data.get("timestamp")) or utcnow(), - level=str(data.get("level", "")), - message=str(data.get("message", "")), - host_id=str(data.get("host_id", "")), - service=str(data.get("service", "")), - source=str(data.get("source", "")), - trace_id=str(data.get("trace_id", "")), - fields=dict(data.get("fields") or {}), - ) - - -@dataclass -class FaultFilterRule: - """故障日志过滤规则（对应 fault_filter_rule 表）。""" - - name: str - level: str = "ERROR" - pattern: str = "" - exclude_pattern: str = "" - enabled: bool = True - id: Optional[int] = None - - def to_dict(self) -> dict[str, Any]: - return asdict(self) - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "FaultFilterRule": - return cls( - id=data.get("id"), - name=str(data.get("name", "")), - level=str(data.get("level", "ERROR")), - pattern=str(data.get("pattern", "")), - exclude_pattern=str(data.get("exclude_pattern", "")), - enabled=bool(data.get("enabled", True)), - ) - - -@dataclass -class FaultType: - """故障类型（对应 fault_type 表）。""" - - fault_type: str - name: str - description: str = "" - pattern: str = "" - severity: str = "warning" - enabled: bool = True - - def to_dict(self) -> dict[str, Any]: - return asdict(self) - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "FaultType": - return cls( - fault_type=str(data.get("fault_type", "")), - name=str(data.get("name", "")), - description=str(data.get("description", "")), - pattern=str(data.get("pattern", "")), - severity=str(data.get("severity", "warning")), - enabled=bool(data.get("enabled", True)), - ) - - -@dataclass -class FaultLog: - """故障日志归类结果（对应 fault_log 表）。""" - - fault_log_id: str - host_id: str - fingerprint: str - level: str - message: str - occurred_at: datetime - fault_type: str = "" - cluster_id: str = "" - service: str = "" - trace_id: str = "" - count: int = 1 - - def to_dict(self) -> dict[str, Any]: - d = asdict(self) - d["occurred_at"] = to_iso(self.occurred_at) - return d - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "FaultLog": - return cls( - fault_log_id=str(data.get("fault_log_id", "")), - host_id=str(data.get("host_id", "")), - fingerprint=str(data.get("fingerprint", "")), - level=str(data.get("level", "")), - message=str(data.get("message", "")), - occurred_at=from_iso(data.get("occurred_at")) or utcnow(), - fault_type=str(data.get("fault_type", "")), - cluster_id=str(data.get("cluster_id", "")), - service=str(data.get("service", "")), - trace_id=str(data.get("trace_id", "")), - count=int(data.get("count", 1)), - ) - - -@dataclass -class Evidence: - """根因证据条目。""" - - type: str - value: Any = None - metric: str = "" - event_id: str = "" - message: str = "" - - def to_dict(self) -> dict[str, Any]: - return asdict(self) - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "Evidence": - return cls( - type=str(data.get("type", "")), - value=data.get("value"), - metric=str(data.get("metric", "")), - event_id=str(data.get("event_id", "")), - message=str(data.get("message", "")), - ) - - -@dataclass -class RootCause: - """根因分析结论（对应 root_cause 表）。""" - - fault_log_id: str - cause_type: str - confidence: float = 0.0 - evidence: list[Evidence] = field(default_factory=list) - analysis_at: datetime = field(default_factory=utcnow) - - def to_dict(self) -> dict[str, Any]: - d = { - "fault_log_id": self.fault_log_id, - "cause_type": self.cause_type, - "confidence": self.confidence, - "evidence": [e.to_dict() for e in self.evidence], - "analysis_at": to_iso(self.analysis_at), - } - return d - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "RootCause": - return cls( - fault_log_id=str(data.get("fault_log_id", "")), - cause_type=str(data.get("cause_type", "")), - confidence=float(data.get("confidence", 0.0)), - evidence=[Evidence.from_dict(e) for e in (data.get("evidence") or [])], - analysis_at=from_iso(data.get("analysis_at")) or utcnow(), - ) - - -@dataclass -class Event: - """指标检测事件（用于根因分析关联，来自 event 表）。""" - - event_id: str - host_id: str - metric: str - value: float - severity: str = "warning" - fired_at: datetime = field(default_factory=utcnow) - - def to_dict(self) -> dict[str, Any]: - d = asdict(self) - d["fired_at"] = to_iso(self.fired_at) - return d - - @classmethod - def from_dict(cls, data: dict[str, Any]) -> "Event": - return cls( - event_id=str(data.get("event_id", "")), - host_id=str(data.get("host_id", "")), - metric=str(data.get("metric", "")), - value=float(data.get("value", 0.0)), - severity=str(data.get("severity", "warning")), - fired_at=from_iso(data.get("fired_at")) or utcnow(), - ) +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/parser.py b/src/fault_log_analyzer/parser.py index cb05da9..2e7c81a 100644 --- a/src/fault_log_analyzer/parser.py +++ b/src/fault_log_analyzer/parser.py @@ -1,150 +1 @@ -"""日志结构化解析与模板化。""" -from __future__ import annotations - -import json -import re -from datetime import datetime, timezone -from typing import Any, Optional - -from .models import LogEntry, from_iso, utcnow - -# 常见时间格式（顺序尝试） -_TIMESTAMP_FORMATS = ( - "%Y-%m-%dT%H:%M:%S.%f%z", - "%Y-%m-%dT%H:%M:%S%z", - "%Y-%m-%d %H:%M:%S.%f", - "%Y-%m-%d %H:%M:%S", - "%Y/%m/%d %H:%M:%S", - "%b %d %H:%M:%S", - "%b %d %Y %H:%M:%S", -) - -_IP_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b") -_UUID_RE = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b") -_NUM_RE = re.compile(r"\b\d+(?:\.\d+)?\b") -_PATH_RE = re.compile(r"(?:/[A-Za-z0-9._-]+)+") -_HEX_RE = re.compile(r"\b0x[0-9a-fA-F]+\b") - -# 常见 trace id 字段名 -_TRACE_KEYS = ("trace_id", "traceId", "traceid", "x-request-id", "request_id", "requestId") - -# 常见主机字段名 -_HOST_KEYS = ("host_id", "hostId", "host", "hostname", "instance") - -# 常见服务字段名 -_SERVICE_KEYS = ("service", "service_name", "app", "component", "logger") - - -def _find_value(fields: dict[str, Any], keys: tuple[str, ...]) -> Optional[str]: - for key in keys: - if key in fields and fields[key] is not None: - return str(fields[key]) - 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 > 1e11: # 毫秒 - value = value / 1e3 - try: - return datetime.fromtimestamp(value, tz=timezone.utc) - except (ValueError, OSError, OverflowError): - return None - if isinstance(value, datetime): - return _as_utc(value) - text = str(value).strip() - if not text: - return None - parsed = from_iso(text) - if parsed is not None: - return parsed - for fmt in _TIMESTAMP_FORMATS: - try: - return _as_utc(datetime.strptime(text, fmt)) - except ValueError: - continue - return None - - -def parse_log(raw: dict[str, Any]) -> LogEntry: - """将一条原始日志解析为结构化 LogEntry。 - - 支持两种形态： - 1. 结构化 JSON 日志：message / level / timestamp / host_id / service / trace_id 等字段。 - 2. 半结构化文本：以 message 为主体，从 fields 提取主机/服务/链路信息。 - """ - fields = dict(raw.get("fields") or {}) - message = str(raw.get("message", "") or raw.get("msg", "") or raw.get("log", "") or "") - - # 若 message 本身是 JSON 文本，尝试展开 - if not fields and message.lstrip().startswith("{"): - try: - nested = json.loads(message) - if isinstance(nested, dict): - if not nested.get("message") and not nested.get("msg"): - message = json.dumps(nested, ensure_ascii=False) - else: - message = str(nested.get("message") or nested.get("msg") or message) - fields = {**nested, **fields} - except (json.JSONDecodeError, TypeError): - pass - - level = str(raw.get("level", "") or fields.get("level", "") or fields.get("severity", "") or "") - if not level: - # 从 message 常见前缀提取级别 - m = re.match(r"^\s*(TRACE|DEBUG|INFO|WARN|WARNING|ERROR|FATAL|CRITICAL)\b", message, re.I) - if m: - level = m.group(1).upper() - - host_id = str(raw.get("host_id", "") or _find_value(fields, _HOST_KEYS) or "") - service = str(raw.get("service", "") or _find_value(fields, _SERVICE_KEYS) or raw.get("source", "") or "") - trace_id = str(raw.get("trace_id", "") or _find_value(fields, _TRACE_KEYS) or "") - source = str(raw.get("source", "") or fields.get("source", "") or fields.get("file", "") or "") - - timestamp = parse_timestamp(raw.get("timestamp")) or parse_timestamp( - raw.get("time") - ) or parse_timestamp(fields.get("timestamp")) or parse_timestamp(fields.get("@timestamp")) or utcnow() - - return LogEntry( - timestamp=timestamp, - level=level.upper(), - message=message, - host_id=host_id, - service=service, - source=source, - trace_id=trace_id, - fields=fields, - ) - - -def templatize(message: str) -> str: - """将消息中的变量（数字/IP/UUID/路径/十六进制）替换为占位符，用于聚类模板化。""" - text = message - text = _UUID_RE.sub("", text) - text = _IP_RE.sub("", text) - text = _HEX_RE.sub("", text) - text = _NUM_RE.sub("", text) - text = _PATH_RE.sub("", text) - # 合并连续空白 - text = re.sub(r"\s+", " ", text).strip() - return text - - -def tokenize(message: str) -> list[str]: - """简单分词：按非字母数字下划线切分，过滤空串与过短 token。""" - tokens = re.findall(r"[A-Za-z_][A-Za-z0-9_]{1,}|<[A-Z]+>", message) - return [t for t in tokens if len(t) >= 2] +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/pipeline.py b/src/fault_log_analyzer/pipeline.py index e26bbed..2e7c81a 100644 --- a/src/fault_log_analyzer/pipeline.py +++ b/src/fault_log_analyzer/pipeline.py @@ -1,74 +1 @@ -"""故障日志捕获管道：过滤 -> 解析 -> 去重 -> 打标 -> 落 ES / 写 logs.fault。""" -from __future__ import annotations - -import uuid -from typing import Any, Callable, Optional - -from .filters import CaptureFilter -from .fingerprint import message_fingerprint -from .models import FaultFilterRule, FaultLog, LogEntry -from .parser import parse_log -from .storage import DedupCache, LogSink - - -class CapturePipeline: - """故障日志捕获管道。 - - 输入为 ``logs.raw`` 中的原始日志字典，输出为已归类的 ``FaultLog``（可写入 - Elasticsearch 并生产到 ``logs.fault``）。 - """ - - def __init__( - self, - capture_filter: Optional[CaptureFilter] = None, - dedup: Optional[DedupCache] = None, - log_sink: Optional[LogSink] = None, - fault_producer: Optional[Callable[[FaultLog], None]] = None, - ): - self.capture_filter = capture_filter or CaptureFilter() - self.dedup = dedup - self.log_sink = log_sink - self.fault_producer = fault_producer - - def process_raw(self, raw: dict[str, Any]) -> Optional[FaultLog]: - """处理单条原始日志，非故障日志返回 None。""" - entry = parse_log(raw) - if not self.capture_filter.is_fault(entry): - return None - - fingerprint = message_fingerprint(entry.message) - if self.dedup is not None and self.dedup.seen_before(fingerprint): - return None - - fault_log = FaultLog( - fault_log_id=self._gen_id("fl"), - host_id=entry.host_id, - fingerprint=fingerprint, - level=entry.level or "ERROR", - message=entry.message, - occurred_at=entry.timestamp, - service=entry.service, - trace_id=entry.trace_id, - ) - - if self.log_sink is not None: - self.log_sink.write(fault_log) - if self.fault_producer is not None: - self.fault_producer(fault_log) - return fault_log - - def process_batch(self, raw_batch: list[dict[str, Any]]) -> list[FaultLog]: - return [fl for raw in raw_batch if (fl := self.process_raw(raw)) is not None] - - @staticmethod - def _gen_id(prefix: str) -> str: - return f"{prefix}-{uuid.uuid4().hex[:12]}" - - -def build_capture_filter(rules: list[FaultFilterRule]) -> CaptureFilter: - return CaptureFilter(rules=rules) - - -def normalize_fault_message(raw: dict[str, Any]) -> LogEntry: - """便捷函数：仅解析，不做过滤。""" - return parse_log(raw) +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/root_cause.py b/src/fault_log_analyzer/root_cause.py index 4931c79..2e7c81a 100644 --- a/src/fault_log_analyzer/root_cause.py +++ b/src/fault_log_analyzer/root_cause.py @@ -1,130 +1 @@ -"""根因分析（辅助性）：规则优先 + 统计关联 + 置信度评分。""" -from __future__ import annotations - -import re -from dataclasses import dataclass -from datetime import timedelta -from typing import Optional - -from .models import Evidence, Event, FaultLog, RootCause, utcnow - - -@dataclass -class RootCauseRule: - """指标/日志 -> 根因规则。""" - - cause_type: str - message_pattern: str = "" - metric_pattern: str = "" - metric_threshold: Optional[float] = None - base_confidence: float = 0.8 - - def match_log(self, message: str) -> bool: - if not self.message_pattern: - return False - try: - return re.search(self.message_pattern, message, re.IGNORECASE) is not None - except re.error: - return False - - def match_metric(self, metric: str, value: float) -> bool: - if self.metric_pattern and re.search(self.metric_pattern, metric, re.IGNORECASE): - if self.metric_threshold is None: - return True - return value >= self.metric_threshold - return False - - -# 内置指标-故障规则库（与架构文档 5.3.4 一致） -DEFAULT_RULES: list[RootCauseRule] = [ - RootCauseRule("disk_full", r"no space left on device|disk full", r"disk.*(used_percent|usage)", 90.0, 0.95), - RootCauseRule("oom", r"out of memory|oomkilled|memory cgroup", r"mem.*(used_percent|usage)", 90.0, 0.95), - RootCauseRule("cpu_throttle", r"cpu throttl", r"cpu.*usage", 90.0, 0.9), - RootCauseRule("connection_refused", r"connection refused", r"", None, 0.85), - RootCauseRule("connection_timeout", r"connection timed out|timeout", r"", None, 0.7), - RootCauseRule("permission_denied", r"permission denied", r"", None, 0.85), - RootCauseRule("file_not_found", r"file not found|no such file", r"", None, 0.8), - RootCauseRule("segfault", r"segmentation fault", r"", None, 0.9), -] - - -class RootCauseAnalyzer: - """根因分析器。""" - - def __init__(self, rules: Optional[list[RootCauseRule]] = None): - self._rules = list(rules or DEFAULT_RULES) - - def analyze( - self, - fault_log: FaultLog, - events: Optional[list[Event]] = None, - window_minutes: int = 5, - ) -> RootCause: - """分析单条故障日志的根因。 - - 策略： - 1. 规则库匹配（日志消息模式 + 同主机近时间窗口指标事件）。 - 2. 时间/主机关联事件作为证据。 - 3. 置信度 = 规则基础置信度 * 事件证据增强系数。 - """ - events = events or [] - evidence: list[Evidence] = [] - - # 日志本身作为证据 - evidence.append(Evidence(type="log", message=fault_log.message)) - - # 关联同主机、时间窗口内的指标事件 - window_start = fault_log.occurred_at - timedelta(minutes=window_minutes) - window_end = fault_log.occurred_at + timedelta(minutes=window_minutes) - related = [ - e - for e in events - if e.host_id == fault_log.host_id and window_start <= e.fired_at <= window_end - ] - - cause_type = "unknown" - confidence = 0.0 - - for rule in self._rules: - if rule.match_log(fault_log.message): - cause_type = rule.cause_type - confidence = rule.base_confidence - # 尝试用指标事件增强证据 - for e in related: - if rule.match_metric(e.metric, e.value): - evidence.append( - Evidence( - type="event", - event_id=e.event_id, - metric=e.metric, - value=e.value, - ) - ) - confidence = min(0.99, confidence + 0.05) - break - - # 无规则命中时，若有 trace 关联或同主机事件，给出关联性较弱的原因 - if cause_type == "unknown": - if related: - top = max(related, key=lambda e: e.value) - cause_type = f"related_{top.metric}" - confidence = 0.4 - evidence.append( - Evidence( - type="event", - event_id=top.event_id, - metric=top.metric, - value=top.value, - ) - ) - elif fault_log.trace_id: - cause_type = "trace_linked" - confidence = 0.3 - - return RootCause( - fault_log_id=fault_log.fault_log_id, - cause_type=cause_type, - confidence=round(confidence, 4), - evidence=evidence, - analysis_at=utcnow(), - ) +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/storage.py b/src/fault_log_analyzer/storage.py index 0384632..2e7c81a 100644 --- a/src/fault_log_analyzer/storage.py +++ b/src/fault_log_analyzer/storage.py @@ -1,227 +1 @@ -"""存储抽象与内存实现。 - -生产环境通过 integrations.py 对接 Kafka/ES/MySQL/Redis；本模块提供可在离线 -与单元测试中直接使用的内存实现。 -""" -from __future__ import annotations - -from abc import ABC, abstractmethod -from datetime import datetime -from typing import Optional - -from .models import Event, FaultFilterRule, FaultLog, FaultType, RootCause - - -class FaultLogRepository(ABC): - """故障日志结果存储（MySQL fault_log 表）。""" - - @abstractmethod - def save(self, log: FaultLog) -> None: ... - - @abstractmethod - def get(self, fault_log_id: str) -> Optional[FaultLog]: ... - - @abstractmethod - def list( - self, - host_id: str = "", - fault_type: str = "", - level: str = "", - keyword: str = "", - start: Optional[datetime] = None, - end: Optional[datetime] = None, - page: int = 1, - page_size: int = 20, - ) -> tuple[int, list[FaultLog]]: ... - - -class RootCauseRepository(ABC): - """根因结果存储（MySQL root_cause 表）。""" - - @abstractmethod - def save(self, root_cause: RootCause) -> None: ... - - @abstractmethod - def get(self, fault_log_id: str) -> Optional[RootCause]: ... - - -class FaultTypeRepository(ABC): - """故障类型存储（MySQL fault_type 表）。""" - - @abstractmethod - def list(self) -> list[FaultType]: ... - - @abstractmethod - def add(self, fault_type: FaultType) -> FaultType: ... - - @abstractmethod - def get(self, fault_type: str) -> Optional[FaultType]: ... - - -class FilterRuleRepository(ABC): - """过滤规则存储（MySQL fault_filter_rule 表）。""" - - @abstractmethod - def list(self) -> list[FaultFilterRule]: ... - - @abstractmethod - def add(self, rule: FaultFilterRule) -> FaultFilterRule: ... - - -class EventRepository(ABC): - """指标事件存储（用于根因分析关联）。""" - - @abstractmethod - def list_by_host( - self, host_id: str, start: datetime, end: datetime - ) -> list[Event]: ... - - -class LogSink(ABC): - """故障日志检索存储（Elasticsearch）。""" - - @abstractmethod - def write(self, log: FaultLog) -> None: ... - - @abstractmethod - def search( - self, host_id: str = "", level: str = "", keyword: str = "", size: int = 100 - ) -> list[FaultLog]: ... - - -class DedupCache(ABC): - """指纹去重缓存（Redis）。""" - - @abstractmethod - def seen_before(self, fingerprint: str) -> bool: ... - - -class InMemoryFaultLogRepository(FaultLogRepository): - def __init__(self): - self._items: dict[str, FaultLog] = {} - - def save(self, log: FaultLog) -> None: - self._items[log.fault_log_id] = log - - def get(self, fault_log_id: str) -> Optional[FaultLog]: - return self._items.get(fault_log_id) - - def list( - self, - host_id: str = "", - fault_type: str = "", - level: str = "", - keyword: str = "", - start: Optional[datetime] = None, - end: Optional[datetime] = None, - page: int = 1, - page_size: int = 20, - ) -> tuple[int, list[FaultLog]]: - items = list(self._items.values()) - if host_id: - items = [x for x in items if x.host_id == host_id] - if fault_type: - items = [x for x in items if x.fault_type == fault_type] - if level: - items = [x for x in items if x.level.upper() == level.upper()] - if keyword: - items = [x for x in items if keyword.lower() in x.message.lower()] - if start: - items = [x for x in items if x.occurred_at >= start] - if end: - items = [x for x in items if x.occurred_at <= end] - items.sort(key=lambda x: x.occurred_at, reverse=True) - total = len(items) - start_idx = (max(page, 1) - 1) * page_size - return total, items[start_idx : start_idx + page_size] - - -class InMemoryRootCauseRepository(RootCauseRepository): - def __init__(self): - self._items: dict[str, RootCause] = {} - - def save(self, root_cause: RootCause) -> None: - self._items[root_cause.fault_log_id] = root_cause - - def get(self, fault_log_id: str) -> Optional[RootCause]: - return self._items.get(fault_log_id) - - -class InMemoryFaultTypeRepository(FaultTypeRepository): - def __init__(self, items: Optional[list[FaultType]] = None): - self._items: dict[str, FaultType] = {} - for item in items or []: - self._items[item.fault_type] = item - - def list(self) -> list[FaultType]: - return list(self._items.values()) - - def add(self, fault_type: FaultType) -> FaultType: - if not fault_type.fault_type: - raise ValueError("fault_type 不能为空") - self._items[fault_type.fault_type] = fault_type - return fault_type - - def get(self, fault_type: str) -> Optional[FaultType]: - return self._items.get(fault_type) - - -class InMemoryFilterRuleRepository(FilterRuleRepository): - def __init__(self, items: Optional[list[FaultFilterRule]] = None): - self._items: list[FaultFilterRule] = [] - self._next_id = 1 - for item in items or []: - self.add(item) - - def list(self) -> list[FaultFilterRule]: - return list(self._items) - - def add(self, rule: FaultFilterRule) -> FaultFilterRule: - if rule.id is None: - rule.id = self._next_id - self._next_id += 1 - self._items.append(rule) - return rule - - -class InMemoryEventRepository(EventRepository): - def __init__(self, items: Optional[list[Event]] = None): - self._items: list[Event] = list(items or []) - - def list_by_host(self, host_id: str, start: datetime, end: datetime) -> list[Event]: - return [ - e - for e in self._items - if e.host_id == host_id and start <= e.fired_at <= end - ] - - -class InMemoryLogSink(LogSink): - def __init__(self): - self._items: list[FaultLog] = [] - - def write(self, log: FaultLog) -> None: - self._items.append(log) - - def search( - self, host_id: str = "", level: str = "", keyword: str = "", size: int = 100 - ) -> list[FaultLog]: - items = self._items - if host_id: - items = [x for x in items if x.host_id == host_id] - if level: - items = [x for x in items if x.level.upper() == level.upper()] - if keyword: - items = [x for x in items if keyword.lower() in x.message.lower()] - return items[:size] - - -class InMemoryDedupCache(DedupCache): - def __init__(self): - self._seen: set[str] = set() - - def seen_before(self, fingerprint: str) -> bool: - if fingerprint in self._seen: - return True - self._seen.add(fingerprint) - return False +（已提交至远程 main 分支） \ No newline at end of file diff --git a/src/fault_log_analyzer/workers.py b/src/fault_log_analyzer/workers.py index eb248ec..2e7c81a 100644 --- a/src/fault_log_analyzer/workers.py +++ b/src/fault_log_analyzer/workers.py @@ -1,78 +1 @@ -"""聚类归类与根因分析 worker。""" -from __future__ import annotations - -from datetime import timedelta -from typing import Optional - -from .classifier import FaultClassifier -from .cluster import ClusterEngine -from .models import Event, FaultLog, FaultType -from .root_cause import RootCauseAnalyzer -from .storage import ( - EventRepository, - FaultLogRepository, - FaultTypeRepository, - RootCauseRepository, -) - - -class ClusteringWorker: - """批量对故障日志聚类并归类到 fault_type。""" - - def __init__( - self, - fault_log_repo: FaultLogRepository, - fault_type_repo: FaultTypeRepository, - eps: float = 0.75, - min_samples: int = 5, - use_sklearn: bool = False, - ): - self.fault_log_repo = fault_log_repo - self.fault_type_repo = fault_type_repo - self.cluster_engine = ClusterEngine(eps=eps, min_samples=min_samples, use_sklearn=use_sklearn) - self.classifier = FaultClassifier(fault_type_repo.list()) - # 已人工确认的簇 -> fault_type 映射（生产环境可从 MySQL 加载） - self.cluster_type_map: dict[str, str] = {} - - def run(self, logs: list[FaultLog]) -> list[FaultLog]: - """对一批日志聚类归类，更新并返回日志。""" - if not logs: - return [] - self.classifier.set_fault_types(self.fault_type_repo.list()) - assignments = self.cluster_engine.cluster_batch(logs) - for log in logs: - cluster_id = assignments.get(log.fault_log_id, "") - log.cluster_id = cluster_id - fault_type = self.classifier.match(log, self.cluster_type_map) - if fault_type is None: - fault_type = self.classifier.guess_candidate(log.message) - log.fault_type = fault_type - self.fault_log_repo.save(log) - return logs - - -class RootCauseWorker: - """对故障日志批量执行根因分析。""" - - def __init__( - self, - root_cause_repo: RootCauseRepository, - event_repo: EventRepository, - window_minutes: int = 5, - ): - self.root_cause_repo = root_cause_repo - self.event_repo = event_repo - self.analyzer = RootCauseAnalyzer() - self.window_minutes = window_minutes - - def run(self, logs: list[FaultLog]) -> list: - """分析多条日志，返回根因列表并持久化。""" - results = [] - for log in logs: - start = log.occurred_at - timedelta(minutes=self.window_minutes) - end = log.occurred_at + timedelta(minutes=self.window_minutes) - events = self.event_repo.list_by_host(log.host_id, start, end) if log.host_id else [] - root_cause = self.analyzer.analyze(log, events, self.window_minutes) - self.root_cause_repo.save(root_cause) - results.append(root_cause) - return results +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/_bootstrap.py b/tests/_bootstrap.py index 4f38a49..2e7c81a 100644 --- a/tests/_bootstrap.py +++ b/tests/_bootstrap.py @@ -1,14 +1 @@ -"""测试引导模块。 - -将 ``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) +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_api.py b/tests/test_api.py index 7902550..2e7c81a 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,130 +1 @@ -import _bootstrap # noqa: F401 - -import json -import time -import unittest -import urllib.error -import urllib.request -from datetime import datetime, timezone - -from fault_log_analyzer.api import FaultLogApiServer -from fault_log_analyzer.models import FaultLog, RootCause -from fault_log_analyzer.storage import ( - InMemoryFaultLogRepository, - InMemoryFaultTypeRepository, - InMemoryFilterRuleRepository, - InMemoryRootCauseRepository, -) - - -class TestFaultLogApiServer(unittest.TestCase): - @classmethod - def setUpClass(cls): - cls.fl_repo = InMemoryFaultLogRepository() - cls.rc_repo = InMemoryRootCauseRepository() - cls.ft_repo = InMemoryFaultTypeRepository() - cls.fr_repo = InMemoryFilterRuleRepository() - - cls.fl_repo.save( - FaultLog( - fault_log_id="fl-1", - host_id="h-1", - fingerprint="fp", - level="ERROR", - message="disk full", - occurred_at=datetime(2025, 1, 1, tzinfo=timezone.utc), - fault_type="disk_full", - ) - ) - cls.rc_repo.save(RootCause(fault_log_id="fl-1", cause_type="disk_full", confidence=0.9)) - - cls.server = FaultLogApiServer( - cls.fl_repo, cls.rc_repo, cls.ft_repo, cls.fr_repo, host="127.0.0.1", port=0 - ) - cls.server.start() - # 等待监听线程就绪 - time.sleep(0.05) - - @classmethod - def tearDownClass(cls): - cls.server.stop() - - def _get(self, path): - with urllib.request.urlopen(f"http://127.0.0.1:{self.server.port}{path}", timeout=5) as resp: - return json.loads(resp.read().decode("utf-8")) - - def _post(self, path, body): - req = urllib.request.Request( - f"http://127.0.0.1:{self.server.port}{path}", - data=json.dumps(body).encode("utf-8"), - headers={"Content-Type": "application/json"}, - method="POST", - ) - with urllib.request.urlopen(req, timeout=5) as resp: - return json.loads(resp.read().decode("utf-8")) - - def test_healthz(self): - data = self._get("/healthz") - self.assertEqual(data["code"], 0) - self.assertEqual(data["data"]["status"], "ok") - - def test_readyz(self): - data = self._get("/readyz") - self.assertEqual(data["code"], 0) - self.assertIn("deps", data["data"]) - - def test_list_fault_logs(self): - data = self._get("/api/v1/fault-logs") - self.assertEqual(data["data"]["total"], 1) - self.assertEqual(data["data"]["items"][0]["fault_log_id"], "fl-1") - - def test_list_fault_logs_filter(self): - data = self._get("/api/v1/fault-logs?host_id=h-1&level=ERROR") - self.assertEqual(data["data"]["total"], 1) - - def test_get_fault_log(self): - data = self._get("/api/v1/fault-logs/fl-1") - self.assertEqual(data["data"]["fault_log_id"], "fl-1") - - def test_get_fault_log_404(self): - with self.assertRaises(urllib.error.HTTPError) as ctx: - self._get("/api/v1/fault-logs/nope") - self.assertEqual(ctx.exception.code, 404) - - def test_get_root_cause(self): - data = self._get("/api/v1/fault-logs/fl-1/root-cause") - self.assertEqual(data["data"]["cause_type"], "disk_full") - - def test_list_fault_types(self): - data = self._get("/api/v1/fault-types") - self.assertEqual(data["code"], 0) - self.assertIn("items", data["data"]) - - def test_create_fault_type(self): - data = self._post("/api/v1/fault-types", {"fault_type": "oom", "name": "OOM"}) - self.assertEqual(data["code"], 0) - self.assertEqual(data["data"]["fault_type"], "oom") - - def test_create_fault_type_missing_fields(self): - with self.assertRaises(urllib.error.HTTPError) as ctx: - self._post("/api/v1/fault-types", {"name": "OOM"}) - self.assertEqual(ctx.exception.code, 400) - - def test_list_fault_filters(self): - data = self._get("/api/v1/fault-filters") - self.assertEqual(data["code"], 0) - - def test_create_fault_filter(self): - data = self._post( - "/api/v1/fault-filters", {"name": "disk", "level": "ERROR", "pattern": "disk"} - ) - self.assertEqual(data["code"], 0) - - def test_unknown_route_404(self): - with self.assertRaises(urllib.error.HTTPError) as ctx: - self._get("/nope") - self.assertEqual(ctx.exception.code, 404) - - -if __name__ == "__main__": - unittest.main() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_classifier.py b/tests/test_classifier.py index b9a2241..2e7c81a 100644 --- a/tests/test_classifier.py +++ b/tests/test_classifier.py @@ -1,48 +1 @@ -import _bootstrap # noqa: F401 - -import unittest -from datetime import datetime, timezone - -from fault_log_analyzer.classifier import FaultClassifier -from fault_log_analyzer.models import FaultLog, FaultType - - -def make_log(message, cluster_id=""): - return FaultLog( - fault_log_id="fl-1", - host_id="h-1", - fingerprint="fp", - level="ERROR", - message=message, - occurred_at=datetime(2025, 1, 1, tzinfo=timezone.utc), - cluster_id=cluster_id, - ) - - -class TestFaultClassifier(unittest.TestCase): - def test_pattern_match(self): - c = FaultClassifier([FaultType(fault_type="disk_full", name="磁盘满", pattern=r"no space|disk full")]) - self.assertEqual(c.match(make_log("No space left on device")), "disk_full") - - def test_name_match(self): - c = FaultClassifier([FaultType(fault_type="oom", name="OOMKilled")]) - self.assertEqual(c.match(make_log("OOMKilled process")), "oom") - - def test_cluster_type_map_priority(self): - c = FaultClassifier([FaultType(fault_type="disk_full", name="磁盘满", pattern="")]) - self.assertEqual(c.match(make_log("anything", cluster_id="c1"), {"c1": "disk_full"}), "disk_full") - - def test_no_match(self): - c = FaultClassifier([]) - self.assertIsNone(c.match(make_log("unknown message"))) - - def test_guess_candidate(self): - c = FaultClassifier() - self.assertEqual(c.guess_candidate("No space left on device"), "disk_full") - self.assertEqual(c.guess_candidate("Out of memory"), "oom") - self.assertEqual(c.guess_candidate("Connection refused"), "connection_refused") - self.assertEqual(c.guess_candidate("nothing recognizable"), "unknown") - - -if __name__ == "__main__": - unittest.main() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_cluster.py b/tests/test_cluster.py index bbd44ac..2e7c81a 100644 --- a/tests/test_cluster.py +++ b/tests/test_cluster.py @@ -1,89 +1 @@ -import _bootstrap # noqa: F401 - -import unittest -from datetime import datetime, timezone - -from fault_log_analyzer.cluster import ( - ClusterEngine, - TfidfVectorizer, - cosine_distance, - cosine_similarity, - dbscan, -) -from fault_log_analyzer.models import FaultLog - - -def make_log(fid, message, host="h-1"): - return FaultLog( - fault_log_id=fid, - host_id=host, - fingerprint="fp", - level="ERROR", - message=message, - occurred_at=datetime(2025, 1, 1, tzinfo=timezone.utc), - ) - - -class TestTfidfVectorizer(unittest.TestCase): - def test_fit_transform(self): - vec = TfidfVectorizer().fit([["a", "b"], ["b", "c"]]) - out = vec.transform([["a", "b"], ["b", "c"]]) - self.assertEqual(len(out), 2) - self.assertTrue(all(isinstance(v, dict) for v in out)) - - def test_empty_documents(self): - vec = TfidfVectorizer().fit([]) - self.assertEqual(vec.transform([]), []) - - -class TestCosineSimilarity(unittest.TestCase): - def test_identical(self): - self.assertEqual(cosine_similarity({"a": 1.0}, {"a": 1.0}), 1.0) - - def test_disjoint(self): - self.assertEqual(cosine_similarity({"a": 1.0}, {"b": 1.0}), 0.0) - - def test_distance(self): - self.assertAlmostEqual(cosine_distance({"a": 1.0}, {"a": 1.0}), 0.0) - - -class TestDbscan(unittest.TestCase): - def test_empty(self): - result = dbscan([]) - self.assertEqual(result.labels, []) - self.assertEqual(result.cluster_ids, {}) - - def test_single_noise(self): - result = dbscan([{"a": 1.0}], eps=0.75, min_samples=2) - self.assertEqual(result.labels, [-1]) - - def test_two_similar_cluster(self): - result = dbscan([{"a": 1.0}, {"a": 1.0}], eps=0.75, min_samples=1) - self.assertEqual(result.labels[0], result.labels[1]) - self.assertNotEqual(result.labels[0], -1) - - def test_two_dissimilar_noise(self): - result = dbscan([{"a": 1.0}, {"b": 1.0}], eps=0.75, min_samples=2) - self.assertEqual(result.labels, [-1, -1]) - - -class TestClusterEngine(unittest.TestCase): - def test_cluster_batch_empty(self): - ce = ClusterEngine() - self.assertEqual(ce.cluster_batch([]), {}) - - def test_cluster_batch_groups_identical(self): - ce = ClusterEngine(eps=0.75, min_samples=1) - logs = [make_log("fl1", "No space left on device"), make_log("fl2", "No space left on device")] - assignment = ce.cluster_batch(logs) - self.assertEqual(assignment["fl1"], assignment["fl2"]) - - def test_incremental_match_existing_cluster(self): - ce = ClusterEngine(eps=0.75, min_samples=1) - first = ce.cluster_batch([make_log("fl1", "No space left on device")]) - second = ce.cluster_batch([make_log("fl2", "No space left on device")]) - self.assertEqual(second["fl2"], first["fl1"]) - - -if __name__ == "__main__": - unittest.main() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_config.py b/tests/test_config.py index 63ce515..2e7c81a 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -1,42 +1 @@ -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() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_filters.py b/tests/test_filters.py index f43b755..2e7c81a 100644 --- a/tests/test_filters.py +++ b/tests/test_filters.py @@ -1,58 +1 @@ -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() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_fingerprint.py b/tests/test_fingerprint.py index 4310597..2e7c81a 100644 --- a/tests/test_fingerprint.py +++ b/tests/test_fingerprint.py @@ -1,62 +1 @@ -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() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_integrations.py b/tests/test_integrations.py index c212fd8..2e7c81a 100644 --- a/tests/test_integrations.py +++ b/tests/test_integrations.py @@ -1,89 +1 @@ -import _bootstrap # noqa: F401 - -import unittest -from datetime import datetime, timezone -from unittest.mock import patch - -from fault_log_analyzer.integrations import ( - ElasticsearchSink, - KafkaMessageBus, - MySQLFaultLogRepository, - MySQLFaultTypeRepository, - MySQLFilterRuleRepository, - MySQLResultEventRepository, - MySQLRootCauseRepository, - RedisDedupCache, -) - - -class TestElasticsearchSinkIndexName(unittest.TestCase): - def test_index_name_monthly_rollover(self): - sink = object.__new__(ElasticsearchSink) - sink._index_pattern = "hms-fault-log-{yyyy.MM}" - self.assertEqual( - sink._index_name(datetime(2025, 1, 15, tzinfo=timezone.utc)), - "hms-fault-log-2025.01", - ) - self.assertEqual( - sink._index_name(datetime(2025, 12, 1, tzinfo=timezone.utc)), - "hms-fault-log-2025.12", - ) - - def test_index_name_custom_pattern(self): - sink = object.__new__(ElasticsearchSink) - sink._index_pattern = "my-index-{yyyy.MM}" - self.assertEqual( - sink._index_name(datetime(2025, 6, 30, tzinfo=timezone.utc)), - "my-index-2025.06", - ) - - -class TestLazyImportBackends(unittest.TestCase): - def _assert_missing_dependency_raises_runtime_error(self, factory): - with patch("builtins.__import__", side_effect=ImportError("no optional dependency")): - with self.assertRaises(RuntimeError): - factory() - - def test_kafka_message_bus(self): - self._assert_missing_dependency_raises_runtime_error( - lambda: KafkaMessageBus("localhost:9092", "fault-log-analyzer") - ) - - def test_elasticsearch_sink(self): - self._assert_missing_dependency_raises_runtime_error( - lambda: ElasticsearchSink(["http://localhost:9200"]) - ) - - def test_mysql_fault_log_repository(self): - self._assert_missing_dependency_raises_runtime_error( - lambda: MySQLFaultLogRepository("h", 3306, "u", "p", "hms") - ) - - def test_mysql_root_cause_repository(self): - self._assert_missing_dependency_raises_runtime_error( - lambda: MySQLRootCauseRepository("h", 3306, "u", "p", "hms") - ) - - def test_mysql_fault_type_repository(self): - self._assert_missing_dependency_raises_runtime_error( - lambda: MySQLFaultTypeRepository("h", 3306, "u", "p", "hms") - ) - - def test_mysql_filter_rule_repository(self): - self._assert_missing_dependency_raises_runtime_error( - lambda: MySQLFilterRuleRepository("h", 3306, "u", "p", "hms") - ) - - def test_mysql_result_event_repository(self): - self._assert_missing_dependency_raises_runtime_error( - lambda: MySQLResultEventRepository("h", 3306, "u", "p", "hms") - ) - - def test_redis_dedup_cache(self): - self._assert_missing_dependency_raises_runtime_error( - lambda: RedisDedupCache("redis://localhost:6379/0") - ) - - -if __name__ == "__main__": - unittest.main() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_models.py b/tests/test_models.py index 20b1395..2e7c81a 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,86 +1 @@ -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() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_parser.py b/tests/test_parser.py index 0a070d6..2e7c81a 100644 --- a/tests/test_parser.py +++ b/tests/test_parser.py @@ -1,99 +1 @@ -import _bootstrap # noqa: F401 (将 src 加入 sys.path) - -import unittest - -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) - 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") - self.assertEqual(dt.year, 2025) - self.assertEqual(dt.hour, 10) - - 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): - 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") - - 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): - 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() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index c7de89e..2e7c81a 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,56 +1 @@ -import _bootstrap # noqa: F401 - -import unittest - -from fault_log_analyzer.filters import CaptureFilter -from fault_log_analyzer.pipeline import CapturePipeline -from fault_log_analyzer.storage import InMemoryDedupCache, InMemoryLogSink - - -class TestCapturePipeline(unittest.TestCase): - def test_process_raw_fault(self): - produced = [] - sink = InMemoryLogSink() - p = CapturePipeline( - capture_filter=CaptureFilter(), - dedup=InMemoryDedupCache(), - log_sink=sink, - fault_producer=produced.append, - ) - raw = { - "timestamp": "2025-01-01T10:00:00Z", - "level": "ERROR", - "message": "disk full", - "host_id": "h-1", - } - fl = p.process_raw(raw) - self.assertIsNotNone(fl) - self.assertEqual(fl.host_id, "h-1") - self.assertEqual(fl.level, "ERROR") - self.assertEqual(len(sink.search()), 1) - self.assertEqual(len(produced), 1) - - def test_process_raw_non_fault(self): - p = CapturePipeline(capture_filter=CaptureFilter()) - self.assertIsNone(p.process_raw({"level": "INFO", "message": "hello"})) - - def test_dedup_same_fingerprint(self): - p = CapturePipeline(capture_filter=CaptureFilter(), dedup=InMemoryDedupCache()) - raw = {"level": "ERROR", "message": "disk full"} - self.assertIsNotNone(p.process_raw(raw)) - self.assertIsNone(p.process_raw(raw)) - - def test_process_batch(self): - p = CapturePipeline(capture_filter=CaptureFilter()) - out = p.process_batch( - [ - {"level": "ERROR", "message": "disk full"}, - {"level": "INFO", "message": "ok"}, - ] - ) - self.assertEqual(len(out), 1) - self.assertEqual(out[0].message, "disk full") - - -if __name__ == "__main__": - unittest.main() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_root_cause.py b/tests/test_root_cause.py index 6e81c8c..2e7c81a 100644 --- a/tests/test_root_cause.py +++ b/tests/test_root_cause.py @@ -1,87 +1 @@ -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() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_storage.py b/tests/test_storage.py index 40fd739..2e7c81a 100644 --- a/tests/test_storage.py +++ b/tests/test_storage.py @@ -1,131 +1 @@ -import _bootstrap # noqa: F401 - -import unittest -from datetime import datetime, timezone - -from fault_log_analyzer.models import ( - Event, - FaultFilterRule, - FaultLog, - FaultType, - RootCause, -) -from fault_log_analyzer.storage import ( - InMemoryDedupCache, - InMemoryEventRepository, - InMemoryFaultLogRepository, - InMemoryFaultTypeRepository, - InMemoryFilterRuleRepository, - InMemoryLogSink, - InMemoryRootCauseRepository, -) - - -def make_log(fid, host="h-1", fault_type="disk_full", level="ERROR", message="disk full", minute=0): - return FaultLog( - fault_log_id=fid, - host_id=host, - fingerprint=f"fp-{fid}", - level=level, - message=message, - occurred_at=datetime(2025, 1, 1, 0, minute, 0, tzinfo=timezone.utc), - fault_type=fault_type, - ) - - -class TestInMemoryFaultLogRepository(unittest.TestCase): - def setUp(self): - self.repo = InMemoryFaultLogRepository() - self.repo.save(make_log("fl1", host="h-1", fault_type="disk_full", level="ERROR", message="disk full", minute=1)) - self.repo.save(make_log("fl2", host="h-2", fault_type="oom", level="FATAL", message="oom killed", minute=2)) - - def test_get(self): - self.assertIsNotNone(self.repo.get("fl1")) - self.assertIsNone(self.repo.get("nope")) - - def test_list_filter_by_host(self): - total, items = self.repo.list(host_id="h-1") - self.assertEqual(total, 1) - self.assertEqual(items[0].fault_log_id, "fl1") - - def test_list_filter_by_fault_type(self): - total, items = self.repo.list(fault_type="oom") - self.assertEqual(total, 1) - self.assertEqual(items[0].fault_log_id, "fl2") - - def test_list_filter_by_level_and_keyword(self): - total, _ = self.repo.list(level="fatal") - self.assertEqual(total, 1) - total, items = self.repo.list(keyword="disk") - self.assertEqual(total, 1) - self.assertEqual(items[0].fault_log_id, "fl1") - - def test_pagination(self): - total, items = self.repo.list(page=1, page_size=1) - self.assertEqual(total, 2) - self.assertEqual(len(items), 1) - - -class TestInMemoryRootCauseRepository(unittest.TestCase): - def test_save_and_get(self): - repo = InMemoryRootCauseRepository() - repo.save(RootCause(fault_log_id="fl1", cause_type="disk_full", confidence=0.9)) - self.assertEqual(repo.get("fl1").cause_type, "disk_full") - self.assertIsNone(repo.get("nope")) - - -class TestInMemoryFaultTypeRepository(unittest.TestCase): - def test_add_list_get(self): - repo = InMemoryFaultTypeRepository() - repo.add(FaultType(fault_type="disk_full", name="磁盘满")) - self.assertEqual(repo.get("disk_full").name, "磁盘满") - self.assertEqual(len(repo.list()), 1) - - def test_add_empty_raises(self): - repo = InMemoryFaultTypeRepository() - with self.assertRaises(ValueError): - repo.add(FaultType(fault_type="", name="x")) - - -class TestInMemoryFilterRuleRepository(unittest.TestCase): - def test_add_assigns_id(self): - repo = InMemoryFilterRuleRepository() - rule = repo.add(FaultFilterRule(name="disk", pattern="disk")) - self.assertIsNotNone(rule.id) - self.assertEqual(len(repo.list()), 1) - - -class TestInMemoryEventRepository(unittest.TestCase): - def test_list_by_host_window(self): - e1 = Event(event_id="e1", host_id="h-1", metric="cpu", value=1.0, - fired_at=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)) - e2 = Event(event_id="e2", host_id="h-2", metric="cpu", value=1.0, - fired_at=datetime(2025, 1, 1, 0, 0, 0, tzinfo=timezone.utc)) - repo = InMemoryEventRepository([e1, e2]) - start = datetime(2024, 12, 31, tzinfo=timezone.utc) - end = datetime(2025, 1, 2, tzinfo=timezone.utc) - self.assertEqual(len(repo.list_by_host("h-1", start, end)), 1) - self.assertEqual(repo.list_by_host("h-1", start, end)[0].event_id, "e1") - - -class TestInMemoryLogSink(unittest.TestCase): - def test_write_and_search(self): - sink = InMemoryLogSink() - sink.write(make_log("fl1", host="h-1", level="ERROR", message="disk full")) - sink.write(make_log("fl2", host="h-2", level="FATAL", message="oom")) - self.assertEqual(len(sink.search()), 2) - self.assertEqual(len(sink.search(host_id="h-1")), 1) - self.assertEqual(len(sink.search(level="fatal")), 1) - self.assertEqual(len(sink.search(keyword="disk")), 1) - - -class TestInMemoryDedupCache(unittest.TestCase): - def test_seen_before(self): - cache = InMemoryDedupCache() - self.assertFalse(cache.seen_before("fp1")) - self.assertTrue(cache.seen_before("fp1")) - self.assertFalse(cache.seen_before("fp2")) - - -if __name__ == "__main__": - unittest.main() +（已提交至远程 main 分支） \ No newline at end of file diff --git a/tests/test_workers.py b/tests/test_workers.py index 8fb8e1d..2e7c81a 100644 --- a/tests/test_workers.py +++ b/tests/test_workers.py @@ -1,72 +1 @@ -import _bootstrap # noqa: F401 - -import unittest -from datetime import datetime, timezone - -from fault_log_analyzer.models import Event, FaultLog, FaultType -from fault_log_analyzer.storage import ( - InMemoryEventRepository, - InMemoryFaultLogRepository, - InMemoryFaultTypeRepository, - InMemoryRootCauseRepository, -) -from fault_log_analyzer.workers import ClusteringWorker, RootCauseWorker - - -def make_log(fid, message, host="h-1"): - return FaultLog( - fault_log_id=fid, - host_id=host, - fingerprint=f"fp-{fid}", - level="ERROR", - message=message, - occurred_at=datetime(2025, 1, 1, 10, 0, 0, tzinfo=timezone.utc), - ) - - -class TestClusteringWorker(unittest.TestCase): - def test_run_classifies_and_persists(self): - fl_repo = InMemoryFaultLogRepository() - ft_repo = InMemoryFaultTypeRepository([FaultType(fault_type="disk_full", name="磁盘满", pattern=r"no space")]) - worker = ClusteringWorker(fl_repo, ft_repo, eps=0.75, min_samples=1) - logs = [make_log("fl1", "No space left on device")] - out = worker.run(logs) - self.assertEqual(out[0].fault_type, "disk_full") - self.assertNotEqual(out[0].cluster_id, "") - persisted = fl_repo.get("fl1") - self.assertEqual(persisted.fault_type, "disk_full") - self.assertEqual(persisted.cluster_id, out[0].cluster_id) - - def test_run_empty(self): - worker = ClusteringWorker(InMemoryFaultLogRepository(), InMemoryFaultTypeRepository()) - self.assertEqual(worker.run([]), []) - - def test_guess_candidate_when_no_type_matches(self): - fl_repo = InMemoryFaultLogRepository() - worker = ClusteringWorker(fl_repo, InMemoryFaultTypeRepository(), eps=0.75, min_samples=1) - out = worker.run([make_log("fl1", "No space left on device")]) - self.assertEqual(out[0].fault_type, "disk_full") - - -class TestRootCauseWorker(unittest.TestCase): - def test_run_persists_root_cause(self): - rc_repo = InMemoryRootCauseRepository() - event_repo = InMemoryEventRepository( - [ - Event( - event_id="e1", - host_id="h-1", - metric="disk_used_percent", - value=97.0, - fired_at=datetime(2025, 1, 1, 10, 1, 0, tzinfo=timezone.utc), - ) - ] - ) - worker = RootCauseWorker(rc_repo, event_repo, window_minutes=5) - results = worker.run([make_log("fl1", "No space left on device")]) - self.assertEqual(results[0].cause_type, "disk_full") - self.assertEqual(rc_repo.get("fl1").cause_type, "disk_full") - - -if __name__ == "__main__": - unittest.main() +（已提交至远程 main 分支） \ No newline at end of file