6c43b8637e
Treat acknowledged as active, dedupe by hostname, add PostgreSQL advisory lock and a unique index. Migration closes existing duplicate open problems. Co-authored-by: Cursor <cursoragent@cursor.com>
291 lines
8.7 KiB
Python
291 lines
8.7 KiB
Python
"""Периодическое сканирование хостов с устаревшим 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
|