develop: 开发 threshold-event-detector 检测模块

README.md

@@ -1,2 +1,65 @@
 # fault-log-analyzer
 
+主机监控系统（HMS）故障日志捕获与分析模块。
+
+## 功能
+
+1. **故障日志捕获管道**：消费 `logs.raw`，按级别/关键字/正则过滤，结构化解析，
+   打标后写入 `logs.fault` 与 Elasticsearch（`hms-fault-log-{yyyy.MM}`）。
+2. **故障日志归类**：特征提取（模板化 + TF-IDF）→ 相似聚类（DBSCAN / 纯 Python 后端）
+   → 映射到 `fault_type`。
+3. **根因分析**：时间/主机关联指标事件 + 指标-故障规则库 + trace 关联，产出
+   `root_cause`（含 evidence 与 confidence）。
+4. **REST API**：故障日志查询、根因查询、故障类型/过滤规则管理接口。
+
+## 目录结构
+
+```
+src/fault_log_analyzer/
+  config.py       配置模型与环境变量加载
+  models.py       数据模型（FaultLog / RootCause / FaultType / ...）
+  parser.py       日志结构化解析与模板化
+  filters.py      故障过滤规则
+  fingerprint.py  MinHash 指纹与去重
+  cluster.py      TF-IDF 特征 + DBSCAN 聚类
+  classifier.py   簇 -> fault_type 归类
+  root_cause.py   根因分析
+  storage.py      存储抽象 + 内存实现（测试/离线）
+  integrations.py 可选真实后端（Kafka/ES/MySQL/Redis）
+  pipeline.py     故障捕获管道
+  workers.py      聚类 / 根因分析 worker
+  api.py          REST API（标准库 http.server 实现）
+  __main__.py     CLI 入口
+```
+
+## 快速开始
+
+```bash
+# 安装（核心零依赖，可选依赖见 requirements.txt）
+pip install -e .
+
+# 运行单元测试
+python -m unittest discover -s tests -v
+
+# 启动服务（内存存储，便于本地验证）
+python -m fault_log_analyzer --storage memory --api-host 127.0.0.1 --api-port 8080
+
+# 健康检查
+curl http://127.0.0.1:8080/healthz
+curl http://127.0.0.1:8080/api/v1/fault-logs
+```
+
+## 环境变量（生产）
+
+| 变量 | 默认值 | 说明 |
+|---|---|---|
+| `KAFKA_BOOTSTRAP_SERVERS` | `localhost:9092` | Kafka 地址 |
+| `KAFKA_GROUP_ID` | `fault-log-analyzer` | 消费者组 |
+| `KAFKA_LOGS_RAW_TOPIC` | `logs.raw` | 原始日志 topic |
+| `KAFKA_LOGS_FAULT_TOPIC` | `logs.fault` | 故障日志 topic |
+| `ES_HOSTS` | `http://localhost:9200` | ES 地址（逗号分隔） |
+| `MYSQL_HOST` / `MYSQL_PORT` / `MYSQL_USER` / `MYSQL_PASSWORD` / `MYSQL_DB` | localhost | MySQL 连接 |
+| `REDIS_URL` | `redis://localhost:6379/0` | Redis 地址 |
+| `CLUSTER_EPS` | `0.75` | DBSCAN eps（余弦距离） |
+| `CLUSTER_MIN_SAMPLES` | `5` | DBSCAN min_samples |
+| `ROOT_CAUSE_WINDOW_MINUTES` | `5` | 根因分析时间窗口 |

pyproject.toml

@@ -0,0 +1,31 @@
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "fault-log-analyzer"
+version = "0.1.0"
+description = "HMS fault log capture, classification and root cause analysis module"
+readme = "README.md"
+requires-python = ">=3.10"
+license = { text = "Apache-2.0" }
+authors = [{ name = "HMS Team" }]
+dependencies = []
+
+[project.optional-dependencies]
+kafka = ["kafka-python>=2.0.2"]
+elasticsearch = ["elasticsearch>=8.0.0"]
+mysql = ["pymysql>=1.1.0"]
+redis = ["redis>=4.5.0"]
+ml = ["numpy>=1.24.0", "scikit-learn>=1.2.0"]
+dev = ["pytest>=7.0.0"]
+
+[project.scripts]
+fault-log-analyzer = "fault_log_analyzer.__main__:main"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+pythonpath = ["src"]

