"""领域模型：指标样本、阈值规则、作用范围、事件与告警。 字段命名与 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.get("rule_id") or ""), 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) notify_channels: List[str] = field(default_factory=list) 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, "notify_channels": list(self.notify_channels), } @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 notify_channels: List[str] = field(default_factory=list) 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, "notify_channels": list(self.notify_channels), }