105 lines
4.0 KiB
Python
105 lines
4.0 KiB
Python
"""告警收敛：去重 + 聚合。
|
||
|
||
- 去重：同一 (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,
|
||
notify_channels=list(event.notify_channels),
|
||
)
|
||
stored = self.alert_store.upsert_firing(alert)
|
||
# 聚合更新时使用已存在告警的渠道列表，避免覆盖丢失
|
||
channels = stored.notify_channels or event.notify_channels
|
||
self._publish(stored)
|
||
if self.notifier is not None and hasattr(self.notifier, "notify"):
|
||
self.notifier.notify(stored, channels)
|
||
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 list(event.notify_channels)
|