requirements.txt

@@ -0,0 +1,8 @@
+# fault-log-analyzer 核心逻辑零第三方依赖（纯 Python 标准库），可直接运行。
+# 生产环境对接真实存储/消息总线时，按需安装以下可选依赖：
+kafka-python>=2.0.2        # Kafka 消费/生产
+elasticsearch>=8.0.0       # Elasticsearch 日志存储
+pymysql>=1.1.0             # MySQL 元数据/结果存储
+redis>=4.5.0               # 去重/缓存
+numpy>=1.24.0              # 可选：向量化加速（聚类）
+scikit-learn>=1.2.0        # 可选：生产级 DBSCAN/TF-IDF 后端

src/fault_log_analyzer/__init__.py

@@ -0,0 +1,28 @@
+"""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",
+]

src/fault_log_analyzer/classifier.py

@@ -0,0 +1,71 @@
+"""簇 -> 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"

src/fault_log_analyzer/cluster.py

@@ -0,0 +1,235 @@
+"""特征提取与日志聚类。
+
+核心为纯 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

src/fault_log_analyzer/config.py

@@ -0,0 +1,93 @@
+"""配置模型与环境变量加载。"""
+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

src/fault_log_analyzer/filters.py

@@ -0,0 +1,70 @@
+"""故障日志过滤规则与捕获过滤器。"""
+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

src/fault_log_analyzer/fingerprint.py

@@ -0,0 +1,76 @@
+"""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))

src/fault_log_analyzer/integrations.py

@@ -0,0 +1,358 @@
+"""可选真实后端适配（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))

src/fault_log_analyzer/models.py

@@ -0,0 +1,236 @@
+"""数据模型定义（纯标准库 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]:
+    if value is None or value == "":
+        return None
+    normalized = value.strip()
+    if normalized.endswith("Z"):
+        normalized = normalized[:-1] + "+00:00"
+    dt = datetime.fromisoformat(normalized)
+    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(),
+        )

src/fault_log_analyzer/parser.py

@@ -0,0 +1,142 @@
+"""日志结构化解析与模板化。"""
+from __future__ import annotations
+
+import json
+import re
+from datetime import datetime
+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 parse_timestamp(value: Any) -> Optional[datetime]:
+    """解析多种时间戳表达，失败返回 None。"""
+    if value is None:
+        return None
+    if isinstance(value, (int, float)):
+        # 秒/毫秒/纳秒时间戳
+        if value > 1e17:  # 纳秒
+            value = value / 1e9
+        elif value > 1e14:  # 毫秒
+            value = value / 1e3
+        try:
+            return datetime.utcfromtimestamp(value)
+        except (ValueError, OSError, OverflowError):
+            return None
+    if isinstance(value, datetime):
+        return 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 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("<UUID>", text)
+    text = _IP_RE.sub("<IP>", text)
+    text = _HEX_RE.sub("<HEX>", text)
+    text = _NUM_RE.sub("<NUM>", text)
+    text = _PATH_RE.sub("<PATH>", 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]

src/fault_log_analyzer/pipeline.py

@@ -0,0 +1,74 @@
+"""故障日志捕获管道：过滤 -> 解析 -> 去重 -> 打标 -> 落 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)

src/fault_log_analyzer/root_cause.py

@@ -0,0 +1,130 @@
+"""根因分析（辅助性）：规则优先 + 统计关联 + 置信度评分。"""
+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(),
+        )

src/fault_log_analyzer/storage.py

@@ -0,0 +1,227 @@
+"""存储抽象与内存实现。
+
+生产环境通过 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

src/fault_log_analyzer/workers.py

@@ -0,0 +1,78 @@
+"""聚类归类与根因分析 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