From 8ff9e309932ce4853d759a4f49381074b06e9733 Mon Sep 17 00:00:00 2001 From: Alexander Zubarev Date: Thu, 6 Aug 2026 18:33:44 +0300 Subject: [PATCH] =?UTF-8?q?=D0=97=D0=B0=D0=B3=D1=80=D1=83=D0=B7=D0=B8?= =?UTF-8?q?=D1=82=D1=8C=20=D1=84=D0=B0=D0=B9=D0=BB=D1=8B=20=D0=B2=20=C2=AB?= =?UTF-8?q?alert-processor/app=C2=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- alert-processor/app/__init__.py | 1 + alert-processor/app/audit_logger.py | 330 +++++++++++++++++ alert-processor/app/config.py | 257 ++++++++++++++ alert-processor/app/correlation.py | 530 ++++++++++++++++++++++++++++ alert-processor/app/fingerprint.py | 22 ++ 5 files changed, 1140 insertions(+) create mode 100644 alert-processor/app/__init__.py create mode 100644 alert-processor/app/audit_logger.py create mode 100644 alert-processor/app/config.py create mode 100644 alert-processor/app/correlation.py create mode 100644 alert-processor/app/fingerprint.py diff --git a/alert-processor/app/__init__.py b/alert-processor/app/__init__.py new file mode 100644 index 0000000..bc63beb --- /dev/null +++ b/alert-processor/app/__init__.py @@ -0,0 +1 @@ +# empty \ No newline at end of file diff --git a/alert-processor/app/audit_logger.py b/alert-processor/app/audit_logger.py new file mode 100644 index 0000000..be0956e --- /dev/null +++ b/alert-processor/app/audit_logger.py @@ -0,0 +1,330 @@ +from __future__ import annotations + +import json +from datetime import datetime, timezone +from typing import Any + +from redis.asyncio import Redis + +from app.models import NotificationDecision, ProcessorForwardEnvelope + + +def _utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +class AuditLogger: + def __init__( + self, + client: Redis, + key_prefix: str, + ttl_seconds: int = 604800, + max_stage_records: int = 200, + ) -> None: + self.client = client + self.key_prefix = key_prefix + self.ttl_seconds = ttl_seconds + self.max_stage_records = max_stage_records + + def _journal_key(self, correlation_id: str) -> str: + return f"{self.key_prefix}:journal:{correlation_id}" + + def _stages_key(self, correlation_id: str) -> str: + return f"{self.key_prefix}:stages:{correlation_id}" + + def _event_index_key(self, event_id: str) -> str: + return f"{self.key_prefix}:event:{event_id}" + + def _recent_key(self) -> str: + return f"{self.key_prefix}:recent" + + async def log_ingest_queued( + self, + envelope: ProcessorForwardEnvelope, + job_id: str, + queue_name: str, + ) -> None: + event = envelope.event + correlation_id = event.correlation_id + event_id = event.event_id or "" + + await self._upsert_journal( + correlation_id=correlation_id, + event_id=event_id or None, + fields={ + "state": "queued", + "job_id": job_id, + "queue_name": queue_name, + "correlation_id": correlation_id, + "event_id": event_id, + "host": event.host or "", + "service": event.service or "", + "trigger_name": event.trigger_name or "", + "severity": event.severity or "", + "queued_at": _utc_now(), + "updated_at": _utc_now(), + }, + ) + + await self.log_stage( + correlation_id=correlation_id, + event_id=event.event_id, + stage="ingest_queued", + status="ok", + details={ + "job_id": job_id, + "queue_name": queue_name, + }, + ) + + async def log_worker_started( + self, + envelope: ProcessorForwardEnvelope, + job_id: str, + attempt: int, + identity: str, + ) -> None: + event = envelope.event + await self._upsert_journal( + correlation_id=event.correlation_id, + event_id=event.event_id, + fields={ + "state": "processing", + "worker_started_at": _utc_now(), + "worker_attempt": attempt, + "processing_identity": identity, + "updated_at": _utc_now(), + }, + ) + + await self.log_stage( + correlation_id=event.correlation_id, + event_id=event.event_id, + stage="worker_started", + status="ok", + details={ + "job_id": job_id, + "attempt": attempt, + "identity": identity, + }, + ) + + async def log_stage( + self, + correlation_id: str, + event_id: str | None, + stage: str, + status: str = "ok", + details: dict[str, Any] | None = None, + ) -> None: + payload = { + "ts": _utc_now(), + "stage": stage, + "status": status, + "details": details or {}, + } + raw = json.dumps(payload, ensure_ascii=False) + + stages_key = self._stages_key(correlation_id) + await self.client.rpush(stages_key, raw) + await self.client.ltrim(stages_key, -self.max_stage_records, -1) + await self.client.expire(stages_key, self.ttl_seconds) + + await self._upsert_journal( + correlation_id=correlation_id, + event_id=event_id, + fields={ + "last_stage": stage, + "last_stage_status": status, + "updated_at": _utc_now(), + }, + ) + + async def log_decision( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + ) -> None: + event = envelope.event + await self._upsert_journal( + correlation_id=event.correlation_id, + event_id=event.event_id, + fields={ + "state": "decision_made", + "decision_at": _utc_now(), + "notify": str(decision.notify).lower(), + "suppressed": str(decision.suppressed).lower(), + "routing_class": decision.routing_class or "", + "decision_reason": decision.reason or "", + "decision_json": json.dumps( + decision.model_dump(mode="json"), + ensure_ascii=False, + ), + "updated_at": _utc_now(), + }, + ) + + await self.log_stage( + correlation_id=event.correlation_id, + event_id=event.event_id, + stage="decision_made", + status="ok", + details={ + "notify": decision.notify, + "suppressed": decision.suppressed, + "routing_class": decision.routing_class, + "reason": decision.reason, + }, + ) + + async def log_delivery( + self, + envelope: ProcessorForwardEnvelope, + delivery_payload: dict[str, Any], + ) -> None: + event = envelope.event + + await self._upsert_journal( + correlation_id=event.correlation_id, + event_id=event.event_id, + fields={ + "state": "delivered", + "delivery_at": _utc_now(), + "delivery_json": json.dumps(delivery_payload, ensure_ascii=False), + "updated_at": _utc_now(), + }, + ) + + await self.log_stage( + correlation_id=event.correlation_id, + event_id=event.event_id, + stage="delivery_completed", + status="ok", + details=delivery_payload, + ) + + async def log_worker_outcome( + self, + envelope: ProcessorForwardEnvelope, + state: str, + error: str | None = None, + attempt: int | None = None, + ) -> None: + event = envelope.event + + fields = { + "state": state, + "updated_at": _utc_now(), + } + if error: + fields["last_error"] = error + fields["last_error_at"] = _utc_now() + if attempt is not None: + fields["worker_attempt"] = attempt + + await self._upsert_journal( + correlation_id=event.correlation_id, + event_id=event.event_id, + fields=fields, + ) + + await self.log_stage( + correlation_id=event.correlation_id, + event_id=event.event_id, + stage=f"worker_{state}", + status="error" if error else "ok", + details={ + "attempt": attempt, + "error": error, + }, + ) + + async def get_event_audit( + self, + correlation_id: str | None = None, + event_id: str | None = None, + ) -> dict[str, Any] | None: + resolved_correlation = correlation_id + + if not resolved_correlation and event_id: + value = await self.client.get(self._event_index_key(event_id)) + if value is None: + return None + resolved_correlation = value.decode() if isinstance(value, bytes) else str(value) + + if not resolved_correlation: + return None + + journal_raw = await self.client.hgetall(self._journal_key(resolved_correlation)) + if not journal_raw: + return None + + stages_raw = await self.client.lrange(self._stages_key(resolved_correlation), 0, -1) + + journal = self._decode_hash(journal_raw) + stages = [json.loads(item.decode() if isinstance(item, bytes) else str(item)) for item in stages_raw] + + if journal.get("decision_json"): + try: + journal["decision"] = json.loads(journal["decision_json"]) + except Exception: + journal["decision"] = None + + if journal.get("delivery_json"): + try: + journal["delivery"] = json.loads(journal["delivery_json"]) + except Exception: + journal["delivery"] = None + + return { + "journal": journal, + "stages": stages, + } + + async def list_recent(self, limit: int = 20) -> list[dict[str, Any]]: + correlation_ids_raw = await self.client.zrevrange(self._recent_key(), 0, max(limit - 1, 0)) + results: list[dict[str, Any]] = [] + + for raw in correlation_ids_raw: + correlation_id = raw.decode() if isinstance(raw, bytes) else str(raw) + audit = await self.get_event_audit(correlation_id=correlation_id) + if audit is not None: + results.append(audit) + + return results + + async def _upsert_journal( + self, + correlation_id: str, + event_id: str | None, + fields: dict[str, Any], + ) -> None: + journal_key = self._journal_key(correlation_id) + + mapping: dict[str, str] = {} + for key, value in fields.items(): + if value is None: + continue + mapping[key] = str(value) + + await self.client.hset(journal_key, mapping=mapping) + await self.client.expire(journal_key, self.ttl_seconds) + + if event_id: + await self.client.set( + self._event_index_key(event_id), + correlation_id, + ex=self.ttl_seconds, + ) + + await self.client.zadd(self._recent_key(), {correlation_id: datetime.now(timezone.utc).timestamp()}) + await self.client.expire(self._recent_key(), self.ttl_seconds) + + @staticmethod + def _decode_hash(raw: dict[bytes, bytes]) -> dict[str, str]: + result: dict[str, str] = {} + for k, v in raw.items(): + key = k.decode() if isinstance(k, bytes) else str(k) + value = v.decode() if isinstance(v, bytes) else str(v) + result[key] = value + return result \ No newline at end of file diff --git a/alert-processor/app/config.py b/alert-processor/app/config.py new file mode 100644 index 0000000..19be907 --- /dev/null +++ b/alert-processor/app/config.py @@ -0,0 +1,257 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass + +from dotenv import load_dotenv + + +load_dotenv() + + +def _parse_bool(value: str | None, default: bool = False) -> bool: + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class Settings: + app_name: str = os.getenv("APP_NAME", "alert-processor") + app_host: str = os.getenv("APP_HOST", "0.0.0.0") + app_port: int = int(os.getenv("APP_PORT", "8081")) + + internal_api_token: str = os.getenv("INTERNAL_API_TOKEN", "").strip() + require_internal_api_token: bool = _parse_bool( + os.getenv("REQUIRE_INTERNAL_API_TOKEN", "true"), + default=True, + ) + + redis_enabled: bool = _parse_bool( + os.getenv("REDIS_ENABLED", "true"), + default=True, + ) + redis_url: str = os.getenv("REDIS_URL", "redis://127.0.0.1:6379/0").strip() + redis_key_prefix: str = os.getenv("REDIS_KEY_PREFIX", "alert").strip() + redis_fingerprint_ttl_seconds: int = int( + os.getenv("REDIS_FINGERPRINT_TTL_SECONDS", "86400") + ) + redis_event_ttl_seconds: int = int( + os.getenv("REDIS_EVENT_TTL_SECONDS", "604800") + ) + + suppress_enabled: bool = _parse_bool( + os.getenv("SUPPRESS_ENABLED", "true"), + default=True, + ) + suppress_window_seconds: int = int( + os.getenv("SUPPRESS_WINDOW_SECONDS", "900") + ) + suppress_apply_to_average: bool = _parse_bool( + os.getenv("SUPPRESS_APPLY_TO_AVERAGE", "true"), + default=True, + ) + + flap_enabled: bool = _parse_bool( + os.getenv("FLAP_ENABLED", "true"), + default=True, + ) + flap_window_seconds: int = int( + os.getenv("FLAP_WINDOW_SECONDS", "120") + ) + flap_threshold: int = int( + os.getenv("FLAP_THRESHOLD", "4") + ) + flap_apply_to_average: bool = _parse_bool( + os.getenv("FLAP_APPLY_TO_AVERAGE", "true"), + default=True, + ) + + correlation_enabled: bool = _parse_bool( + os.getenv("CORRELATION_ENABLED", "true"), + default=True, + ) + correlation_window_seconds: int = int( + os.getenv("CORRELATION_WINDOW_SECONDS", "180") + ) + correlation_suppress_children: bool = _parse_bool( + os.getenv("CORRELATION_SUPPRESS_CHILDREN", "true"), + default=True, + ) + + matrix_enabled: bool = _parse_bool( + os.getenv("MATRIX_ENABLED", "false"), + default=False, + ) + matrix_homeserver_url: str = os.getenv("MATRIX_HOMESERVER_URL", "").strip().rstrip("/") + matrix_room_id: str = os.getenv("MATRIX_ROOM_ID", "").strip() + + matrix_access_token: str = os.getenv("MATRIX_ACCESS_TOKEN", "").strip() + matrix_refresh_token: str = os.getenv("MATRIX_REFRESH_TOKEN", "").strip() + + matrix_oauth_token_endpoint: str = os.getenv("MATRIX_OAUTH_TOKEN_ENDPOINT", "").strip() + matrix_oauth_client_id: str = os.getenv("MATRIX_OAUTH_CLIENT_ID", "").strip() + matrix_oauth_client_secret: str = os.getenv("MATRIX_OAUTH_CLIENT_SECRET", "").strip() + + matrix_access_token_expires_in_seconds: int = int( + os.getenv("MATRIX_ACCESS_TOKEN_EXPIRES_IN_SECONDS", "300") + ) + matrix_refresh_margin_seconds: int = int( + os.getenv("MATRIX_REFRESH_MARGIN_SECONDS", "60") + ) + matrix_token_state_file: str = os.getenv( + "MATRIX_TOKEN_STATE_FILE", + ".matrix_token_state.json", + ).strip() + + matrix_message_type: str = os.getenv("MATRIX_MESSAGE_TYPE", "m.notice").strip() + matrix_request_timeout_seconds: float = float( + os.getenv("MATRIX_REQUEST_TIMEOUT_SECONDS", "10") + ) + matrix_verify_tls: bool = _parse_bool( + os.getenv("MATRIX_VERIFY_TLS", "true"), + default=True, + ) + + mail_enabled: bool = _parse_bool( + os.getenv("MAIL_ENABLED", "false"), + default=False, + ) + mail_smtp_host: str = os.getenv("MAIL_SMTP_HOST", "").strip() + mail_smtp_port: int = int(os.getenv("MAIL_SMTP_PORT", "587")) + mail_smtp_username: str = os.getenv("MAIL_SMTP_USERNAME", "").strip() + mail_smtp_password: str = os.getenv("MAIL_SMTP_PASSWORD", "").strip() + mail_from: str = os.getenv("MAIL_FROM", "").strip() + mail_to: str = os.getenv("MAIL_TO", "").strip() + mail_use_starttls: bool = _parse_bool( + os.getenv("MAIL_USE_STARTTLS", "true"), + default=True, + ) + mail_use_tls: bool = _parse_bool( + os.getenv("MAIL_USE_TLS", "false"), + default=False, + ) + mail_timeout_seconds: float = float( + os.getenv("MAIL_TIMEOUT_SECONDS", "15") + ) + mail_subject_prefix: str = os.getenv( + "MAIL_SUBJECT_PREFIX", + "[LLM-Zabbix]", + ).strip() + + zabbix_api_enabled: bool = _parse_bool( + os.getenv("ZABBIX_API_ENABLED", "false"), + default=False, + ) + zabbix_api_url: str = os.getenv("ZABBIX_API_URL", "").strip() + zabbix_api_token: str = os.getenv("ZABBIX_API_TOKEN", "").strip() + zabbix_web_url: str = os.getenv("ZABBIX_WEB_URL", "").strip().rstrip("/") + zabbix_api_timeout_seconds: float = float( + os.getenv("ZABBIX_API_TIMEOUT_SECONDS", "10") + ) + zabbix_api_verify_tls: bool = _parse_bool( + os.getenv("ZABBIX_API_VERIFY_TLS", "true"), + default=True, + ) + zabbix_graph_period_hours: int = int( + os.getenv("ZABBIX_GRAPH_PERIOD_HOURS", "1") + ) + zabbix_graph_timezone: str = os.getenv( + "ZABBIX_GRAPH_TIMEZONE", + "UTC", + ).strip() + zabbix_enrich_only_notify: bool = _parse_bool( + os.getenv("ZABBIX_ENRICH_ONLY_NOTIFY", "true"), + default=True, + ) + + llm_enabled: bool = _parse_bool( + os.getenv("LLM_ENABLED", "false"), + default=False, + ) + llm_base_url: str = os.getenv("LLM_BASE_URL", "").strip().rstrip("/") + llm_model: str = os.getenv("LLM_MODEL", "").strip() + llm_timeout_seconds: float = float( + os.getenv("LLM_TIMEOUT_SECONDS", "45") + ) + llm_verify_tls: bool = _parse_bool( + os.getenv("LLM_VERIFY_TLS", "true"), + default=True, + ) + llm_temperature: float = float( + os.getenv("LLM_TEMPERATURE", "0.1") + ) + llm_max_steps: int = int( + os.getenv("LLM_MAX_STEPS", "4") + ) + llm_max_commands: int = int( + os.getenv("LLM_MAX_COMMANDS", "4") + ) + + llm_triage_enabled: bool = _parse_bool( + os.getenv("LLM_TRIAGE_ENABLED", "false"), + default=False, + ) + llm_triage_cache_ttl_seconds: int = int( + os.getenv("LLM_TRIAGE_CACHE_TTL_SECONDS", "3600") + ) + + queue_enabled: bool = _parse_bool( + os.getenv("QUEUE_ENABLED", "true"), + default=True, + ) + queue_name: str = os.getenv("QUEUE_NAME", "alert:queue:events").strip() + queue_processing_name: str = os.getenv( + "QUEUE_PROCESSING_NAME", + "alert:queue:processing", + ).strip() + queue_deadletter_name: str = os.getenv( + "QUEUE_DEADLETTER_NAME", + "alert:queue:deadletter", + ).strip() + queue_block_timeout_seconds: int = int( + os.getenv("QUEUE_BLOCK_TIMEOUT_SECONDS", "5") + ) + queue_max_attempts: int = int( + os.getenv("QUEUE_MAX_ATTEMPTS", "3") + ) + queue_dedup_ttl_seconds: int = int( + os.getenv("QUEUE_DEDUP_TTL_SECONDS", "86400") + ) + queue_requeue_processing_on_startup: bool = _parse_bool( + os.getenv("QUEUE_REQUEUE_PROCESSING_ON_STARTUP", "true"), + default=True, + ) + correlation_kind_rules_path: str = os.getenv( + "CORRELATION_KIND_RULES_PATH", + "./config/event_kind_rules.yaml", + ).strip() + correlation_root_cause_path: str = os.getenv( + "CORRELATION_ROOT_CAUSE_PATH", + "./config/root_cause_map.yaml", + ).strip() + audit_enabled: bool = _parse_bool( + os.getenv("AUDIT_ENABLED", "true"), + default=True, + ) + audit_key_prefix: str = os.getenv( + "AUDIT_KEY_PREFIX", + "alert:audit", + ).strip() + audit_ttl_seconds: int = int( + os.getenv("AUDIT_TTL_SECONDS", "604800") + ) + audit_max_stage_records: int = int( + os.getenv("AUDIT_MAX_STAGE_RECORDS", "200") + ) + llm_correlation_enabled: bool = _parse_bool( + os.getenv("LLM_CORRELATION_ENABLED", "false"), + default=False, + ) + llm_correlation_min_confidence: str = os.getenv( + "LLM_CORRELATION_MIN_CONFIDENCE", + "medium", + ).strip().lower() + + +settings = Settings() \ No newline at end of file diff --git a/alert-processor/app/correlation.py b/alert-processor/app/correlation.py new file mode 100644 index 0000000..ab6d324 --- /dev/null +++ b/alert-processor/app/correlation.py @@ -0,0 +1,530 @@ +from __future__ import annotations + +import logging +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +from app.config import settings +from app.models import NotificationDecision, ProcessorForwardEnvelope + +logger = logging.getLogger(__name__) + + +def _norm(value: str | None) -> str: + return (value or "").strip().lower() + + +def severity_rank(severity: str | None) -> int: + value = _norm(severity) + if value in {"disaster"}: + return 4 + if value in {"high"}: + return 3 + if value in {"average"}: + return 2 + if value in {"warning", "information", "not classified", "not_classified"}: + return 1 + return 0 + + +@dataclass(frozen=True) +class EventKindRule: + kind: str + trigger_patterns: tuple[str, ...] = () + item_key_patterns: tuple[str, ...] = () + service_patterns: tuple[str, ...] = () + tag_patterns: tuple[str, ...] = () + scope_in: tuple[str, ...] = () + severity_in: tuple[str, ...] = () + + def matches( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + ) -> bool: + event = envelope.event + zbx_context = event.zabbix_context or {} + + trigger_name = _norm(event.trigger_name) + item_key = _norm(zbx_context.get("item_key")) + service = _norm(zbx_context.get("service") or event.service) + alert_scope = _norm(zbx_context.get("alert_scope")) + severity = _norm(decision.severity) + + tags_joined = " ".join( + f"{_norm(str(k))}={_norm(str(v))}" for k, v in (event.tags or {}).items() + ) + + if self.scope_in and alert_scope not in self.scope_in: + return False + + if self.severity_in and severity not in self.severity_in: + return False + + if self.trigger_patterns and not any(re.search(pattern, trigger_name) for pattern in self.trigger_patterns): + return False + + if self.item_key_patterns and not any(re.search(pattern, item_key) for pattern in self.item_key_patterns): + return False + + if self.service_patterns and not any(re.search(pattern, service) for pattern in self.service_patterns): + return False + + if self.tag_patterns and not all(re.search(pattern, tags_joined) for pattern in self.tag_patterns): + return False + + return True + + +@dataclass +class CorrelationEventRecord: + host: str + event_id: str | None + correlation_id: str + kind: str + severity: str | None + routing_class: str | None + fingerprint: str | None + root_candidate: bool + role: str + group_id: str | None + parent_event_id: str | None + parent_correlation_id: str | None + timestamp: float + tags: dict[str, str] = field(default_factory=dict) + service: str | None = None + scope: str | None = None + component: str | None = None + domain: str | None = None + + +@dataclass +class CorrelationAssessment: + applied: bool + role: str + kind: str + group_id: str | None + reason: str | None + parent_event_id: str | None = None + parent_correlation_id: str | None = None + root_cause_candidate: bool = False + correlated_event_count: int = 0 + suppress_child: bool = False + source: str = "deterministic" + confidence: str | None = "high" + + +@dataclass +class CorrelationRegistry: + event_kind_rules: list[EventKindRule] = field(default_factory=list) + root_cause_map: dict[str, set[str]] = field(default_factory=dict) + + def classify_event_kind( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + ) -> str: + for rule in self.event_kind_rules: + if rule.matches(envelope, decision): + return rule.kind + return "unknown" + + def is_root_candidate(self, kind: str) -> bool: + return kind in self.root_cause_map + + def explains(self, parent_kind: str, child_kind: str) -> bool: + if parent_kind == child_kind: + return False + return child_kind in self.root_cause_map.get(parent_kind, set()) + + @classmethod + def default(cls) -> "CorrelationRegistry": + return cls( + event_kind_rules=[ + EventKindRule( + kind="postgresql_unavailable", + trigger_patterns=(r"postgresql.+unavailable", r"postgres.+unavailable"), + ), + EventKindRule( + kind="host_down", + trigger_patterns=(r"host unavailable", r"icmp.+unreachable", r"unreachable"), + scope_in=("host_os",), + ), + EventKindRule( + kind="network_port_down", + trigger_patterns=(r"port down", r"interface down"), + ), + EventKindRule( + kind="ap_down", + trigger_patterns=(r"ap down",), + ), + EventKindRule( + kind="disk_full", + item_key_patterns=(r"^vfs\.fs\.",), + trigger_patterns=(r"space", r"full", r"disk"), + scope_in=("host_os",), + ), + EventKindRule( + kind="container_unavailable", + trigger_patterns=( + r"container", + r"docker", + r"health state", + r"not running", + r"stopped", + r"unavailable", + r"down", + r"failed", + ), + scope_in=("container",), + ), + EventKindRule( + kind="cpu_high", + item_key_patterns=(r"^system\.cpu\.",), + trigger_patterns=(r"cpu utilization",), + scope_in=("host_os",), + ), + EventKindRule( + kind="memory_high", + item_key_patterns=(r"^vm\.memory\.", r"^system\.swap\."), + trigger_patterns=(r"memory utilization",), + scope_in=("host_os",), + ), + EventKindRule( + kind="ssl_expiry", + trigger_patterns=(r"ssl", r"certificate", r"expire"), + ), + EventKindRule( + kind="service_unavailable", + trigger_patterns=(r"service unavailable", r"unavailable"), + ), + EventKindRule( + kind="site_unavailable", + trigger_patterns=(r"site.+is down", r"failed step of scenario"), + ), + EventKindRule( + kind="db_connection_error", + trigger_patterns=(r"connection", r"db"), + ), + ], + root_cause_map={ + "host_down": { + "service_unavailable", + "container_unavailable", + "postgresql_unavailable", + "cpu_high", + "memory_high", + "disk_full", + "ssl_expiry", + "db_connection_error", + "site_unavailable", + }, + "network_port_down": { + "host_down", + "service_unavailable", + "ap_down", + "site_unavailable", + }, + "postgresql_unavailable": { + "service_unavailable", + "db_connection_error", + "site_unavailable", + }, + "disk_full": { + "postgresql_unavailable", + "container_unavailable", + "service_unavailable", + "db_connection_error", + "site_unavailable", + }, + "container_unavailable": { + "service_unavailable", + "site_unavailable", + }, + }, + ) + + @classmethod + def from_yaml_files( + cls, + event_kind_rules_path: str, + root_cause_map_path: str, + ) -> "CorrelationRegistry": + base = cls.default() + + custom_kind_rules = _load_event_kind_rules(event_kind_rules_path) + custom_root_cause_map = _load_root_cause_map(root_cause_map_path) + + merged_kind_rules = custom_kind_rules + base.event_kind_rules + + merged_root_cause_map: dict[str, set[str]] = { + root_kind: set(children) + for root_kind, children in base.root_cause_map.items() + } + + for root_kind, explained in custom_root_cause_map.items(): + if root_kind not in merged_root_cause_map: + merged_root_cause_map[root_kind] = set() + merged_root_cause_map[root_kind].update(explained) + + logger.info( + "Correlation registry loaded: custom_kind_rules=%s custom_root_kinds=%s total_kind_rules=%s total_root_kinds=%s", + len(custom_kind_rules), + len(custom_root_cause_map), + len(merged_kind_rules), + len(merged_root_cause_map), + ) + + return cls( + event_kind_rules=merged_kind_rules, + root_cause_map=merged_root_cause_map, + ) + + +def _load_yaml(path: str) -> dict[str, Any]: + file_path = Path(path) + if not file_path.exists(): + logger.warning("Correlation YAML file not found: %s. Using defaults/merge fallback.", path) + return {} + + try: + with file_path.open("r", encoding="utf-8") as f: + data = yaml.safe_load(f) or {} + if not isinstance(data, dict): + logger.warning("Correlation YAML root must be a mapping: %s", path) + return {} + return data + except Exception as exc: + logger.exception("Failed to load correlation YAML %s: %s", path, exc) + return {} + + +def _load_event_kind_rules(path: str) -> list[EventKindRule]: + data = _load_yaml(path) + raw_rules = data.get("event_kind_rules", []) + if not isinstance(raw_rules, list): + logger.warning("event_kind_rules must be a list in %s", path) + return [] + + rules: list[EventKindRule] = [] + + for idx, item in enumerate(raw_rules, start=1): + if not isinstance(item, dict): + logger.warning("Skipping invalid event_kind_rules[%s] in %s", idx, path) + continue + + kind = _norm(item.get("kind")) + if not kind: + logger.warning("Skipping event_kind_rules[%s] without kind in %s", idx, path) + continue + + rules.append( + EventKindRule( + kind=kind, + trigger_patterns=tuple(item.get("trigger_patterns", []) or []), + item_key_patterns=tuple(item.get("item_key_patterns", []) or []), + service_patterns=tuple(item.get("service_patterns", []) or []), + tag_patterns=tuple(item.get("tag_patterns", []) or []), + scope_in=tuple(_norm(v) for v in (item.get("scope_in", []) or [])), + severity_in=tuple(_norm(v) for v in (item.get("severity_in", []) or [])), + ) + ) + + return rules + + +def _load_root_cause_map(path: str) -> dict[str, set[str]]: + data = _load_yaml(path) + raw_map = data.get("root_cause_map", {}) + if not isinstance(raw_map, dict): + logger.warning("root_cause_map must be a mapping in %s", path) + return {} + + result: dict[str, set[str]] = {} + for root_kind_raw, payload in raw_map.items(): + root_kind = _norm(root_kind_raw) + if not root_kind: + continue + + explains: list[str] = [] + if isinstance(payload, dict): + explains_raw = payload.get("explains", []) + if isinstance(explains_raw, list): + explains = [_norm(v) for v in explains_raw if _norm(v)] + elif isinstance(payload, list): + explains = [_norm(v) for v in payload if _norm(v)] + + if explains: + result[root_kind] = set(explains) + + return result + + +def _shared_service_scope_or_domain( + envelope: ProcessorForwardEnvelope, + candidate: CorrelationEventRecord, +) -> bool: + event = envelope.event + zbx_context = event.zabbix_context or {} + + current_tags = {str(k).lower(): str(v).strip() for k, v in (event.tags or {}).items()} + current_service = _norm(current_tags.get("service") or zbx_context.get("service") or event.service) + current_scope = _norm(current_tags.get("scope") or zbx_context.get("scope")) + current_domain = _norm(current_tags.get("domain") or zbx_context.get("domain")) + + candidate_service = _norm(candidate.service or candidate.tags.get("service")) + candidate_scope = _norm(candidate.scope or candidate.tags.get("scope")) + candidate_domain = _norm(candidate.domain or candidate.tags.get("domain")) + + if current_scope and candidate_scope and current_scope == candidate_scope: + return True + if current_service and candidate_service and current_service == candidate_service: + return True + if current_domain and candidate_domain and current_domain == candidate_domain: + return True + if event.host and candidate.host and _norm(event.host) == _norm(candidate.host): + return True + return False + + +def assess_correlation( + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + recent_events: list[CorrelationEventRecord], + registry: CorrelationRegistry, +) -> CorrelationAssessment: + kind = registry.classify_event_kind(envelope, decision) + root_candidate = registry.is_root_candidate(kind) + + if kind == "unknown": + return CorrelationAssessment( + applied=False, + role="standalone", + kind=kind, + group_id=None, + reason=None, + root_cause_candidate=False, + correlated_event_count=0, + ) + + parent_candidates = [ + item + for item in recent_events + if item.root_candidate + and registry.explains(item.kind, kind) + and _shared_service_scope_or_domain(envelope, item) + ] + parent_candidates.sort( + key=lambda item: (item.timestamp, severity_rank(item.severity)), + reverse=True, + ) + + if parent_candidates: + parent = parent_candidates[0] + suppress_child = ( + settings.correlation_suppress_children + and decision.routing_class in { + "average_priority", + "low_priority", + "triage_low_priority_notify", + "triage_low_priority_hold", + "triage_low_priority_suppressed", + } + ) + + return CorrelationAssessment( + applied=True, + role="child", + kind=kind, + group_id=parent.group_id or parent.event_id or parent.correlation_id, + reason=( + f"Likely downstream of {parent.kind}; matched recent event " + f"in the same service/scope/domain window within {settings.correlation_window_seconds}s" + ), + parent_event_id=parent.event_id, + parent_correlation_id=parent.correlation_id, + root_cause_candidate=False, + correlated_event_count=0, + suppress_child=suppress_child, + ) + + if root_candidate: + correlated_count = sum( + 1 + for item in recent_events + if registry.explains(kind, item.kind) and _shared_service_scope_or_domain(envelope, item) + ) + + return CorrelationAssessment( + applied=True, + role="root", + kind=kind, + group_id=envelope.event.event_id or envelope.event.correlation_id, + reason=( + f"Potential root cause candidate; {correlated_count} related alerts seen " + f"in matching service/scope/domain window within {settings.correlation_window_seconds}s" + ), + root_cause_candidate=True, + correlated_event_count=correlated_count, + suppress_child=False, + ) + + return CorrelationAssessment( + applied=True, + role="standalone", + kind=kind, + group_id=envelope.event.event_id or envelope.event.correlation_id, + reason="No root cause candidate found in matching service/scope/domain correlation window", + root_cause_candidate=False, + correlated_event_count=0, + suppress_child=False, + ) + + +def apply_correlation_to_decision( + decision: NotificationDecision, + assessment: CorrelationAssessment, +) -> NotificationDecision: + if not assessment.applied: + return decision + + update = { + "correlation_applied": True, + "correlation_role": assessment.role, + "correlation_group_id": assessment.group_id, + "correlation_kind": assessment.kind, + "correlation_reason": assessment.reason, + "correlation_source": assessment.source, + "correlation_confidence": assessment.confidence, + "parent_event_id": assessment.parent_event_id, + "parent_correlation_id": assessment.parent_correlation_id, + "root_cause_candidate": assessment.root_cause_candidate, + "correlated_event_count": assessment.correlated_event_count, + } + + if assessment.role == "child" and assessment.suppress_child: + update.update( + { + "notify": False, + "suppressed": True, + "channels": [], + "routing_class": "correlated_child_suppressed", + "reason": "Correlated child alert suppressed by RCA baseline", + "suppress_reason": assessment.reason, + } + ) + return decision.model_copy(update=update) + + if assessment.role == "child": + update["reason"] = f"{decision.reason}. RCA baseline matched a parent event in the same service/scope." + return decision.model_copy(update=update) + + if assessment.role == "root": + update["reason"] = f"{decision.reason}. RCA baseline marked this event as root cause candidate" + return decision.model_copy(update=update) + + return decision.model_copy(update=update) diff --git a/alert-processor/app/fingerprint.py b/alert-processor/app/fingerprint.py new file mode 100644 index 0000000..234add5 --- /dev/null +++ b/alert-processor/app/fingerprint.py @@ -0,0 +1,22 @@ +from __future__ import annotations + +import re + +from app.models import NormalizedEvent + + +def _norm(value: str | None) -> str: + if not value: + return "unknown" + value = value.strip().lower() + value = re.sub(r"\s+", "_", value) + value = re.sub(r"[^a-z0-9_\-\.]+", "_", value) + return value.strip("_") or "unknown" + + +def build_fingerprint(event: NormalizedEvent) -> str: + host = _norm(event.host) + service = _norm(event.service) + trigger = _norm(event.trigger_name) + + return f"{host}|{service}|{trigger}"