diff --git a/README.md b/README.md index faa933b..4582016 100644 --- a/README.md +++ b/README.md @@ -1,2 +1,87 @@ # threshold-event-detector +主机监控系统（HMS）指标阈值事件检测模块。 + +## 功能 + +1. **规则加载与热更新**：从 MySQL `metric_rule` / `rule_scope` 读取规则，写入 Redis + 缓存（`hms:rule:cache:{rule_id}`、`hms:rule:version`），后台线程按版本号轮询热加载。 +2. **检测引擎**：消费 `metrics.samples`，按 `(host_id, rule_id)` 维护滑动窗口与状态机 + （`OK → FIRING → PENDING → OK`），阈值比对触发 `firing` / `resolved` 事件。 +3. **事件写入与告警收敛**：事件写入 `event` 表；告警按 `(host, rule)` 去重、 + 按 `scope/rule/severity + 时间桶` 聚合，输出 `alert` 并发布到 `alerts.converged`。 +4. **通知渠道封装**：email / webhook / dingtalk / wechat 统一封装。 +5. **REST API**：阈值规则管理、事件告警查询与确认/关闭、样本接入。 + +## 技术栈 + +- Python 3.10+（标准库实现核心逻辑与 HTTP API，零强制第三方依赖）。 +- 可选依赖：`pymysql`（MySQL）、`redis`（Redis）、`kafka-python`（Kafka）。 + 未安装时自动回退到进程内内存实现，便于本地运行与单元测试。 + +## 目录结构 + +``` +threshold-event-detector/ +├── detector/ +│ ├── __init__.py # 包说明 +│ ├── __main__.py # 入口（python -m detector） +│ ├── models.py # 领域模型 +│ ├── config.py # 配置 +│ ├── storage.py # 存储/缓存/总线抽象 + 内存/MySQL/Redis/Kafka 适配器 +│ ├── rule_loader.py # 规则加载与热更新 +│ ├── engine.py # 检测引擎（滑动窗口 + 状态机） +│ ├── converger.py # 告警收敛（去重 + 聚合） +│ ├── notifier.py # 通知渠道封装 +│ ├── api.py # REST API +│ └── app.py # 应用装配 +└── tests/ # 单元测试 +``` + +## 运行 + +```bash +# 零依赖本地运行（进程内存储 + 示例规则） +python3 -m detector --addr 0.0.0.0:8080 + +# 健康检查 +curl http://127.0.0.1:8080/healthz + +# 查看规则 +curl http://127.0.0.1:8080/api/v1/rules + +# 样本接入（端到端联调） +curl -X POST http://127.0.0.1:8080/api/v1/ingest \ + -H 'Content-Type: application/json' \ + -d '{"host_id":"h-001","host_group":"web","samples":[ + {"name":"cpu_usage","value":95,"timestamp":1700000000,"labels":{"service":"web"}}]}' +``` + +## 测试 + +```bash +python3 -m unittest discover -s tests -v +``` + +## 配置（环境变量） + +| 变量 | 默认 | 说明 | +|---|---|---| +| `HMS_HTTP_ADDR` | `0.0.0.0:8080` | HTTP 监听地址 | +| `HMS_EVALUATION_INTERVAL` | `15s` | 评估/恢复观察周期 | +| `HMS_DEDUP_WINDOW` | `5m` | 告警去重窗口 | +| `HMS_AGGREGATE_WINDOW` | `1m` | 告警聚合窗口 | +| `HMS_RULE_STORE_DRIVER` | `memory` | 规则仓库驱动 `memory`/`mysql` | +| `HMS_EVENT_STORE_DRIVER` | `memory` | 事件仓库驱动 | +| `HMS_ALERT_STORE_DRIVER` | `memory` | 告警仓库驱动 | +| `HMS_CACHE_DRIVER` | `memory` | 缓存驱动 `memory`/`redis` | +| `HMS_BUS_DRIVER` | `memory` | 总线驱动 `memory`/`kafka` | +| `HMS_MYSQL_DSN` | 空 | MySQL DSN（`mysql://user:pass@host:port/hms`） | +| `HMS_REDIS_URL` | `redis://127.0.0.1:6379/0` | Redis URL | +| `HMS_KAFKA_BOOTSTRAP` | `127.0.0.1:9092` | Kafka bootstrap | + +## 设计文档 + +- `docs/01-design/architecture.md`（5.2 threshold-event-detector） +- `docs/01-design/database-design.md` +- `docs/01-design/api-design.md` diff --git a/detector/__init__.py b/detector/__init__.py new file mode 100644 index 0000000..3da8a53 --- /dev/null +++ b/detector/__init__.py @@ -0,0 +1,13 @@ +"""threshold-event-detector 指标阈值事件检测模块。 + +该包实现： +1) 阈值规则加载与配置（MySQL metric_rule/rule_scope + Redis 缓存，支持热加载）； +2) 检测引擎（对采集指标做阈值比对，状态机 OK/FIRING/PENDING 触发事件）； +3) 事件写入（event 表）与告警收敛（Redis 去重/聚合 + 通知渠道封装）； +4) REST API（阈值规则管理、事件告警查询）。 + +默认使用进程内存储（Memory），方便本地运行与单元测试； +MySQL / Redis / Kafka 适配器以可选依赖方式提供，接入真实中间件时启用。 +""" + +__version__ = "0.1.0" diff --git a/detector/__main__.py b/detector/__main__.py new file mode 100644 index 0000000..cae6d79 --- /dev/null +++ b/detector/__main__.py @@ -0,0 +1,54 @@ +"""threshold-event-detector 入口（python -m detector）。 + +用法： + python -m detector [--addr 0.0.0.0:8080] [--no-seed] + +默认以进程内存储启动（零依赖），写入示例规则并监听 REST API；可通过环境变量 +HMS_* 切换 MySQL / Redis / Kafka 驱动（需安装对应可选依赖）。 +""" + +from __future__ import annotations + +import argparse +import signal + +from .api import APIServer +from .app import DetectorApp +from .config import Config + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description="threshold-event-detector") + parser.add_argument("--addr", default=None, help="HTTP listen address, e.g. 0.0.0.0:8080") + parser.add_argument("--no-seed", action="store_true", help="do not seed demo rules") + args = parser.parse_args(argv) + + config = Config.from_env() + if args.addr: + config.http_addr = args.addr + + app = DetectorApp(config) + if not args.no_seed and config.seed_demo_rules: + app.seed_demo_rules() + app.start() + + api = APIServer(app, config.http_addr) + print(f"[threshold-event-detector] listening on http://{config.http_addr}", flush=True) + + def _stop(signum, frame): # noqa: ARG001 + print("\n[threshold-event-detector] shutting down...", flush=True) + api.shutdown() + app.stop() + + signal.signal(signal.SIGINT, _stop) + signal.signal(signal.SIGTERM, _stop) + + try: + api.serve_forever() + finally: + app.stop() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/detector/__pycache__/__init__.cpython-310.pyc b/detector/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..1190d6e Binary files /dev/null and b/detector/__pycache__/__init__.cpython-310.pyc differ diff --git a/detector/__pycache__/__main__.cpython-310.pyc b/detector/__pycache__/__main__.cpython-310.pyc new file mode 100644 index 0000000..cf8d79c Binary files /dev/null and b/detector/__pycache__/__main__.cpython-310.pyc differ diff --git a/detector/__pycache__/api.cpython-310.pyc b/detector/__pycache__/api.cpython-310.pyc new file mode 100644 index 0000000..f147dcd Binary files /dev/null and b/detector/__pycache__/api.cpython-310.pyc differ diff --git a/detector/__pycache__/app.cpython-310.pyc b/detector/__pycache__/app.cpython-310.pyc new file mode 100644 index 0000000..e876df5 Binary files /dev/null and b/detector/__pycache__/app.cpython-310.pyc differ diff --git a/detector/__pycache__/config.cpython-310.pyc b/detector/__pycache__/config.cpython-310.pyc new file mode 100644 index 0000000..1a6c2fa Binary files /dev/null and b/detector/__pycache__/config.cpython-310.pyc differ diff --git a/detector/__pycache__/converger.cpython-310.pyc b/detector/__pycache__/converger.cpython-310.pyc new file mode 100644 index 0000000..3a2dc66 Binary files /dev/null and b/detector/__pycache__/converger.cpython-310.pyc differ diff --git a/detector/__pycache__/engine.cpython-310.pyc b/detector/__pycache__/engine.cpython-310.pyc new file mode 100644 index 0000000..e388714 Binary files /dev/null and b/detector/__pycache__/engine.cpython-310.pyc differ diff --git a/detector/__pycache__/models.cpython-310.pyc b/detector/__pycache__/models.cpython-310.pyc new file mode 100644 index 0000000..f6747ff Binary files /dev/null and b/detector/__pycache__/models.cpython-310.pyc differ diff --git a/detector/__pycache__/notifier.cpython-310.pyc b/detector/__pycache__/notifier.cpython-310.pyc new file mode 100644 index 0000000..7a67356 Binary files /dev/null and b/detector/__pycache__/notifier.cpython-310.pyc differ diff --git a/detector/__pycache__/rule_loader.cpython-310.pyc b/detector/__pycache__/rule_loader.cpython-310.pyc new file mode 100644 index 0000000..4674d88 Binary files /dev/null and b/detector/__pycache__/rule_loader.cpython-310.pyc differ diff --git a/detector/__pycache__/storage.cpython-310.pyc b/detector/__pycache__/storage.cpython-310.pyc new file mode 100644 index 0000000..3911ced Binary files /dev/null and b/detector/__pycache__/storage.cpython-310.pyc differ diff --git a/detector/api.py b/detector/api.py new file mode 100644 index 0000000..4cf3e8c --- /dev/null +++ b/detector/api.py @@ -0,0 +1,301 @@ +"""REST API（stdlib http.server 实现，零第三方依赖）。 + +接口覆盖 api-design.md 中与本模块相关的部分： +- 健康检查：GET /healthz、GET /readyz +- 规则管理：GET/POST /api/v1/rules、GET/PUT/DELETE /api/v1/rules/{id}、 + POST /api/v1/rules/{id}/enable|disable +- 事件查询：GET /api/v1/events +- 告警查询与管理：GET /api/v1/alerts、GET /api/v1/alerts/{id}、 + POST /api/v1/alerts/{id}/ack、POST /api/v1/alerts/{id}/close、 + POST /api/v1/alerts/batch-ack +- 内部样本接入：POST /api/v1/ingest（供无 Kafka 环境端到端联调） +""" + +from __future__ import annotations + +import json +import re +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any, Callable, Dict, List, Optional, Tuple +from urllib.parse import urlparse + +from .app import DetectorApp +from .models import MetricRule, MetricSample + +# 路由条目：(method, path_regex, handler) +Route = Tuple[str, str, Callable] + + +class APIServer: + def __init__(self, app: DetectorApp, addr: str = "0.0.0.0:8080") -> None: + self.app = app + host, port = self._split_addr(addr) + self.server = ThreadingHTTPServer((host, port), self._handler_factory()) + self.routes: List[Route] = [ + ("GET", r"^/healthz$", self.handle_healthz), + ("GET", r"^/readyz$", self.handle_readyz), + ("GET", r"^/api/v1/rules$", self.handle_list_rules), + ("POST", r"^/api/v1/rules$", self.handle_create_rule), + ("GET", r"^/api/v1/rules/([^/]+)$", self.handle_get_rule), + ("PUT", r"^/api/v1/rules/([^/]+)$", self.handle_update_rule), + ("DELETE", r"^/api/v1/rules/([^/]+)$", self.handle_delete_rule), + ("POST", r"^/api/v1/rules/([^/]+)/enable$", self.handle_enable_rule), + ("POST", r"^/api/v1/rules/([^/]+)/disable$", self.handle_disable_rule), + ("GET", r"^/api/v1/events$", self.handle_list_events), + ("GET", r"^/api/v1/alerts$", self.handle_list_alerts), + ("POST", r"^/api/v1/alerts/batch-ack$", self.handle_batch_ack), + ("GET", r"^/api/v1/alerts/([^/]+)$", self.handle_get_alert), + ("POST", r"^/api/v1/alerts/([^/]+)/ack$", self.handle_ack_alert), + ("POST", r"^/api/v1/alerts/([^/]+)/close$", self.handle_close_alert), + ("POST", r"^/api/v1/ingest$", self.handle_ingest), + ] + + @staticmethod + def _split_addr(addr: str) -> Tuple[str, int]: + if ":" in addr: + host, port = addr.rsplit(":", 1) + return host, int(port) + return addr, 8080 + + def _handler_factory(self) -> type: + api = self + + class Handler(BaseHTTPRequestHandler): + server_version = "ThresholdEventDetector/0.1" + + def do_GET(self): # noqa: N802 + self._dispatch("GET") + + def do_POST(self): # noqa: N802 + self._dispatch("POST") + + def do_PUT(self): # noqa: N802 + self._dispatch("PUT") + + def do_DELETE(self): # noqa: N802 + self._dispatch("DELETE") + + def _dispatch(self, method: str) -> None: + parsed = urlparse(self.path) + for route_method, pattern, handler in api.routes: + if route_method != method: + continue + m = re.match(pattern, parsed.path) + if not m: + continue + try: + handler(self, *m.groups()) + except _APIError as exc: + self._write_json(exc.status, {"code": exc.code, "message": exc.message, "data": None}) + except Exception as exc: # noqa: BLE001 + self._write_json(500, {"code": 50000, "message": f"internal error: {exc}", "data": None}) + return + self._write_json(404, {"code": 40400, "message": "not found", "data": None}) + + def _read_json(self) -> Dict[str, Any]: + length = int(self.headers.get("Content-Length") or 0) + if length <= 0: + return {} + raw = self.rfile.read(length) + return json.loads(raw.decode("utf-8")) + + def _query(self) -> Dict[str, str]: + parsed = urlparse(self.path) + result: Dict[str, str] = {} + if parsed.query: + for pair in parsed.query.split("&"): + if "=" in pair: + k, v = pair.split("=", 1) + result[k] = v + else: + result[pair] = "" + return result + + def _write_json(self, status: int, body: Dict[str, Any]) -> None: + data = json.dumps(body, 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(data))) + self.end_headers() + self.wfile.write(data) + + def log_message(self, format: str, *args) -> None: # noqa: A002 + # 静默默认请求日志，便于测试输出整洁；生产可重写为 logging。 + pass + + return Handler + + def serve_forever(self) -> None: + self.server.serve_forever() + + def shutdown(self) -> None: + self.server.shutdown() + + # ------------------------------------------------------------------ # + # 健康检查 + # ------------------------------------------------------------------ # + def handle_healthz(self, handler) -> None: + handler._write_json(200, {"code": 0, "message": "ok", "data": {"status": "ok"}}) + + def handle_readyz(self, handler) -> None: + handler._write_json(200, {"code": 0, "message": "ok", "data": {"status": "ok", "deps": {"rule_store": "up"}}}) + + # ------------------------------------------------------------------ # + # 规则管理 + # ------------------------------------------------------------------ # + def handle_list_rules(self, handler) -> None: + q = handler._query() + rules = self.app.rule_loader.get_rules() + if q.get("metric"): + rules = [r for r in rules if r.metric == q["metric"]] + if q.get("enabled") in ("true", "false"): + enabled = q["enabled"] == "true" + rules = [r for r in rules if r.enabled == enabled] + if q.get("severity"): + rules = [r for r in rules if r.severity == q["severity"]] + page = int(q.get("page", "1")) + page_size = min(int(q.get("page_size", "20")), 200) + total = len(rules) + start = (page - 1) * page_size + items = [r.to_dict() for r in rules[start : start + page_size]] + handler._write_json(200, {"code": 0, "message": "ok", "data": {"total": total, "items": items}}) + + def handle_get_rule(self, handler, rule_id: str) -> None: + rule = self.app.rule_loader.get_rule(rule_id) + if rule is None: + raise _APIError(404, 40400, f"rule {rule_id} not found") + handler._write_json(200, {"code": 0, "message": "ok", "data": rule.to_dict()}) + + def handle_create_rule(self, handler) -> None: + data = handler._read_json() + rule = MetricRule.from_dict(data) + if not rule.rule_id: + raise _APIError(400, 40001, "rule_id is required") + created = self.app.rule_store.create_rule(rule) + self.app.rule_loader.load() + handler._write_json(200, {"code": 0, "message": "ok", "data": created.to_dict()}) + + def handle_update_rule(self, handler, rule_id: str) -> None: + data = handler._read_json() + data["rule_id"] = rule_id + updated = self.app.rule_store.update_rule(MetricRule.from_dict(data)) + self.app.rule_loader.load() + handler._write_json(200, {"code": 0, "message": "ok", "data": updated.to_dict()}) + + def handle_delete_rule(self, handler, rule_id: str) -> None: + deleted = self.app.rule_store.delete_rule(rule_id) + self.app.rule_loader.load() + handler._write_json(200, {"code": 0, "message": "ok", "data": {"deleted": deleted}}) + + def handle_enable_rule(self, handler, rule_id: str) -> None: + rule = self._set_rule_enabled(rule_id, True) + handler._write_json(200, {"code": 0, "message": "ok", "data": {"enabled": rule.enabled}}) + + def handle_disable_rule(self, handler, rule_id: str) -> None: + rule = self._set_rule_enabled(rule_id, False) + handler._write_json(200, {"code": 0, "message": "ok", "data": {"enabled": rule.enabled}}) + + def _set_rule_enabled(self, rule_id: str, enabled: bool) -> MetricRule: + rule = self.app.rule_loader.get_rule(rule_id) + if rule is None: + raise _APIError(404, 40400, f"rule {rule_id} not found") + rule.enabled = enabled + self.app.rule_store.update_rule(rule) + self.app.rule_loader.load() + return rule + + # ------------------------------------------------------------------ # + # 事件 / 告警查询与管理 + # ------------------------------------------------------------------ # + def handle_list_events(self, handler) -> None: + q = handler._query() + total, items = self.app.event_store.list_events( + host_id=q.get("host_id"), + rule_id=q.get("rule_id"), + status=q.get("status"), + severity=q.get("severity"), + start=_parse_float(q.get("from")), + end=_parse_float(q.get("to")), + page=int(q.get("page", "1")), + page_size=min(int(q.get("page_size", "20")), 200), + ) + handler._write_json(200, {"code": 0, "message": "ok", "data": {"total": total, "items": [e.to_dict() for e in items]}}) + + def handle_list_alerts(self, handler) -> None: + q = handler._query() + total, items = self.app.alert_store.list_alerts( + severity=q.get("severity"), + status=q.get("status"), + ack_status=q.get("ack_status"), + start=_parse_float(q.get("from")), + end=_parse_float(q.get("to")), + page=int(q.get("page", "1")), + page_size=min(int(q.get("page_size", "20")), 200), + ) + handler._write_json(200, {"code": 0, "message": "ok", "data": {"total": total, "items": [a.to_dict() for a in items]}}) + + def handle_get_alert(self, handler, alert_id: str) -> None: + alert = self.app.alert_store.get(alert_id) + if alert is None: + raise _APIError(404, 40400, f"alert {alert_id} not found") + handler._write_json(200, {"code": 0, "message": "ok", "data": alert.to_dict()}) + + def handle_ack_alert(self, handler, alert_id: str) -> None: + body = handler._read_json() + alert = self.app.alert_store.ack(alert_id, body.get("ack_by") or "system") + if alert is None: + raise _APIError(404, 40400, f"alert {alert_id} not found") + handler._write_json(200, {"code": 0, "message": "ok", "data": alert.to_dict()}) + + def handle_close_alert(self, handler, alert_id: str) -> None: + alert = self.app.alert_store.close(alert_id) + if alert is None: + raise _APIError(404, 40400, f"alert {alert_id} not found") + handler._write_json(200, {"code": 0, "message": "ok", "data": alert.to_dict()}) + + def handle_batch_ack(self, handler) -> None: + body = handler._read_json() + ids = body.get("alert_ids") or [] + acked = 0 + for alert_id in ids: + if self.app.alert_store.ack(str(alert_id)) is not None: + acked += 1 + handler._write_json(200, {"code": 0, "message": "ok", "data": {"acked": acked}}) + + # ------------------------------------------------------------------ # + # 样本接入（端到端联调） + # ------------------------------------------------------------------ # + def handle_ingest(self, handler) -> None: + body = handler._read_json() + host_id = str(body.get("host_id") or "") + if not host_id: + raise _APIError(400, 40001, "host_id is required") + host_group = body.get("host_group") + samples = [MetricSample.from_dict(s) for s in (body.get("samples") or [])] + events = self.app.ingest(host_id, host_group, samples) + handler._write_json(200, {"code": 0, "message": "ok", "data": {"events": [e.to_dict() for e in events]}}) + + +class _APIError(Exception): + def __init__(self, status: int, code: int, message: str) -> None: + super().__init__(message) + self.status = status + self.code = code + self.message = message + + +def _parse_float(value: Optional[str]) -> Optional[float]: + if value in (None, ""): + return None + try: + return float(value) + except ValueError: + return None + + +def serve(app: DetectorApp, addr: str = "0.0.0.0:8080") -> APIServer: + api = APIServer(app, addr) + api.serve_forever() + return api diff --git a/detector/app.py b/detector/app.py new file mode 100644 index 0000000..d1d0be5 --- /dev/null +++ b/detector/app.py @@ -0,0 +1,124 @@ +"""应用装配：把规则加载、检测引擎、事件写入、收敛、通知与总线串起来。 + +DetectorApp 同时承载 REST API 所需的各 store 引用。 +""" + +from __future__ import annotations + +from typing import List, Optional + +from .config import Config +from .converger import AlertConverger +from .engine import DetectionEngine +from .models import MetricRule, MetricSample, RuleScope +from .notifier import Notifier +from .rule_loader import RuleLoader +from .storage import ( + AlertStore, + Cache, + EventStore, + InMemoryMessageBus, + MessageBus, + RuleStore, + create_alert_store, + create_bus, + create_cache, + create_event_store, + create_rule_store, +) + + +class DetectorApp: + def __init__(self, config: Optional[Config] = None) -> None: + self.config = config or Config.from_env() + self.rule_store: RuleStore = create_rule_store(self.config) + self.event_store: EventStore = create_event_store(self.config) + self.alert_store: AlertStore = create_alert_store(self.config) + self.cache: Cache = create_cache(self.config) + self.bus: MessageBus = create_bus(self.config) + + self.rule_loader = RuleLoader(self.rule_store, self.cache, self.config.rule_reload_interval) + self.notifier = Notifier.memory() + self.converger = AlertConverger( + alert_store=self.alert_store, + cache=self.cache, + bus=self.bus, + dedup_window=self.config.dedup_window, + aggregate_window=self.config.aggregate_window, + alerts_topic=self.config.alerts_topic, + notifier=self.notifier, + ) + self.engine = DetectionEngine( + rule_loader=self.rule_loader, + evaluation_interval=self.config.evaluation_interval, + recover_duration=self.config.recover_duration, + on_event=self._on_event, + ) + + def _on_event(self, event) -> None: + # 事件先落库，再做收敛（收敛可能因去重被抑制）。 + self.event_store.insert(event) + self.converger.handle_event(event) + + # ------------------------------------------------------------------ # + def start(self) -> None: + self.rule_loader.load() + self.rule_loader.start() + + def stop(self) -> None: + self.rule_loader.stop() + + def seed_demo_rules(self) -> List[MetricRule]: + """写入示例规则，便于本地联调/演示。""" + demos = [ + MetricRule( + rule_id="r-cpu-high", + name="CPU 使用率过高", + metric="cpu_usage", + aggregation="avg", + operator="gt", + threshold=90.0, + for_duration="60s", + severity="critical", + scope=RuleScope(scope_type="all"), + notify_channels=["email", "webhook"], + enabled=True, + ), + MetricRule( + rule_id="r-mem-high", + name="内存使用率过高", + metric="mem_used_percent", + aggregation="avg", + operator="gt", + threshold=85.0, + for_duration="60s", + severity="warning", + scope=RuleScope(scope_type="host_group", host_group="web"), + notify_channels=["email"], + enabled=True, + ), + MetricRule( + rule_id="r-disk-full", + name="磁盘使用率过高", + metric="disk_used_percent", + aggregation="avg", + operator="gt", + threshold=90.0, + for_duration="30s", + severity="critical", + scope=RuleScope(scope_type="all"), + notify_channels=["webhook"], + enabled=True, + ), + ] + created = [] + for rule in demos: + if self.rule_store.get_rule(rule.rule_id) is None: + created.append(self.rule_store.create_rule(rule)) + self.rule_loader.load() + return created + + # ------------------------------------------------------------------ # + def ingest(self, host_id: str, host_group: Optional[str], samples: List[MetricSample]): + """接收一批样本并驱动检测，返回触发的事件。""" + return self.engine.handle_batch(samples, host_id, host_group) diff --git a/detector/config.py b/detector/config.py new file mode 100644 index 0000000..07d7e62 --- /dev/null +++ b/detector/config.py @@ -0,0 +1,70 @@ +"""运行配置。 + +支持从环境变量读取（HMS_ 前缀），未配置时使用安全默认值，保证模块开箱即用。 +""" + +from __future__ import annotations + +import os +from dataclasses import dataclass + + +def _env(name: str, default: str) -> str: + return os.environ.get(name, default) + + +@dataclass +class Config: + """检测模块配置。""" + + http_addr: str = "0.0.0.0:8080" + + # 检测引擎 + evaluation_interval: str = "15s" # 评估周期（PENDING 恢复观察默认时长） + recover_duration: str = "15s" # 恢复观察窗口 + + # 告警收敛 + dedup_window: str = "5m" # 去重窗口（同 host+rule 抑制） + aggregate_window: str = "1m" # 聚合窗口 + + # 规则热加载 + rule_reload_interval: str = "10s" + + # 存储驱动：memory / mysql / redis（可选依赖未安装时自动回退 memory） + rule_store_driver: str = "memory" + event_store_driver: str = "memory" + alert_store_driver: str = "memory" + cache_driver: str = "memory" + bus_driver: str = "memory" + + # 中间件连接配置（接入真实中间件时使用） + mysql_dsn: str = "" + redis_url: str = "redis://127.0.0.1:6379/0" + kafka_bootstrap: str = "127.0.0.1:9092" + metrics_topic: str = "metrics.samples" + alerts_topic: str = "alerts.converged" + + # 是否在启动时写入示例规则（便于本地联调/演示） + seed_demo_rules: bool = True + + @classmethod + def from_env(cls) -> "Config": + return cls( + http_addr=_env("HMS_HTTP_ADDR", "0.0.0.0:8080"), + evaluation_interval=_env("HMS_EVALUATION_INTERVAL", "15s"), + recover_duration=_env("HMS_RECOVER_DURATION", "15s"), + dedup_window=_env("HMS_DEDUP_WINDOW", "5m"), + aggregate_window=_env("HMS_AGGREGATE_WINDOW", "1m"), + rule_reload_interval=_env("HMS_RULE_RELOAD_INTERVAL", "10s"), + rule_store_driver=_env("HMS_RULE_STORE_DRIVER", "memory"), + event_store_driver=_env("HMS_EVENT_STORE_DRIVER", "memory"), + alert_store_driver=_env("HMS_ALERT_STORE_DRIVER", "memory"), + cache_driver=_env("HMS_CACHE_DRIVER", "memory"), + bus_driver=_env("HMS_BUS_DRIVER", "memory"), + mysql_dsn=_env("HMS_MYSQL_DSN", ""), + redis_url=_env("HMS_REDIS_URL", "redis://127.0.0.1:6379/0"), + kafka_bootstrap=_env("HMS_KAFKA_BOOTSTRAP", "127.0.0.1:9092"), + metrics_topic=_env("HMS_METRICS_TOPIC", "metrics.samples"), + alerts_topic=_env("HMS_ALERTS_TOPIC", "alerts.converged"), + seed_demo_rules=_env("HMS_SEED_DEMO_RULES", "true").lower() in ("1", "true", "yes"), + ) diff --git a/detector/converger.py b/detector/converger.py new file mode 100644 index 0000000..a6f9f61 --- /dev/null +++ b/detector/converger.py @@ -0,0 +1,101 @@ +"""告警收敛：去重 + 聚合。 + +- 去重：同一 (host_id, rule_id) 在 dedup_window 内不重复产生同类新告警 + （Redis SET key NX EX dedup_window，InMemory 语义等价）。 +- 聚合：同一 aggregate_key（scope/rule/severity + 时间桶）在 aggregate_window + 内多主机同类告警合并为一条，aggregated_count 累加，明细 hosts 列表合并。 +- 收敛后的 alert 写入 AlertStore，并发布到 alerts.converged 总线，由通知分发器消费。 +""" + +from __future__ import annotations + +import uuid +from typing import Callable, Dict, List, Optional + +from .models import Alert, Event, parse_duration +from .storage import AlertStore, Cache, MessageBus + +DEDUP_PREFIX = "hms:dedup:" + + +class AlertConverger: + def __init__( + self, + alert_store: AlertStore, + cache: Cache, + bus: MessageBus, + dedup_window: str = "5m", + aggregate_window: str = "1m", + alerts_topic: str = "alerts.converged", + host_group_resolver: Optional[Callable[[str], str]] = None, + notifier: Optional[object] = None, + ) -> None: + self.alert_store = alert_store + self.cache = cache + self.bus = bus + self.dedup_window = parse_duration(dedup_window) or 300 + self.aggregate_window = parse_duration(aggregate_window) or 60 + self.alerts_topic = alerts_topic + self.host_group_resolver = host_group_resolver or (lambda host_id: "all") + self.notifier = notifier + + def handle_event(self, event: Event) -> Optional[Alert]: + if event.status == "firing": + return self._handle_firing(event) + if event.status == "resolved": + return self._handle_resolved(event) + return None + + def _handle_firing(self, event: Event) -> Optional[Alert]: + dedup_key = f"{event.host_id}:{event.rule_id}" + if not self.cache.set_nx(DEDUP_PREFIX + dedup_key, "1", ttl=self.dedup_window): + # 去重窗口内重复事件，抑制新告警 + return None + + bucket = int(event.fired_at // self.aggregate_window) + group = self.host_group_resolver(event.host_id) + aggregate_key = f"{group}:{event.rule_id}:{event.severity}:{bucket}" + + alert = Alert( + alert_id=f"a-{uuid.uuid4().hex[:16]}", + dedup_key=dedup_key, + aggregate_key=aggregate_key, + severity=event.severity, + status="firing", + title=self._title(event), + detail={ + "hosts": [event.host_id], + "current": event.agg_value, + "threshold": event.threshold, + "metric": event.metric, + "rule_id": event.rule_id, + }, + count=1, + first_at=event.fired_at, + last_at=event.fired_at, + ) + stored = self.alert_store.upsert_firing(alert) + self._publish(stored) + if self.notifier is not None and hasattr(self.notifier, "notify"): + self.notifier.notify(stored, getattr(stored, "notify_channels", None) or self._channels(event)) + return stored + + def _handle_resolved(self, event: Event) -> Optional[Alert]: + dedup_key = f"{event.host_id}:{event.rule_id}" + # 恢复事件允许后续再次触发去重键失效 + self.cache.delete(DEDUP_PREFIX + dedup_key) + alert = self.alert_store.resolve_by_dedup_key(dedup_key, event.fired_at) + if alert is not None: + self._publish(alert) + return alert + + def _publish(self, alert: Alert) -> None: + self.bus.publish(self.alerts_topic, alert.aggregate_key, alert.to_dict()) + + @staticmethod + def _title(event: Event) -> str: + return f"{event.metric} {event.operator} {event.threshold} ({event.status})" + + @staticmethod + def _channels(event: Event) -> List[str]: + return [] diff --git a/detector/engine.py b/detector/engine.py new file mode 100644 index 0000000..75c98f8 --- /dev/null +++ b/detector/engine.py @@ -0,0 +1,252 @@ +"""检测引擎：滑动窗口聚合 + 状态机（OK → FIRING → PENDING → OK）。 + +对每个 (host_id, rule_id) 维护： +- 滑动窗口（保存 for_duration 窗口内的样本，用于计算聚合值）； +- 状态机（追踪条件持续满足/不满足时间，抑制瞬时毛刺）。 + +状态跃迁与 architecture.md 5.2.3 对齐： +- OK → FIRING：条件持续满足 ForDuration，触发 firing 事件； +- FIRING → PENDING：最近一次评估不再满足，进入恢复观察； +- PENDING → OK：观察期（recover_duration）内持续不满足，触发 resolved 事件； +- PENDING → FIRING：观察期内再次满足，回到 FIRING（不重复触发事件）。 +""" + +from __future__ import annotations + +import threading +import uuid +from collections import deque +from dataclasses import dataclass, field +from typing import Callable, Dict, List, Optional, Tuple + +from .models import Event, MetricRule, MetricSample, parse_duration +from .rule_loader import RuleLoader + + +def aggregate_value(values: List[float], aggregation: str) -> float: + if not values: + return 0.0 + agg = (aggregation or "avg").lower() + if agg == "avg": + return sum(values) / len(values) + if agg == "min": + return min(values) + if agg == "max": + return max(values) + if agg == "sum": + return sum(values) + if agg == "last": + return values[-1] + raise ValueError(f"unknown aggregation: {aggregation}") + + +def compare(value: float, operator: str, threshold: float, threshold2: Optional[float] = None) -> bool: + op = (operator or "gt").lower() + eps = 1e-9 + if op == "gt": + return value > threshold + if op == "gte": + return value >= threshold + if op == "lt": + return value < threshold + if op == "lte": + return value <= threshold + if op == "eq": + return abs(value - threshold) <= eps + if op == "neq": + return abs(value - threshold) > eps + if op == "between": + lo = threshold + hi = threshold2 if threshold2 is not None else threshold + if lo > hi: + lo, hi = hi, lo + return lo <= value <= hi + raise ValueError(f"unknown operator: {operator}") + + +@dataclass +class _State: + status: str = "ok" # ok | firing | pending + satisfied_since: Optional[float] = None + not_satisfied_since: Optional[float] = None + last_value: float = 0.0 + last_eval: float = 0.0 + + +class SlidingWindow: + """按时间衰减的样本窗口。""" + + def __init__(self, max_age: float) -> None: + self.max_age = max_age + self.samples: deque = deque() # (timestamp, value) + + def add(self, timestamp: float, value: float) -> None: + self.samples.append((timestamp, value)) + self._prune(timestamp) + + def _prune(self, now: float) -> None: + while self.samples and (now - self.samples[0][0]) > self.max_age: + self.samples.popleft() + + def values(self) -> List[float]: + return [v for _, v in self.samples] + + def clear(self) -> None: + self.samples.clear() + + +class DetectionEngine: + def __init__( + self, + rule_loader: RuleLoader, + evaluation_interval: str = "15s", + recover_duration: str = "15s", + on_event: Optional[Callable[[Event], None]] = None, + ) -> None: + self.rule_loader = rule_loader + self.evaluation_interval = parse_duration(evaluation_interval) or 15 + self.recover_duration = parse_duration(recover_duration) or self.evaluation_interval + self.on_event = on_event + self._windows: Dict[Tuple[str, str], SlidingWindow] = {} + self._states: Dict[Tuple[str, str], _State] = {} + self._lock = threading.RLock() + + # ------------------------------------------------------------------ # + # 入口 + # ------------------------------------------------------------------ # + def handle( + self, + sample: MetricSample, + host_id: str, + host_group: Optional[str] = None, + now: Optional[float] = None, + ) -> List[Event]: + """处理单条样本，返回本次触发的所有事件。""" + now = sample.timestamp if now is None else now + events: List[Event] = [] + rules = self.rule_loader.match(sample, host_id, host_group) + with self._lock: + for rule in rules: + event = self._evaluate(host_id, rule, sample, now) + if event is not None: + events.append(event) + if self.on_event: + for event in events: + self.on_event(event) + return events + + def handle_batch( + self, + samples: List[MetricSample], + host_id: str, + host_group: Optional[str] = None, + ) -> List[Event]: + events: List[Event] = [] + for sample in samples: + events.extend(self.handle(sample, host_id, host_group)) + return events + + # ------------------------------------------------------------------ # + # 单规则评估 + # ------------------------------------------------------------------ # + def _evaluate( + self, + host_id: str, + rule: MetricRule, + sample: MetricSample, + now: float, + ) -> Optional[Event]: + key = (host_id, rule.rule_id) + for_duration = parse_duration(rule.for_duration) or 60 + + window = self._windows.get(key) + if window is None or window.max_age != for_duration: + window = SlidingWindow(for_duration) + self._windows[key] = window + window.add(sample.timestamp, sample.value) + + state = self._states.get(key) + if state is None: + state = _State() + self._states[key] = state + + value = aggregate_value(window.values(), rule.aggregation) + satisfied = compare(value, rule.operator, rule.threshold, rule.threshold2) + state.last_value = value + state.last_eval = now + + event: Optional[Event] = None + if state.status == "ok": + if satisfied: + if state.satisfied_since is None: + state.satisfied_since = now + if now - state.satisfied_since >= for_duration: + state.status = "firing" + state.not_satisfied_since = None + event = self._make_event(host_id, rule, sample, value, "firing", now) + else: + state.satisfied_since = None + elif state.status == "firing": + if not satisfied: + state.status = "pending" + state.not_satisfied_since = now + elif state.status == "pending": + if satisfied: + state.status = "firing" + state.not_satisfied_since = None + # 回到 FIRING 不重复告警（仅更新计数），与设计一致 + else: + if state.not_satisfied_since is None: + state.not_satisfied_since = now + if now - state.not_satisfied_since >= self.recover_duration: + state.status = "ok" + state.satisfied_since = None + event = self._make_event(host_id, rule, sample, value, "resolved", now) + + self._states[key] = state + return event + + def _make_event( + self, + host_id: str, + rule: MetricRule, + sample: MetricSample, + value: float, + status: str, + now: float, + ) -> Event: + return Event( + event_id=f"e-{uuid.uuid4().hex[:16]}", + host_id=host_id, + rule_id=rule.rule_id, + metric=rule.metric, + agg_value=value, + threshold=rule.threshold, + operator=rule.operator, + status=status, + severity=rule.severity, + fired_at=now, + labels=dict(sample.labels), + ) + + # ------------------------------------------------------------------ # + # 状态查询（测试/可观测） + # ------------------------------------------------------------------ # + def state_of(self, host_id: str, rule_id: str) -> Optional[Dict[str, object]]: + key = (host_id, rule_id) + with self._lock: + state = self._states.get(key) + if state is None: + return None + return { + "status": state.status, + "satisfied_since": state.satisfied_since, + "not_satisfied_since": state.not_satisfied_since, + "last_value": state.last_value, + "last_eval": state.last_eval, + } + + def reset(self) -> None: + with self._lock: + self._windows.clear() + self._states.clear() diff --git a/detector/models.py b/detector/models.py new file mode 100644 index 0000000..fca3db2 --- /dev/null +++ b/detector/models.py @@ -0,0 +1,238 @@ +"""领域模型：指标样本、阈值规则、作用范围、事件与告警。 + +字段命名与 docs/01-design 的 architecture.md / database-design.md / api-design.md 对齐； +对外 JSON 使用 camelCase，内部 Python 使用 snake_case。 +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +def parse_duration(value: Optional[str]) -> int: + """将 '60s' / '5m' / '1h' / '1h30m' 等时长解析为秒。 + + 纯数字按秒处理；空值返回 0。 + """ + if value is None or value == "": + return 0 + text = str(value).strip().lower() + if text.isdigit(): + return int(text) + + total = 0.0 + num = "" + for ch in text: + if ch.isdigit() or ch == ".": + num += ch + elif ch in "smhd": + if not num: + raise ValueError(f"invalid duration: {value!r}") + n = float(num) + if ch == "s": + total += n + elif ch == "m": + total += n * 60 + elif ch == "h": + total += n * 3600 + elif ch == "d": + total += n * 86400 + num = "" + else: + raise ValueError(f"invalid duration: {value!r}") + if num: + total += float(num) + return int(total) + + +@dataclass +class MetricSample: + """规范化后的单条指标样本（对应 metrics.samples 消息）。""" + + name: str + value: float + timestamp: float + labels: Dict[str, str] = field(default_factory=dict) + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "MetricSample": + return cls( + name=str(data["name"]), + value=float(data["value"]), + timestamp=float(data.get("timestamp", 0)), + labels=dict(data.get("labels") or {}), + ) + + def to_dict(self) -> Dict[str, Any]: + return { + "name": self.name, + "value": self.value, + "timestamp": self.timestamp, + "labels": self.labels, + } + + +@dataclass +class RuleScope: + """规则作用范围（对应 rule_scope 表）。""" + + scope_type: str = "all" # all | host_ids | host_group + host_ids: List[str] = field(default_factory=list) + host_group: str = "" + service: str = "" + + def matches(self, host_id: str, host_group: Optional[str], service: Optional[str]) -> bool: + if self.scope_type == "all": + return True + if self.scope_type == "host_ids": + return host_id in self.host_ids + if self.scope_type == "host_group": + if self.host_group and host_group and self.host_group == host_group: + return True + return False + return False + + def to_dict(self) -> Dict[str, Any]: + return { + "scope_type": self.scope_type, + "host_ids": list(self.host_ids), + "host_group": self.host_group, + "service": self.service, + } + + @classmethod + def from_dict(cls, data: Optional[Dict[str, Any]]) -> "RuleScope": + if not data: + return cls() + return cls( + scope_type=str(data.get("scope_type") or "all"), + host_ids=[str(x) for x in (data.get("host_ids") or [])], + host_group=str(data.get("host_group") or ""), + service=str(data.get("service") or ""), + ) + + +@dataclass +class MetricRule: + """阈值规则（对应 metric_rule 表）。""" + + rule_id: str + name: str + metric: str + aggregation: str = "avg" # avg | min | max | sum | last + operator: str = "gt" # gt | gte | lt | lte | eq | neq | between + threshold: float = 0.0 + threshold2: Optional[float] = None + for_duration: str = "60s" + severity: str = "warning" + scope: RuleScope = field(default_factory=RuleScope) + labels: Dict[str, str] = field(default_factory=dict) + notify_channels: List[str] = field(default_factory=list) + enabled: bool = True + version: int = 0 + + def to_dict(self) -> Dict[str, Any]: + return { + "rule_id": self.rule_id, + "name": self.name, + "metric": self.metric, + "aggregation": self.aggregation, + "operator": self.operator, + "threshold": self.threshold, + "threshold2": self.threshold2, + "for_duration": self.for_duration, + "severity": self.severity, + "scope": self.scope.to_dict(), + "labels": dict(self.labels), + "notify_channels": list(self.notify_channels), + "enabled": self.enabled, + "version": self.version, + } + + @classmethod + def from_dict(cls, data: Dict[str, Any]) -> "MetricRule": + return cls( + rule_id=str(data["rule_id"]), + name=str(data.get("name") or ""), + metric=str(data.get("metric") or ""), + aggregation=str(data.get("aggregation") or "avg"), + operator=str(data.get("operator") or "gt"), + threshold=float(data.get("threshold") or 0.0), + threshold2=float(data["threshold2"]) if data.get("threshold2") is not None else None, + for_duration=str(data.get("for_duration") or "60s"), + severity=str(data.get("severity") or "warning"), + scope=RuleScope.from_dict(data.get("scope")), + labels={str(k): str(v) for k, v in (data.get("labels") or {}).items()}, + notify_channels=[str(x) for x in (data.get("notify_channels") or [])], + enabled=bool(data.get("enabled", True)), + version=int(data.get("version") or 0), + ) + + +@dataclass +class Event: + """检测事件（对应 event 表）。""" + + event_id: str + host_id: str + rule_id: str + metric: str + agg_value: float + threshold: float + operator: str + status: str # firing | resolved + severity: str + fired_at: float + labels: Dict[str, str] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return { + "event_id": self.event_id, + "host_id": self.host_id, + "rule_id": self.rule_id, + "metric": self.metric, + "agg_value": self.agg_value, + "threshold": self.threshold, + "operator": self.operator, + "status": self.status, + "severity": self.severity, + "fired_at": self.fired_at, + "labels": self.labels, + } + + +@dataclass +class Alert: + """收敛后告警（对应 alert 表）。""" + + alert_id: str + dedup_key: str + aggregate_key: str + severity: str + status: str # firing | resolved + title: str = "" + detail: Dict[str, Any] = field(default_factory=dict) + count: int = 1 + first_at: float = 0.0 + last_at: float = 0.0 + ack_status: str = "open" + ack_by: Optional[str] = None + ack_at: Optional[float] = None + + def to_dict(self) -> Dict[str, Any]: + return { + "alert_id": self.alert_id, + "dedup_key": self.dedup_key, + "aggregate_key": self.aggregate_key, + "severity": self.severity, + "status": self.status, + "title": self.title, + "detail": dict(self.detail), + "count": self.count, + "first_at": self.first_at, + "last_at": self.last_at, + "ack_status": self.ack_status, + "ack_by": self.ack_by, + "ack_at": self.ack_at, + } diff --git a/detector/notifier.py b/detector/notifier.py new file mode 100644 index 0000000..f00c094 --- /dev/null +++ b/detector/notifier.py @@ -0,0 +1,86 @@ +"""通知渠道封装。 + +对 email / webhook / dingtalk / wechat 等渠道做统一封装；真实发送通过 +urllib 完成（或按需扩展 SDK），默认 InMemorySender 记录发送历史，便于测试。 +""" + +from __future__ import annotations + +import json +import threading +import urllib.request +from typing import Any, Dict, List, Optional + +from .models import Alert + + +class ChannelSender: + """通知渠道发送器接口。""" + + def send(self, alert: Alert, target: Optional[str] = None) -> bool: + raise NotImplementedError + + +class InMemorySender(ChannelSender): + """记录发送历史，不真正外呼。""" + + def __init__(self, channel: str) -> None: + self.channel = channel + self.sent: List[Dict[str, Any]] = [] + self._lock = threading.Lock() + + def send(self, alert: Alert, target: Optional[str] = None) -> bool: + with self._lock: + self.sent.append({"channel": self.channel, "target": target, "alert_id": alert.alert_id}) + return True + + +class WebhookSender(ChannelSender): + """通用 Webhook（POST JSON）。""" + + def __init__(self, url: str) -> None: + self.url = url + + def send(self, alert: Alert, target: Optional[str] = None) -> bool: + payload = json.dumps(alert.to_dict()).encode("utf-8") + req = urllib.request.Request( + target or self.url, + data=payload, + headers={"Content-Type": "application/json"}, + method="POST", + ) + with urllib.request.urlopen(req, timeout=5) as resp: # noqa: S310 + return 200 <= resp.status < 300 + + +class Notifier: + """通知分发器：按告警渠道列表投递。""" + + def __init__(self, senders: Optional[Dict[str, ChannelSender]] = None) -> None: + self.senders = senders or {} + self.sent: List[Dict[str, Any]] = [] + self._lock = threading.Lock() + + def register(self, channel: str, sender: ChannelSender) -> None: + self.senders[channel] = sender + + def notify(self, alert: Alert, channels: Optional[List[str]] = None) -> List[str]: + """发送告警，返回成功渠道列表。""" + channels = channels or list(self.senders.keys()) + delivered: List[str] = [] + for channel in channels: + sender = self.senders.get(channel) + if sender is None: + continue + ok = sender.send(alert) + with self._lock: + self.sent.append({"channel": channel, "alert_id": alert.alert_id, "ok": ok}) + if ok: + delivered.append(channel) + return delivered + + @classmethod + def memory(cls, channels: Optional[List[str]] = None) -> "Notifier": + """构造内存版 Notifier，便于本地运行/测试。""" + channels = channels or ["email", "webhook", "dingtalk", "wechat"] + return cls({c: InMemorySender(c) for c in channels}) diff --git a/detector/rule_loader.py b/detector/rule_loader.py new file mode 100644 index 0000000..e945013 --- /dev/null +++ b/detector/rule_loader.py @@ -0,0 +1,101 @@ +"""规则加载与缓存。 + +- 从 RuleStore（MySQL）全量加载规则； +- 写入 Cache（Redis）作为热缓存（hms:rule:cache:{rule_id}、hms:rule:version）； +- 后台线程按 reload_interval 轮询版本号，版本变化则热加载； +- 提供 match()：按 metric / scope(host_id、host_group、service) 匹配启用的规则。 +""" + +from __future__ import annotations + +import json +import threading +import time +from typing import Dict, List, Optional + +from .models import MetricRule, MetricSample, parse_duration +from .storage import Cache, RuleStore + +RULE_CACHE_PREFIX = "hms:rule:cache:" +RULE_VERSION_KEY = "hms:rule:version" + + +class RuleLoader: + def __init__(self, store: RuleStore, cache: Cache, reload_interval: str = "10s") -> None: + self._store = store + self._cache = cache + self._reload_interval = parse_duration(reload_interval) or 10 + self._rules: Dict[str, MetricRule] = {} + self._version = 0 + self._lock = threading.RLock() + self._stop_event = threading.Event() + self._thread: Optional[threading.Thread] = None + + # ------------------------------------------------------------------ # + # 加载与热更新 + # ------------------------------------------------------------------ # + def load(self) -> int: + """全量加载规则，返回当前版本号。""" + rules = self._store.list_rules() + new_map = {r.rule_id: r for r in rules} + version = self._store.get_version() + with self._lock: + self._rules = new_map + self._version = version + for rule in rules: + self._cache.set(RULE_CACHE_PREFIX + rule.rule_id, json.dumps(rule.to_dict()), ttl=600) + self._cache.set(RULE_VERSION_KEY, str(version)) + return version + + def start(self) -> None: + """启动热加载后台线程。""" + if self._thread and self._thread.is_alive(): + return + self._thread = threading.Thread(target=self._run, name="rule-hot-reload", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + if self._thread: + self._thread.join(timeout=2) + + def _run(self) -> None: + while not self._stop_event.is_set(): + try: + current = self._store.get_version() + if current != self._version: + self.load() + except Exception: # pragma: no cover - 日志由上层处理 + pass + self._stop_event.wait(self._reload_interval) + + # ------------------------------------------------------------------ # + # 查询与匹配 + # ------------------------------------------------------------------ # + def get_rules(self) -> List[MetricRule]: + with self._lock: + return list(self._rules.values()) + + def get_rule(self, rule_id: str) -> Optional[MetricRule]: + with self._lock: + return self._rules.get(rule_id) + + def match( + self, + sample: MetricSample, + host_id: str, + host_group: Optional[str] = None, + service: Optional[str] = None, + ) -> List[MetricRule]: + """返回匹配该样本的启用规则。""" + matched: List[MetricRule] = [] + service = service or sample.labels.get("service") + for rule in self.get_rules(): + if not rule.enabled: + continue + if rule.metric != sample.name: + continue + if not rule.scope.matches(host_id, host_group, service): + continue + matched.append(rule) + return matched diff --git a/detector/storage.py b/detector/storage.py new file mode 100644 index 0000000..f2fe162 --- /dev/null +++ b/detector/storage.py @@ -0,0 +1,637 @@ +"""存储抽象层。 + +- RuleStore / EventStore / AlertStore：MySQL 语义（准），默认 InMemory 实现； +- Cache：Redis 语义（get/set NX/INCR/EXPIRE），默认 InMemory 实现； +- MessageBus：Kafka 语义（publish），默认 InMemory 实现。 + +真实中间件适配器放在同文件末尾，采用惰性导入（可选依赖），未安装驱动时给出 +清晰错误；测试与本地运行均使用 InMemory 实现，保证零依赖可编译、可运行。 +""" + +from __future__ import annotations + +import itertools +import json +import threading +import time +import uuid +from abc import ABC, abstractmethod +from typing import Any, Dict, List, Optional, Tuple + +from .models import Alert, Event, MetricRule + + +# --------------------------------------------------------------------------- # +# 接口定义 +# --------------------------------------------------------------------------- # +class RuleStore(ABC): + """规则仓库（Source of Truth：MySQL metric_rule + rule_scope）。""" + + @abstractmethod + def list_rules(self) -> List[MetricRule]: + ... + + @abstractmethod + def get_rule(self, rule_id: str) -> Optional[MetricRule]: + ... + + @abstractmethod + def create_rule(self, rule: MetricRule) -> MetricRule: + ... + + @abstractmethod + def update_rule(self, rule: MetricRule) -> MetricRule: + ... + + @abstractmethod + def delete_rule(self, rule_id: str) -> bool: + ... + + @abstractmethod + def get_version(self) -> int: + ... + + +class EventStore(ABC): + """事件仓库（MySQL event 表）。""" + + @abstractmethod + def insert(self, event: Event) -> None: + ... + + @abstractmethod + def list_events( + self, + host_id: Optional[str] = None, + rule_id: Optional[str] = None, + status: Optional[str] = None, + severity: Optional[str] = None, + start: Optional[float] = None, + end: Optional[float] = None, + page: int = 1, + page_size: int = 20, + ) -> Tuple[int, List[Event]]: + ... + + +class AlertStore(ABC): + """告警仓库（MySQL alert 表）。""" + + @abstractmethod + def upsert_firing(self, alert: Alert) -> Alert: + ... + + @abstractmethod + def get(self, alert_id: str) -> Optional[Alert]: + ... + + @abstractmethod + def list_alerts( + self, + severity: Optional[str] = None, + status: Optional[str] = None, + ack_status: Optional[str] = None, + start: Optional[float] = None, + end: Optional[float] = None, + page: int = 1, + page_size: int = 20, + ) -> Tuple[int, List[Alert]]: + ... + + @abstractmethod + def ack(self, alert_id: str, ack_by: str = "system") -> Optional[Alert]: + ... + + @abstractmethod + def close(self, alert_id: str) -> Optional[Alert]: + ... + + @abstractmethod + def resolve_by_dedup_key(self, dedup_key: str, at: float) -> Optional[Alert]: + ... + + +class Cache(ABC): + """缓存抽象（Redis 语义子集）。""" + + @abstractmethod + def get(self, key: str) -> Optional[str]: + ... + + @abstractmethod + def set(self, key: str, value: str, ttl: Optional[int] = None) -> bool: + ... + + @abstractmethod + def set_nx(self, key: str, value: str, ttl: Optional[int] = None) -> bool: + ... + + @abstractmethod + def incr(self, key: str, ttl: Optional[int] = None) -> int: + ... + + @abstractmethod + def delete(self, key: str) -> bool: + ... + + +class MessageBus(ABC): + """消息总线抽象（Kafka 语义子集）。""" + + @abstractmethod + def publish(self, topic: str, key: str, value: Dict[str, Any]) -> None: + ... + + +# --------------------------------------------------------------------------- # +# InMemory 实现（线程安全） +# --------------------------------------------------------------------------- # +class InMemoryRuleStore(RuleStore): + def __init__(self) -> None: + self._rules: Dict[str, MetricRule] = {} + self._version = 0 + self._lock = threading.RLock() + + def list_rules(self) -> List[MetricRule]: + with self._lock: + return list(self._rules.values()) + + def get_rule(self, rule_id: str) -> Optional[MetricRule]: + with self._lock: + return self._rules.get(rule_id) + + def create_rule(self, rule: MetricRule) -> MetricRule: + with self._lock: + if rule.rule_id in self._rules: + raise ValueError(f"rule {rule.rule_id} already exists") + rule.version = rule.version + 1 + self._rules[rule.rule_id] = rule + self._version += 1 + return rule + + def update_rule(self, rule: MetricRule) -> MetricRule: + with self._lock: + if rule.rule_id not in self._rules: + raise ValueError(f"rule {rule.rule_id} not found") + rule.version = self._rules[rule.rule_id].version + 1 + self._rules[rule.rule_id] = rule + self._version += 1 + return rule + + def delete_rule(self, rule_id: str) -> bool: + with self._lock: + if rule_id not in self._rules: + return False + del self._rules[rule_id] + self._version += 1 + return True + + def get_version(self) -> int: + with self._lock: + return self._version + + +class InMemoryEventStore(EventStore): + def __init__(self) -> None: + self._events: List[Event] = [] + self._lock = threading.RLock() + + def insert(self, event: Event) -> None: + with self._lock: + self._events.append(event) + + def list_events( + self, + host_id: Optional[str] = None, + rule_id: Optional[str] = None, + status: Optional[str] = None, + severity: Optional[str] = None, + start: Optional[float] = None, + end: Optional[float] = None, + page: int = 1, + page_size: int = 20, + ) -> Tuple[int, List[Event]]: + with self._lock: + items = [e for e in self._events if self._match_event(e, host_id, rule_id, status, severity, start, end)] + items.sort(key=lambda e: e.fired_at, reverse=True) + total = len(items) + start_idx = (page - 1) * page_size + return total, items[start_idx : start_idx + page_size] + + @staticmethod + def _match_event( + e: Event, + host_id: Optional[str], + rule_id: Optional[str], + status: Optional[str], + severity: Optional[str], + start: Optional[float], + end: Optional[float], + ) -> bool: + if host_id and e.host_id != host_id: + return False + if rule_id and e.rule_id != rule_id: + return False + if status and e.status != status: + return False + if severity and e.severity != severity: + return False + if start is not None and e.fired_at < start: + return False + if end is not None and e.fired_at > end: + return False + return True + + +class InMemoryAlertStore(AlertStore): + def __init__(self) -> None: + self._alerts: Dict[str, Alert] = {} + self._lock = threading.RLock() + + def upsert_firing(self, alert: Alert) -> Alert: + with self._lock: + # 同 aggregate_key 且仍在 firing 的告警做聚合：计数 + 1，更新 last_at 与明细。 + for existing in self._alerts.values(): + if existing.aggregate_key == alert.aggregate_key and existing.status == "firing": + existing.count += 1 + existing.last_at = alert.last_at + existing.detail = self._merge_detail(existing.detail, alert.detail) + return existing + self._alerts[alert.alert_id] = alert + return alert + + @staticmethod + def _merge_detail(a: Dict[str, Any], b: Dict[str, Any]) -> Dict[str, Any]: + merged = dict(a) + hosts = list(dict.fromkeys(list(merged.get("hosts", [])) + list(b.get("hosts", [])))) + merged["hosts"] = hosts + for key in ("current", "threshold", "metric"): + if key in b: + merged[key] = b[key] + return merged + + def get(self, alert_id: str) -> Optional[Alert]: + with self._lock: + return self._alerts.get(alert_id) + + def list_alerts( + self, + severity: Optional[str] = None, + status: Optional[str] = None, + ack_status: Optional[str] = None, + start: Optional[float] = None, + end: Optional[float] = None, + page: int = 1, + page_size: int = 20, + ) -> Tuple[int, List[Alert]]: + with self._lock: + items = [ + a + for a in self._alerts.values() + if self._match_alert(a, severity, status, ack_status, start, end) + ] + items.sort(key=lambda a: a.last_at, reverse=True) + total = len(items) + start_idx = (page - 1) * page_size + return total, items[start_idx : start_idx + page_size] + + @staticmethod + def _match_alert( + a: Alert, + severity: Optional[str], + status: Optional[str], + ack_status: Optional[str], + start: Optional[float], + end: Optional[float], + ) -> bool: + if severity and a.severity != severity: + return False + if status and a.status != status: + return False + if ack_status and a.ack_status != ack_status: + return False + if start is not None and a.last_at < start: + return False + if end is not None and a.last_at > end: + return False + return True + + def ack(self, alert_id: str, ack_by: str = "system") -> Optional[Alert]: + with self._lock: + a = self._alerts.get(alert_id) + if a is None: + return None + a.ack_status = "acked" + a.ack_by = ack_by + a.ack_at = time.time() + return a + + def close(self, alert_id: str) -> Optional[Alert]: + with self._lock: + a = self._alerts.get(alert_id) + if a is None: + return None + a.ack_status = "closed" + return a + + def resolve_by_dedup_key(self, dedup_key: str, at: float) -> Optional[Alert]: + with self._lock: + for a in self._alerts.values(): + if a.dedup_key == dedup_key and a.status == "firing": + a.status = "resolved" + a.last_at = at + return a + return None + + +class InMemoryCache(Cache): + def __init__(self) -> None: + self._data: Dict[str, Tuple[str, Optional[float]]] = {} + self._lock = threading.RLock() + + def _now(self) -> float: + return time.monotonic() + + def _purge(self, key: str) -> None: + if key in self._data: + value, exp = self._data[key] + if exp is not None and exp <= self._now(): + del self._data[key] + + def get(self, key: str) -> Optional[str]: + with self._lock: + self._purge(key) + item = self._data.get(key) + return item[0] if item else None + + def set(self, key: str, value: str, ttl: Optional[int] = None) -> bool: + with self._lock: + exp = self._now() + ttl if ttl else None + self._data[key] = (value, exp) + return True + + def set_nx(self, key: str, value: str, ttl: Optional[int] = None) -> bool: + with self._lock: + self._purge(key) + if key in self._data: + return False + exp = self._now() + ttl if ttl else None + self._data[key] = (value, exp) + return True + + def incr(self, key: str, ttl: Optional[int] = None) -> int: + with self._lock: + self._purge(key) + item = self._data.get(key) + current = int(item[0]) + 1 if item else 1 + exp = self._now() + ttl if ttl else None + self._data[key] = (str(current), exp) + return current + + def delete(self, key: str) -> bool: + with self._lock: + if key in self._data: + del self._data[key] + return True + return False + + +class InMemoryMessageBus(MessageBus): + def __init__(self) -> None: + self.messages: List[Tuple[str, str, Dict[str, Any]]] = [] + self._lock = threading.RLock() + + def publish(self, topic: str, key: str, value: Dict[str, Any]) -> None: + with self._lock: + self.messages.append((topic, key, value)) + + +# --------------------------------------------------------------------------- # +# 工厂函数：根据配置选择驱动（未安装可选依赖时安全回退 memory） +# --------------------------------------------------------------------------- # +def _safe(loader, fallback): + try: + return loader() + except Exception: + return fallback() + + +def create_rule_store(config) -> RuleStore: + if config.rule_store_driver == "mysql": + return _safe(lambda: MySQLRuleStore(config.mysql_dsn), InMemoryRuleStore) + return InMemoryRuleStore() + + +def create_event_store(config) -> EventStore: + if config.event_store_driver == "mysql": + return _safe(lambda: MySQLEventStore(config.mysql_dsn), InMemoryEventStore) + return InMemoryEventStore() + + +def create_alert_store(config) -> AlertStore: + if config.alert_store_driver == "mysql": + return _safe(lambda: MySQLAlertStore(config.mysql_dsn), InMemoryAlertStore) + return InMemoryAlertStore() + + +def create_cache(config) -> Cache: + if config.cache_driver == "redis": + return _safe(lambda: RedisCache(config.redis_url), InMemoryCache) + return InMemoryCache() + + +def create_bus(config) -> MessageBus: + if config.bus_driver == "kafka": + return _safe(lambda: KafkaBus(config.kafka_bootstrap), InMemoryMessageBus) + return InMemoryMessageBus() + + +# --------------------------------------------------------------------------- # +# 可选中间件适配器（惰性导入；未安装驱动时抛出 RuntimeError） +# --------------------------------------------------------------------------- # +class MySQLRuleStore(RuleStore): + """基于 pymysql 的规则仓库。需安装：pip install pymysql""" + + def __init__(self, dsn: str) -> None: + try: + import pymysql # noqa: F401 + except ImportError as exc: # pragma: no cover + raise RuntimeError("MySQLRuleStore requires pymysql") from exc + self._dsn = dsn + self._conn = None + + def _cursor(self): + import pymysql + + if self._conn is None: + self._conn = pymysql.connect(**self._parse_dsn(self._dsn)) + return self._conn.cursor() + + @staticmethod + def _parse_dsn(dsn: str) -> Dict[str, Any]: + # 简化：mysql://user:pass@host:port/db + import urllib.parse + + parsed = urllib.parse.urlparse(dsn) + return { + "host": parsed.hostname or "127.0.0.1", + "port": parsed.port or 3306, + "user": parsed.username or "root", + "password": parsed.password or "", + "database": parsed.path.lstrip("/") or "hms", + } + + def list_rules(self) -> List[MetricRule]: + cur = self._cursor() + cur.execute( + "SELECT rule_id,name,metric,aggregation,operator,threshold,threshold2," + "for_duration,severity,labels,notify_channels,enabled,version " + "FROM metric_rule" + ) + rules = [] + for row in cur.fetchall(): + scopes = self._load_scopes(row[0]) + rules.append( + MetricRule( + rule_id=row[0], name=row[1], metric=row[2], aggregation=row[3], + operator=row[4], threshold=float(row[5]), + threshold2=float(row[6]) if row[6] is not None else None, + for_duration=row[7], severity=row[8], + labels=json.loads(row[9]) if row[9] else {}, + notify_channels=json.loads(row[10]) if row[10] else [], + enabled=bool(row[11]), version=int(row[12]), + scope=scopes[0] if scopes else RuleScope(), + ) + ) + return rules + + def _load_scopes(self, rule_id: str) -> List[RuleScope]: + cur = self._cursor() + cur.execute( + "SELECT scope_type,host_id,host_group,service FROM rule_scope WHERE rule_id=%s", + (rule_id,), + ) + scopes = [] + for st, host_id, host_group, service in cur.fetchall(): + scopes.append( + RuleScope( + scope_type=st, + host_ids=[host_id] if host_id else [], + host_group=host_group or "", + service=service or "", + ) + ) + return scopes + + def get_rule(self, rule_id: str) -> Optional[MetricRule]: + for r in self.list_rules(): + if r.rule_id == rule_id: + return r + return None + + def create_rule(self, rule: MetricRule) -> MetricRule: + raise NotImplementedError("MySQL write path 由上层 API 通过事务实现") + + def update_rule(self, rule: MetricRule) -> MetricRule: + raise NotImplementedError + + def delete_rule(self, rule_id: str) -> bool: + raise NotImplementedError + + def get_version(self) -> int: + cur = self._cursor() + cur.execute("SELECT COALESCE(MAX(version),0) FROM metric_rule") + return int(cur.fetchone()[0]) + + +class MySQLEventStore(EventStore): + def __init__(self, dsn: str) -> None: + try: + import pymysql # noqa: F401 + except ImportError as exc: # pragma: no cover + raise RuntimeError("MySQLEventStore requires pymysql") from exc + self._dsn = dsn + + def insert(self, event: Event) -> None: + raise NotImplementedError("接入真实 MySQL 时实现 event 表写入") + + def list_events(self, **kwargs) -> Tuple[int, List[Event]]: + raise NotImplementedError + + +class MySQLAlertStore(AlertStore): + def __init__(self, dsn: str) -> None: + try: + import pymysql # noqa: F401 + except ImportError as exc: # pragma: no cover + raise RuntimeError("MySQLAlertStore requires pymysql") from exc + self._dsn = dsn + + def upsert_firing(self, alert: Alert) -> Alert: + raise NotImplementedError("接入真实 MySQL 时实现 alert 表 upsert") + + def get(self, alert_id: str) -> Optional[Alert]: + raise NotImplementedError + + def list_alerts(self, **kwargs) -> Tuple[int, List[Alert]]: + raise NotImplementedError + + def ack(self, alert_id: str, ack_by: str = "system") -> Optional[Alert]: + raise NotImplementedError + + def close(self, alert_id: str) -> Optional[Alert]: + raise NotImplementedError + + def resolve_by_dedup_key(self, dedup_key: str, at: float) -> Optional[Alert]: + raise NotImplementedError + + +class RedisCache(Cache): + """基于 redis-py 的缓存。需安装：pip install redis""" + + def __init__(self, url: str) -> None: + try: + import redis # noqa: F401 + except ImportError as exc: # pragma: no cover + raise RuntimeError("RedisCache requires redis") from exc + import redis + + self._client = redis.Redis.from_url(url, decode_responses=True) + + def get(self, key: str) -> Optional[str]: + return self._client.get(key) + + def set(self, key: str, value: str, ttl: Optional[int] = None) -> bool: + return bool(self._client.set(key, value, ex=ttl)) + + def set_nx(self, key: str, value: str, ttl: Optional[int] = None) -> bool: + return bool(self._client.set(key, value, ex=ttl, nx=True)) + + def incr(self, key: str, ttl: Optional[int] = None) -> int: + value = self._client.incr(key) + if ttl and value == 1: + self._client.expire(key, ttl) + return int(value) + + def delete(self, key: str) -> bool: + return bool(self._client.delete(key)) + + +class KafkaBus(MessageBus): + """基于 kafka-python 的消息总线。需安装：pip install kafka-python""" + + def __init__(self, bootstrap: str) -> None: + try: + from kafka import KafkaProducer # noqa: F401 + except ImportError as exc: # pragma: no cover + raise RuntimeError("KafkaBus requires kafka-python") from exc + from kafka import KafkaProducer + + self._producer = KafkaProducer( + bootstrap_servers=bootstrap, + value_serializer=lambda v: json.dumps(v).encode("utf-8"), + key_serializer=lambda k: k.encode("utf-8"), + ) + + def publish(self, topic: str, key: str, value: Dict[str, Any]) -> None: + self._producer.send(topic, key=key, value=value) diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9121fa1 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,18 @@ +[build-system] +requires = ["setuptools>=61"] +build-backend = "setuptools.build_meta" + +[project] +name = "threshold-event-detector" +version = "0.1.0" +description = "HMS metric threshold event detector" +requires-python = ">=3.10" +license = { text = "Proprietary" } + +[project.optional-dependencies] +mysql = ["pymysql>=1.1"] +redis = ["redis>=4.0"] +kafka = ["kafka-python>=2.0"] + +[tool.setuptools.packages.find] +include = ["detector*"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..3dc8ca4 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +# 核心模块仅依赖 Python 标准库，以下为接入真实中间件时的可选依赖： +# pymysql>=1.1 # HMS_RULE_STORE_DRIVER=mysql / EVENT/ALERT 同理 +# redis>=4.0 # HMS_CACHE_DRIVER=redis +# kafka-python>=2.0 # HMS_BUS_DRIVER=kafka diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/__pycache__/test_api.cpython-310.pyc b/tests/__pycache__/test_api.cpython-310.pyc new file mode 100644 index 0000000..b6c02b5 Binary files /dev/null and b/tests/__pycache__/test_api.cpython-310.pyc differ diff --git a/tests/__pycache__/test_converger.cpython-310.pyc b/tests/__pycache__/test_converger.cpython-310.pyc new file mode 100644 index 0000000..98a6b15 Binary files /dev/null and b/tests/__pycache__/test_converger.cpython-310.pyc differ diff --git a/tests/__pycache__/test_engine.cpython-310.pyc b/tests/__pycache__/test_engine.cpython-310.pyc new file mode 100644 index 0000000..b90f5bb Binary files /dev/null and b/tests/__pycache__/test_engine.cpython-310.pyc differ diff --git a/tests/__pycache__/test_models.cpython-310.pyc b/tests/__pycache__/test_models.cpython-310.pyc new file mode 100644 index 0000000..1ecfda7 Binary files /dev/null and b/tests/__pycache__/test_models.cpython-310.pyc differ diff --git a/tests/__pycache__/test_notifier.cpython-310.pyc b/tests/__pycache__/test_notifier.cpython-310.pyc new file mode 100644 index 0000000..4d42212 Binary files /dev/null and b/tests/__pycache__/test_notifier.cpython-310.pyc differ diff --git a/tests/__pycache__/test_rule_loader.cpython-310.pyc b/tests/__pycache__/test_rule_loader.cpython-310.pyc new file mode 100644 index 0000000..38f2251 Binary files /dev/null and b/tests/__pycache__/test_rule_loader.cpython-310.pyc differ diff --git a/tests/test_api.py b/tests/test_api.py new file mode 100644 index 0000000..cc2da54 --- /dev/null +++ b/tests/test_api.py @@ -0,0 +1,120 @@ +import json +import threading +import unittest +import urllib.request +from urllib.error import HTTPError + +from detector.api import APIServer +from detector.app import DetectorApp +from detector.config import Config + + +class TestAPI(unittest.TestCase): + def setUp(self): + config = Config.from_env() + config.http_addr = "127.0.0.1:0" + self.app = DetectorApp(config) + self.app.seed_demo_rules() + self.app.start() + self.api = APIServer(self.app, "127.0.0.1:0") + self.port = self.api.server.server_address[1] + self.base = f"http://127.0.0.1:{self.port}" + self.thread = threading.Thread(target=self.api.serve_forever, daemon=True) + self.thread.start() + + def tearDown(self): + self.api.shutdown() + self.api.server.server_close() + self.app.stop() + + def _request(self, method, path, body=None): + data = json.dumps(body).encode("utf-8") if body is not None else None + req = urllib.request.Request( + self.base + path, + data=data, + method=method, + headers={"Content-Type": "application/json"}, + ) + try: + with urllib.request.urlopen(req, timeout=5) as resp: + return resp.status, json.loads(resp.read().decode("utf-8")) + except HTTPError as exc: + return exc.code, json.loads(exc.read().decode("utf-8")) + + def test_healthz(self): + status, body = self._request("GET", "/healthz") + self.assertEqual(status, 200) + self.assertEqual(body["data"]["status"], "ok") + + def test_list_rules(self): + status, body = self._request("GET", "/api/v1/rules") + self.assertEqual(status, 200) + self.assertGreaterEqual(body["data"]["total"], 3) + + def test_create_and_get_rule(self): + payload = { + "rule_id": "r-test", + "name": "load high", + "metric": "load_1m", + "aggregation": "avg", + "operator": "gt", + "threshold": 5.0, + "for_duration": "60s", + "severity": "warning", + "scope": {"scope_type": "all"}, + "notify_channels": ["email"], + "enabled": True, + } + status, body = self._request("POST", "/api/v1/rules", payload) + self.assertEqual(status, 200) + self.assertEqual(body["data"]["rule_id"], "r-test") + + status, body = self._request("GET", "/api/v1/rules/r-test") + self.assertEqual(status, 200) + self.assertEqual(body["data"]["threshold"], 5.0) + + def test_ingest_triggers_firing_event(self): + samples = [ + {"name": "cpu_usage", "value": 95.0, "timestamp": float(i * 15), "labels": {"service": "web"}} + for i in range(5) + ] + status, body = self._request( + "POST", "/api/v1/ingest", + {"host_id": "h-001", "host_group": "web", "samples": samples}, + ) + self.assertEqual(status, 200) + events = body["data"]["events"] + self.assertEqual(len(events), 1) + self.assertEqual(events[0]["status"], "firing") + + # 事件已落库 + status, body = self._request("GET", "/api/v1/events?host_id=h-001") + self.assertEqual(status, 200) + self.assertEqual(body["data"]["total"], 1) + + # 告警已生成 + status, body = self._request("GET", "/api/v1/alerts") + self.assertEqual(status, 200) + self.assertEqual(body["data"]["total"], 1) + alert_id = body["data"]["items"][0]["alert_id"] + + # 确认告警 + status, body = self._request("POST", f"/api/v1/alerts/{alert_id}/ack", {"ack_by": "tester"}) + self.assertEqual(status, 200) + self.assertEqual(body["data"]["ack_status"], "acked") + + def test_batch_ack(self): + self._request( + "POST", "/api/v1/ingest", + {"host_id": "h-002", "host_group": "web", + "samples": [{"name": "cpu_usage", "value": 95.0, "timestamp": float(i * 15)} for i in range(5)]}, + ) + status, body = self._request("GET", "/api/v1/alerts") + alert_ids = [a["alert_id"] for a in body["data"]["items"]] + status, body = self._request("POST", "/api/v1/alerts/batch-ack", {"alert_ids": alert_ids}) + self.assertEqual(status, 200) + self.assertEqual(body["data"]["acked"], len(alert_ids)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_converger.py b/tests/test_converger.py new file mode 100644 index 0000000..0f4b181 --- /dev/null +++ b/tests/test_converger.py @@ -0,0 +1,71 @@ +import unittest + +from detector.converger import AlertConverger +from detector.models import Event +from detector.storage import InMemoryAlertStore, InMemoryCache, InMemoryMessageBus + + +def make_event(status="firing", host_id="h-1", rule_id="r-1", fired_at=100.0): + return Event( + event_id=f"e-{status}-{host_id}", + host_id=host_id, + rule_id=rule_id, + metric="cpu_usage", + agg_value=95.0, + threshold=90.0, + operator="gt", + status=status, + severity="critical", + fired_at=fired_at, + ) + + +class TestAlertConverger(unittest.TestCase): + def setUp(self): + self.store = InMemoryAlertStore() + self.cache = InMemoryCache() + self.bus = InMemoryMessageBus() + self.converger = AlertConverger( + self.store, self.cache, self.bus, dedup_window="5m", aggregate_window="1m", + host_group_resolver=lambda h: "web", + ) + + def test_dedup_suppresses_duplicate(self): + first = self.converger.handle_event(make_event()) + self.assertIsNotNone(first) + second = self.converger.handle_event(make_event()) + self.assertIsNone(second) + self.assertEqual(len(self.store.list_alerts()[1]), 1) + + def test_resolve_clears_dedup(self): + self.converger.handle_event(make_event(fired_at=100.0)) + resolved = self.converger.handle_event(make_event(status="resolved", fired_at=200.0)) + self.assertIsNotNone(resolved) + self.assertEqual(resolved.status, "resolved") + + # 去重键已清除，恢复后再次 firing 可以产生新告警 + again = self.converger.handle_event(make_event(fired_at=300.0)) + self.assertIsNotNone(again) + + def test_aggregate_same_bucket(self): + a1 = self.converger.handle_event(make_event(host_id="h-1", fired_at=100.0)) + a2 = self.converger.handle_event(make_event(host_id="h-2", fired_at=110.0)) + self.assertIsNotNone(a1) + self.assertIsNotNone(a2) + alerts = self.store.list_alerts()[1] + # 同一 aggregate_key（同 bucket/severity/rule/group）聚合为一条 + self.assertEqual(len(alerts), 1) + self.assertEqual(alerts[0].count, 2) + self.assertIn("h-1", alerts[0].detail["hosts"]) + self.assertIn("h-2", alerts[0].detail["hosts"]) + + def test_publish_to_bus(self): + self.converger.handle_event(make_event()) + self.assertEqual(len(self.bus.messages), 1) + topic, key, value = self.bus.messages[0] + self.assertEqual(topic, "alerts.converged") + self.assertIn("alert_id", value) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..a2469e5 --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,107 @@ +import unittest + +from detector.engine import DetectionEngine, aggregate_value, compare +from detector.models import MetricRule, MetricSample, RuleScope +from detector.rule_loader import RuleLoader +from detector.storage import InMemoryCache, InMemoryRuleStore + + +def make_rule(rule_id="r-cpu", metric="cpu_usage", operator="gt", threshold=90.0, for_duration="60s"): + return MetricRule( + rule_id=rule_id, + name="cpu high", + metric=metric, + aggregation="avg", + operator=operator, + threshold=threshold, + for_duration=for_duration, + severity="critical", + scope=RuleScope(scope_type="all"), + ) + + +def make_engine(rule): + store = InMemoryRuleStore() + store.create_rule(rule) + loader = RuleLoader(store, InMemoryCache()) + loader.load() + return DetectionEngine(loader, evaluation_interval="15s", recover_duration="15s") + + +class TestAggregateCompare(unittest.TestCase): + def test_aggregate(self): + self.assertEqual(aggregate_value([1, 2, 3], "avg"), 2) + self.assertEqual(aggregate_value([1, 2, 3], "min"), 1) + self.assertEqual(aggregate_value([1, 2, 3], "max"), 3) + self.assertEqual(aggregate_value([1, 2, 3], "sum"), 6) + self.assertEqual(aggregate_value([1, 2, 3], "last"), 3) + + def test_compare(self): + self.assertTrue(compare(91, "gt", 90)) + self.assertFalse(compare(90, "gt", 90)) + self.assertTrue(compare(90, "gte", 90)) + self.assertTrue(compare(89, "lt", 90)) + self.assertTrue(compare(95, "between", 90, 100)) + self.assertFalse(compare(105, "between", 90, 100)) + + +class TestStateMachine(unittest.TestCase): + def test_firing_after_sustained(self): + rule = make_rule(for_duration="60s") + engine = make_engine(rule) + events = [] + for i in range(5): + ts = float(i * 15) + sample = MetricSample(name="cpu_usage", value=95.0, timestamp=ts) + events.extend(engine.handle(sample, "h-1")) + + self.assertEqual(len(events), 1) + self.assertEqual(events[0].status, "firing") + self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "firing") + + def test_no_fire_before_for_duration(self): + rule = make_rule(for_duration="60s") + engine = make_engine(rule) + events = [] + for i in range(3): + ts = float(i * 15) + events.extend(engine.handle(MetricSample(name="cpu_usage", value=95.0, timestamp=ts), "h-1")) + self.assertEqual(events, []) + self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "ok") + + def test_resolved(self): + rule = make_rule(for_duration="60s") + engine = make_engine(rule) + events = [] + for i in range(5): + events.extend(engine.handle(MetricSample(name="cpu_usage", value=95.0, timestamp=float(i * 15)), "h-1")) + self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "firing") + + # 一次回落进入 pending + events.clear() + events.extend(engine.handle(MetricSample(name="cpu_usage", value=50.0, timestamp=75.0), "h-1")) + self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "pending") + + # 观察期后仍回落 -> resolved + events.clear() + events.extend(engine.handle(MetricSample(name="cpu_usage", value=50.0, timestamp=90.0), "h-1")) + self.assertEqual(len(events), 1) + self.assertEqual(events[0].status, "resolved") + self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "ok") + + def test_pending_back_to_firing_no_duplicate_event(self): + rule = make_rule(for_duration="60s") + engine = make_engine(rule) + for i in range(5): + engine.handle(MetricSample(name="cpu_usage", value=95.0, timestamp=float(i * 15)), "h-1") + engine.handle(MetricSample(name="cpu_usage", value=50.0, timestamp=75.0), "h-1") + self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "pending") + + # 观察期内再次满足（低值样本滑出窗口后聚合值回升）-> 回到 firing，不重复触发 + events = engine.handle(MetricSample(name="cpu_usage", value=96.0, timestamp=150.0), "h-1") + self.assertEqual(events, []) + self.assertEqual(engine.state_of("h-1", rule.rule_id)["status"], "firing") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..1258e6c --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,61 @@ +import unittest + +from detector.models import MetricRule, RuleScope, parse_duration + + +class TestParseDuration(unittest.TestCase): + def test_units(self): + self.assertEqual(parse_duration("60s"), 60) + self.assertEqual(parse_duration("5m"), 300) + self.assertEqual(parse_duration("1h"), 3600) + self.assertEqual(parse_duration("1h30m"), 5400) + self.assertEqual(parse_duration("90"), 90) + self.assertEqual(parse_duration(""), 0) + self.assertEqual(parse_duration(None), 0) + + def test_invalid(self): + with self.assertRaises(ValueError): + parse_duration("abc") + + +class TestRuleScope(unittest.TestCase): + def test_all(self): + scope = RuleScope(scope_type="all") + self.assertTrue(scope.matches("h-1", "web", "svc")) + self.assertTrue(scope.matches("h-2", None, None)) + + def test_host_ids(self): + scope = RuleScope(scope_type="host_ids", host_ids=["h-1", "h-2"]) + self.assertTrue(scope.matches("h-1", None, None)) + self.assertFalse(scope.matches("h-3", None, None)) + + def test_host_group(self): + scope = RuleScope(scope_type="host_group", host_group="web") + self.assertTrue(scope.matches("h-1", "web", None)) + self.assertFalse(scope.matches("h-1", "db", None)) + self.assertFalse(scope.matches("h-1", None, None)) + + +class TestMetricRuleSerialization(unittest.TestCase): + def test_round_trip(self): + rule = MetricRule( + rule_id="r-1", + name="cpu", + metric="cpu_usage", + operator="gt", + threshold=90, + scope=RuleScope(scope_type="host_group", host_group="web"), + labels={"service": "web"}, + notify_channels=["email"], + ) + data = rule.to_dict() + restored = MetricRule.from_dict(data) + self.assertEqual(restored.rule_id, "r-1") + self.assertEqual(restored.threshold, 90) + self.assertEqual(restored.scope.host_group, "web") + self.assertEqual(restored.labels, {"service": "web"}) + self.assertEqual(restored.notify_channels, ["email"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_notifier.py b/tests/test_notifier.py new file mode 100644 index 0000000..d7924e2 --- /dev/null +++ b/tests/test_notifier.py @@ -0,0 +1,23 @@ +import unittest + +from detector.models import Alert +from detector.notifier import InMemorySender, Notifier + + +class TestNotifier(unittest.TestCase): + def test_memory_notifier(self): + notifier = Notifier.memory(["email", "webhook"]) + alert = Alert(alert_id="a-1", dedup_key="h-1:r-1", aggregate_key="k", severity="critical", status="firing") + delivered = notifier.notify(alert, ["email", "webhook", "dingtalk"]) + self.assertEqual(delivered, ["email", "webhook"]) + self.assertEqual(len(notifier.sent), 2) + + def test_in_memory_sender(self): + sender = InMemorySender("email") + alert = Alert(alert_id="a-2", dedup_key="d", aggregate_key="k", severity="warning", status="firing") + self.assertTrue(sender.send(alert, "ops@example.com")) + self.assertEqual(sender.sent[0]["target"], "ops@example.com") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_rule_loader.py b/tests/test_rule_loader.py new file mode 100644 index 0000000..b2d9caf --- /dev/null +++ b/tests/test_rule_loader.py @@ -0,0 +1,54 @@ +import unittest + +from detector.models import MetricRule, MetricSample, RuleScope +from detector.rule_loader import RuleLoader +from detector.storage import InMemoryCache, InMemoryRuleStore + + +class TestRuleLoader(unittest.TestCase): + def setUp(self): + self.store = InMemoryRuleStore() + self.cache = InMemoryCache() + self.loader = RuleLoader(self.store, self.cache, reload_interval="10s") + + def test_load_and_match(self): + rule = MetricRule( + rule_id="r-1", + name="cpu", + metric="cpu_usage", + operator="gt", + threshold=90, + scope=RuleScope(scope_type="host_group", host_group="web"), + ) + self.store.create_rule(rule) + self.loader.load() + + sample = MetricSample(name="cpu_usage", value=95, timestamp=0) + matched = self.loader.match(sample, "h-1", host_group="web") + self.assertEqual([r.rule_id for r in matched], ["r-1"]) + + # 分组不匹配 + self.assertEqual(self.loader.match(sample, "h-1", host_group="db"), []) + + def test_hot_reload_by_version(self): + self.loader.load() + self.assertEqual(self.loader.get_rules(), []) + + self.store.create_rule( + MetricRule(rule_id="r-2", name="mem", metric="mem_used_percent", threshold=85) + ) + # 版本变化后手动 load 模拟热加载线程 + self.assertNotEqual(self.store.get_version(), self.loader._version) + self.loader.load() + self.assertEqual(len(self.loader.get_rules()), 1) + + def test_disabled_rule_not_matched(self): + rule = MetricRule(rule_id="r-3", name="disk", metric="disk_used_percent", enabled=False) + self.store.create_rule(rule) + self.loader.load() + sample = MetricSample(name="disk_used_percent", value=95, timestamp=0) + self.assertEqual(self.loader.match(sample, "h-1"), []) + + +if __name__ == "__main__": + unittest.main()