"""检测引擎：滑动窗口聚合 + 状态机（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), notify_channels=list(rule.notify_channels), ) # ------------------------------------------------------------------ # # 状态查询（测试/可观测） # ------------------------------------------------------------------ # 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()