"""Периодическое сканирование хостов с устаревшим agent.heartbeat (proactive host_silence).""" from __future__ import annotations import logging from dataclasses import dataclass from datetime import datetime, timedelta, timezone from sqlalchemy import func, select, text from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.config import get_settings from app.models import Host, Problem from app.services.problem_rules import RULE_HOST_SILENCE, RuleMatch, build_fingerprint, host_silence_match_for_host logger = logging.getLogger(__name__) ACTIVE_HOST_SILENCE_STATUSES = ("open", "acknowledged") _HOST_SILENCE_SCAN_LOCK_KEY = 915_001_001 @dataclass(frozen=True) class HostSilenceScanResult: host_id: int hostname: str problem: Problem created: bool def _host_silence_scan_lock(db: Session) -> bool: """Один scan за раз (uvicorn workers + systemd timer).""" bind = db.get_bind() if bind.dialect.name != "postgresql": return True acquired = db.execute( text("SELECT pg_try_advisory_xact_lock(:key)"), {"key": _HOST_SILENCE_SCAN_LOCK_KEY}, ).scalar() return bool(acquired) def _find_active_host_silence_problem( db: Session, *, host_id: int, hostname: str | None, ) -> Problem | None: active = db.scalar( select(Problem) .where( Problem.host_id == host_id, Problem.rule_id == RULE_HOST_SILENCE, Problem.status.in_(ACTIVE_HOST_SILENCE_STATUSES), ) .order_by(Problem.last_seen_at.desc()) .limit(1) ) if active is not None: return active name = (hostname or "").strip() if not name: return None sibling = db.scalar( select(Problem) .join(Host, Problem.host_id == Host.id) .where( Problem.rule_id == RULE_HOST_SILENCE, Problem.status.in_(ACTIVE_HOST_SILENCE_STATUSES), func.lower(Host.hostname) == name.lower(), ) .order_by(Problem.last_seen_at.desc()) .limit(1) ) if sibling is not None and sibling.host_id != host_id: sibling.host_id = host_id db.flush() return sibling def _apply_host_silence_refresh( problem: Problem, *, host_id: int, match: RuleMatch, fingerprint: str, ref: datetime, ) -> None: problem.host_id = host_id problem.summary = match.summary problem.title = match.title problem.fingerprint = fingerprint problem.last_seen_at = ref problem.updated_at = ref def open_or_refresh_host_silence_problem( db: Session, host_id: int, match: RuleMatch, *, now: datetime | None = None, hostname: str | None = None, ) -> tuple[Problem | None, bool]: """ Открыть или обновить open/ack Problem host_silence без ingest-события. Returns (problem, created). problem is None when suppressed by manual-resolve cooldown. """ ref = now or datetime.now(timezone.utc) if hostname is None: host = db.get(Host, host_id) hostname = host.hostname if host is not None else None fingerprint = build_fingerprint( host_id, match.correlation_type, match.rule_id, match.fingerprint_suffix, ) active_problem = _find_active_host_silence_problem(db, host_id=host_id, hostname=hostname) if active_problem is not None: _apply_host_silence_refresh( active_problem, host_id=host_id, match=match, fingerprint=fingerprint, ref=ref, ) db.flush() return active_problem, False settings = get_settings() cooldown_hours = settings.sac_host_silence_manual_resolve_cooldown_hours if cooldown_hours > 0: cooldown = timedelta(hours=cooldown_hours) manual_recent = db.scalar( select(Problem) .where( Problem.host_id == host_id, Problem.rule_id == match.rule_id, Problem.fingerprint == fingerprint, Problem.status == "resolved", Problem.resolved_by == "manual", Problem.updated_at >= ref - cooldown, ) .order_by(Problem.updated_at.desc()) .limit(1) ) if manual_recent is not None: logger.debug( "host_silence suppressed (manual cooldown %sh) host_id=%s problem_id=%s", cooldown_hours, host_id, manual_recent.id, ) return None, False manual_expired = db.scalar( select(Problem) .where( Problem.host_id == host_id, Problem.rule_id == match.rule_id, Problem.fingerprint == fingerprint, Problem.status == "resolved", Problem.resolved_by == "manual", Problem.updated_at < ref - cooldown, ) .order_by(Problem.updated_at.desc()) .limit(1) ) if manual_expired is not None: active_problem = _find_active_host_silence_problem(db, host_id=host_id, hostname=hostname) if active_problem is not None: _apply_host_silence_refresh( active_problem, host_id=host_id, match=match, fingerprint=fingerprint, ref=ref, ) db.flush() return active_problem, False manual_expired.status = "open" manual_expired.resolved_by = None _apply_host_silence_refresh( manual_expired, host_id=host_id, match=match, fingerprint=fingerprint, ref=ref, ) db.flush() return manual_expired, True problem = Problem( host_id=host_id, title=match.title, summary=match.summary, severity=match.severity, status="open", rule_id=match.rule_id, fingerprint=fingerprint, event_count=0, last_seen_at=ref, ) try: with db.begin_nested(): db.add(problem) db.flush() except IntegrityError: active_problem = _find_active_host_silence_problem(db, host_id=host_id, hostname=hostname) if active_problem is None: raise _apply_host_silence_refresh( active_problem, host_id=host_id, match=match, fingerprint=fingerprint, ref=ref, ) db.flush() return active_problem, False return problem, True def run_host_silence_scan(db: Session, *, now: datetime | None = None) -> list[HostSilenceScanResult]: """Найти stale-хосты и создать/обновить Problems rule:host_silence.""" if not _host_silence_scan_lock(db): logger.debug("host_silence scan skipped (another runner holds advisory lock)") return [] ref = now or datetime.now(timezone.utc) hosts = db.scalars(select(Host).order_by(Host.id)).all() results: list[HostSilenceScanResult] = [] for host in hosts: match = host_silence_match_for_host(db, host.id, hostname=host.hostname, now=ref) if match is None: continue problem, created = open_or_refresh_host_silence_problem( db, host.id, match, now=ref, hostname=host.hostname, ) if problem is None: continue results.append( HostSilenceScanResult( host_id=host.id, hostname=host.hostname, problem=problem, created=created, ) ) if created: logger.info( "host_silence scan: new problem host=%s id=%s problem_id=%s", host.hostname, host.id, problem.id, ) return results def dispatch_host_silence_notifications(db: Session, results: list[HostSilenceScanResult]) -> int: """Telegram/webhook/email только для вновь созданных Problems.""" from app.services.notify_dispatch import notify_problem sent = 0 for item in results: if not item.created: continue try: notify_problem(item.problem, event=None, db=db) sent += 1 except Exception: logger.exception( "host_silence notify failed host=%s problem_id=%s", item.hostname, item.problem.id, ) return sent