PTah
101 lines
3.3 KiB
Python
101 lines
3.3 KiB
Python
"""Cooldown / dedup for outbound notifications (F-NOT-03)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.config import get_settings
|
|
from app.models import Event, Problem
|
|
from app.models.notification_cooldown import NotificationCooldown
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
COOLDOWN_KIND_EVENT = "event"
|
|
COOLDOWN_KIND_PROBLEM = "problem"
|
|
|
|
# Не душим служебные проверки ingest
|
|
_EVENT_COOLDOWN_EXEMPT_TYPES = frozenset({"agent.test", "agent.lifecycle"})
|
|
|
|
|
|
def _as_utc(dt: datetime) -> datetime:
|
|
if dt.tzinfo is None:
|
|
return dt.replace(tzinfo=timezone.utc)
|
|
return dt.astimezone(timezone.utc)
|
|
|
|
|
|
def build_event_cooldown_key(event: Event) -> str:
|
|
if event.dedup_key and str(event.dedup_key).strip():
|
|
return f"event:{event.dedup_key.strip()}"
|
|
|
|
details: dict[str, Any] = event.details if isinstance(event.details, dict) else {}
|
|
host = event.host.hostname if event.host else "unknown"
|
|
ip = str(details.get("source_ip") or details.get("ip_address") or details.get("ip") or "").strip()
|
|
user = str(details.get("user") or details.get("username") or "").strip()
|
|
return f"event:{event.type}|{host}|{ip}|{user}"
|
|
|
|
|
|
def build_problem_cooldown_key(problem: Problem) -> str:
|
|
fp = (problem.fingerprint or "").strip() or f"id:{problem.id}"
|
|
return f"problem:{fp}"
|
|
|
|
|
|
def _cooldown_seconds_for_kind(kind: str) -> int:
|
|
settings = get_settings()
|
|
if not settings.sac_notify_cooldown_enabled:
|
|
return 0
|
|
if kind == COOLDOWN_KIND_PROBLEM:
|
|
return max(0, int(settings.sac_notify_problem_cooldown_sec))
|
|
return max(0, int(settings.sac_notify_event_cooldown_sec))
|
|
|
|
|
|
def allow_notify(db: Session | None, *, key: str, kind: str) -> bool:
|
|
"""True = можно слать сейчас; при True обновляет last_notified_at."""
|
|
if kind == COOLDOWN_KIND_EVENT:
|
|
cooldown_sec = _cooldown_seconds_for_kind(COOLDOWN_KIND_EVENT)
|
|
else:
|
|
cooldown_sec = _cooldown_seconds_for_kind(COOLDOWN_KIND_PROBLEM)
|
|
|
|
if cooldown_sec <= 0:
|
|
return True
|
|
if db is None:
|
|
return True
|
|
|
|
now = datetime.now(timezone.utc)
|
|
row = db.get(NotificationCooldown, key)
|
|
if row is not None:
|
|
last = _as_utc(row.last_notified_at)
|
|
elapsed = (now - last).total_seconds()
|
|
if elapsed >= 0 and elapsed < cooldown_sec:
|
|
logger.info(
|
|
"notify cooldown skip kind=%s key=%s elapsed=%.0fs need=%ss",
|
|
kind,
|
|
key[:120],
|
|
elapsed,
|
|
cooldown_sec,
|
|
)
|
|
return False
|
|
|
|
if row is None:
|
|
db.add(NotificationCooldown(cooldown_key=key, kind=kind, last_notified_at=now))
|
|
else:
|
|
row.kind = kind
|
|
row.last_notified_at = now
|
|
db.flush()
|
|
return True
|
|
|
|
|
|
def should_notify_event(event: Event, db: Session | None) -> bool:
|
|
if event.type in _EVENT_COOLDOWN_EXEMPT_TYPES:
|
|
return True
|
|
key = build_event_cooldown_key(event)
|
|
return allow_notify(db, key=key, kind=COOLDOWN_KIND_EVENT)
|
|
|
|
|
|
def should_notify_problem(problem: Problem, db: Session | None) -> bool:
|
|
key = build_problem_cooldown_key(problem)
|
|
return allow_notify(db, key=key, kind=COOLDOWN_KIND_PROBLEM)
|