From fa0c42f0b16c4e779510aba06de7126603911a76 Mon Sep 17 00:00:00 2001 From: Alexander Zubarev Date: Thu, 6 Aug 2026 18:34:32 +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/queue_repo.py | 126 +++++++ alert-processor/app/redis_repo.py | 464 +++++++++++++++++++++++++ alert-processor/app/tag_utils.py | 94 +++++ alert-processor/app/worker.py | 495 +++++++++++++++++++++++++++ alert-processor/app/worker_health.py | 177 ++++++++++ 5 files changed, 1356 insertions(+) create mode 100644 alert-processor/app/queue_repo.py create mode 100644 alert-processor/app/redis_repo.py create mode 100644 alert-processor/app/tag_utils.py create mode 100644 alert-processor/app/worker.py create mode 100644 alert-processor/app/worker_health.py diff --git a/alert-processor/app/queue_repo.py b/alert-processor/app/queue_repo.py new file mode 100644 index 0000000..77b7a04 --- /dev/null +++ b/alert-processor/app/queue_repo.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime, timezone + +from redis.asyncio import Redis + +from app.models import ProcessorForwardEnvelope + + +@dataclass +class QueueMessage: + job_id: str + attempt: int + envelope: ProcessorForwardEnvelope + raw_message: str + + +class RedisQueueRepository: + def __init__( + self, + client: Redis, + queue_name: str, + processing_name: str, + deadletter_name: str, + dedup_ttl_seconds: int, + ) -> None: + self.client = client + self.queue_name = queue_name + self.processing_name = processing_name + self.deadletter_name = deadletter_name + self.dedup_ttl_seconds = dedup_ttl_seconds + + async def ping(self) -> bool: + result = await self.client.ping() + return bool(result) + + async def enqueue_event(self, envelope: ProcessorForwardEnvelope) -> str: + job_id = envelope.event.correlation_id + payload = { + "job_id": job_id, + "attempt": 0, + "enqueued_at": datetime.now(timezone.utc).isoformat(), + "envelope": envelope.model_dump(mode="json"), + } + raw = json.dumps(payload, ensure_ascii=False) + await self.client.lpush(self.queue_name, raw) + return job_id + + async def claim_event(self, timeout_seconds: int) -> QueueMessage | None: + raw = await self.client.brpoplpush( + self.queue_name, + self.processing_name, + timeout=timeout_seconds, + ) + if raw is None: + return None + + raw_str = raw.decode() if isinstance(raw, bytes) else str(raw) + payload = json.loads(raw_str) + + return QueueMessage( + job_id=str(payload["job_id"]), + attempt=int(payload.get("attempt", 0)), + envelope=ProcessorForwardEnvelope.model_validate(payload["envelope"]), + raw_message=raw_str, + ) + + async def ack_message(self, raw_message: str) -> None: + await self.client.lrem(self.processing_name, 1, raw_message) + + async def requeue_message( + self, + message: QueueMessage, + error: str | None = None, + ) -> None: + payload = json.loads(message.raw_message) + payload["attempt"] = int(payload.get("attempt", 0)) + 1 + payload["last_error"] = error or "" + payload["requeued_at"] = datetime.now(timezone.utc).isoformat() + + new_raw = json.dumps(payload, ensure_ascii=False) + await self.client.lpush(self.queue_name, new_raw) + await self.ack_message(message.raw_message) + + async def deadletter_message( + self, + message: QueueMessage, + error: str | None = None, + ) -> None: + payload = json.loads(message.raw_message) + payload["deadlettered_at"] = datetime.now(timezone.utc).isoformat() + payload["last_error"] = error or "" + + raw = json.dumps(payload, ensure_ascii=False) + await self.client.lpush(self.deadletter_name, raw) + await self.ack_message(message.raw_message) + + async def requeue_processing_messages(self) -> int: + moved = 0 + while True: + raw = await self.client.rpoplpush(self.processing_name, self.queue_name) + if raw is None: + break + moved += 1 + return moved + + def build_processing_identity(self, envelope: ProcessorForwardEnvelope) -> str: + event = envelope.event + phase = (event.event_type or "problem").strip().lower() + stable_id = event.event_id or event.correlation_id + return f"{stable_id}:{phase}" + + async def is_processed(self, identity: str) -> bool: + key = self._processed_key(identity) + exists = await self.client.exists(key) + return bool(exists) + + async def mark_processed(self, identity: str) -> None: + key = self._processed_key(identity) + now = datetime.now(timezone.utc).isoformat() + await self.client.set(key, now, ex=self.dedup_ttl_seconds) + + def _processed_key(self, identity: str) -> str: + return f"alert:queue:processed:{identity}" \ No newline at end of file diff --git a/alert-processor/app/redis_repo.py b/alert-processor/app/redis_repo.py new file mode 100644 index 0000000..a4d14c8 --- /dev/null +++ b/alert-processor/app/redis_repo.py @@ -0,0 +1,464 @@ +from __future__ import annotations + +import json +from dataclasses import dataclass +from datetime import datetime, timezone +from typing import Iterable + +from redis.asyncio import Redis + +from app.correlation import CorrelationEventRecord +from app.models import NormalizedEvent, NotificationDecision + + +@dataclass +class FingerprintState: + fingerprint: str + count: int + first_seen: str + last_seen: str + last_severity: str | None + last_event_id: str | None + last_correlation_id: str | None + + +@dataclass +class SuppressState: + active: bool + last_notified_at: str | None + ttl_seconds: int + + +@dataclass +class OpenIncidentState: + active: bool + fingerprint: str + severity: str | None + routing_class: str | None + channels: list[str] + opened_at: str | None + last_problem_event_id: str | None + last_correlation_id: str | None + + +@dataclass +class FlapState: + active: bool + event_count: int + phases: list[str] + window_seconds: int + + +@dataclass +class TriageCacheState: + active: bool + verdict: str + classification: str | None + reason: str | None + source: str | None + ttl_seconds: int + + +class RedisStateRepository: + def __init__( + self, + client: Redis, + key_prefix: str, + fingerprint_ttl_seconds: int, + event_ttl_seconds: int, + suppress_window_seconds: int, + flap_window_seconds: int, + flap_threshold: int, + triage_cache_ttl_seconds: int, + ) -> None: + self.client = client + self.key_prefix = key_prefix + self.fingerprint_ttl_seconds = fingerprint_ttl_seconds + self.event_ttl_seconds = event_ttl_seconds + self.suppress_window_seconds = suppress_window_seconds + self.flap_window_seconds = flap_window_seconds + self.flap_threshold = flap_threshold + self.triage_cache_ttl_seconds = triage_cache_ttl_seconds + + def _fingerprint_key(self, fingerprint: str) -> str: + return f"{self.key_prefix}:fingerprint:{fingerprint}" + + def _event_key(self, event_id: str) -> str: + return f"{self.key_prefix}:event:{event_id}" + + def _suppress_key(self, fingerprint: str) -> str: + return f"{self.key_prefix}:suppress:{fingerprint}" + + def _incident_key(self, fingerprint: str) -> str: + return f"{self.key_prefix}:incident:{fingerprint}" + + def _flap_key(self, fingerprint: str) -> str: + return f"{self.key_prefix}:flap:{fingerprint}" + + def _triage_key(self, fingerprint: str, severity: str | None) -> str: + sev = (severity or "unknown").strip().lower().replace(" ", "_") + return f"{self.key_prefix}:triage:{fingerprint}:{sev}" + + def _correlation_key(self, scope_key: str) -> str: + safe_scope = scope_key.strip().lower().replace(" ", "_") + return f"{self.key_prefix}:correlation:{safe_scope}" + + async def ping(self) -> bool: + result = await self.client.ping() + return bool(result) + + async def update_fingerprint_state( + self, + fingerprint: str, + event: NormalizedEvent, + ) -> FingerprintState: + key = self._fingerprint_key(fingerprint) + now = datetime.now(timezone.utc).isoformat() + + exists = await self.client.exists(key) + if not exists: + await self.client.hset( + key, + mapping={ + "fingerprint": fingerprint, + "count": 1, + "first_seen": now, + "last_seen": now, + "last_severity": event.severity or "", + "last_event_id": event.event_id or "", + "last_correlation_id": event.correlation_id, + }, + ) + else: + await self.client.hincrby(key, "count", 1) + await self.client.hset( + key, + mapping={ + "last_seen": now, + "last_severity": event.severity or "", + "last_event_id": event.event_id or "", + "last_correlation_id": event.correlation_id, + }, + ) + + await self.client.expire(key, self.fingerprint_ttl_seconds) + + raw = await self.client.hgetall(key) + return FingerprintState( + fingerprint=(raw.get(b"fingerprint") or b"").decode(), + count=int((raw.get(b"count") or b"0").decode()), + first_seen=(raw.get(b"first_seen") or b"").decode(), + last_seen=(raw.get(b"last_seen") or b"").decode(), + last_severity=((raw.get(b"last_severity") or b"").decode() or None), + last_event_id=((raw.get(b"last_event_id") or b"").decode() or None), + last_correlation_id=((raw.get(b"last_correlation_id") or b"").decode() or None), + ) + + async def save_event_snapshot( + self, + event: NormalizedEvent, + fingerprint: str, + repeat_count: int, + ) -> None: + if not event.event_id: + return + + key = self._event_key(event.event_id) + await self.client.hset( + key, + mapping={ + "event_id": event.event_id, + "problem_id": event.problem_id or "", + "correlation_id": event.correlation_id, + "host": event.host or "", + "service": event.service or "", + "trigger_name": event.trigger_name or "", + "severity": event.severity or "", + "event_type": event.event_type or "", + "fingerprint": fingerprint, + "repeat_count": repeat_count, + "received_at": event.received_at.isoformat(), + }, + ) + await self.client.expire(key, self.event_ttl_seconds) + + async def get_suppress_state(self, fingerprint: str) -> SuppressState: + key = self._suppress_key(fingerprint) + value = await self.client.get(key) + ttl = await self.client.ttl(key) + + if value is None: + return SuppressState( + active=False, + last_notified_at=None, + ttl_seconds=0, + ) + + last_notified_at = value.decode() if isinstance(value, bytes) else str(value) + return SuppressState( + active=True, + last_notified_at=last_notified_at, + ttl_seconds=max(ttl, 0), + ) + + async def activate_suppress_window(self, fingerprint: str) -> str: + key = self._suppress_key(fingerprint) + now = datetime.now(timezone.utc).isoformat() + await self.client.set( + key, + now, + ex=self.suppress_window_seconds, + ) + return now + + async def clear_suppress_window(self, fingerprint: str) -> bool: + key = self._suppress_key(fingerprint) + deleted = await self.client.delete(key) + return bool(deleted) + + async def upsert_open_incident( + self, + fingerprint: str, + event: NormalizedEvent, + decision: NotificationDecision, + ) -> None: + key = self._incident_key(fingerprint) + now = datetime.now(timezone.utc).isoformat() + + await self.client.hset( + key, + mapping={ + "fingerprint": fingerprint, + "severity": event.severity or "", + "routing_class": decision.routing_class, + "channels": ",".join(decision.channels), + "opened_at": now, + "last_problem_event_id": event.event_id or "", + "last_correlation_id": event.correlation_id, + }, + ) + await self.client.expire(key, self.fingerprint_ttl_seconds) + + async def get_open_incident(self, fingerprint: str) -> OpenIncidentState | None: + key = self._incident_key(fingerprint) + raw = await self.client.hgetall(key) + if not raw: + return None + + channels_raw = (raw.get(b"channels") or b"").decode() + channels = [item for item in channels_raw.split(",") if item] + + return OpenIncidentState( + active=True, + fingerprint=(raw.get(b"fingerprint") or b"").decode(), + severity=((raw.get(b"severity") or b"").decode() or None), + routing_class=((raw.get(b"routing_class") or b"").decode() or None), + channels=channels, + opened_at=((raw.get(b"opened_at") or b"").decode() or None), + last_problem_event_id=((raw.get(b"last_problem_event_id") or b"").decode() or None), + last_correlation_id=((raw.get(b"last_correlation_id") or b"").decode() or None), + ) + + async def clear_open_incident(self, fingerprint: str) -> bool: + key = self._incident_key(fingerprint) + deleted = await self.client.delete(key) + return bool(deleted) + + async def record_phase_transition( + self, + fingerprint: str, + event_phase: str, + correlation_id: str, + event_id: str | None, + ) -> FlapState: + key = self._flap_key(fingerprint) + now_dt = datetime.now(timezone.utc) + now_ts = now_dt.timestamp() + member = f"{int(now_ts * 1000)}|{event_phase}|{correlation_id}|{event_id or ''}" + + min_score = now_ts - self.flap_window_seconds + + await self.client.zadd(key, {member: now_ts}) + await self.client.zremrangebyscore(key, 0, min_score) + await self.client.expire(key, max(self.flap_window_seconds * 2, 300)) + + raw_members = await self.client.zrange(key, 0, -1) + phases = self._extract_phases(raw_members) + event_count = len(phases) + + active = ( + event_count >= self.flap_threshold + and "problem" in phases + and "recovery" in phases + ) + + return FlapState( + active=active, + event_count=event_count, + phases=phases, + window_seconds=self.flap_window_seconds, + ) + + async def get_triage_cache( + self, + fingerprint: str, + severity: str | None, + ) -> TriageCacheState | None: + key = self._triage_key(fingerprint, severity) + raw = await self.client.hgetall(key) + if not raw: + return None + + ttl = await self.client.ttl(key) + return TriageCacheState( + active=True, + verdict=((raw.get(b"verdict") or b"").decode() or "hold"), + classification=((raw.get(b"classification") or b"").decode() or None), + reason=((raw.get(b"reason") or b"").decode() or None), + source=((raw.get(b"source") or b"").decode() or None), + ttl_seconds=max(ttl, 0), + ) + + async def save_triage_cache( + self, + fingerprint: str, + severity: str | None, + verdict: str, + classification: str | None, + reason: str | None, + source: str = "llm", + ) -> None: + key = self._triage_key(fingerprint, severity) + await self.client.hset( + key, + mapping={ + "verdict": verdict, + "classification": classification or "", + "reason": reason or "", + "source": source, + }, + ) + await self.client.expire(key, self.triage_cache_ttl_seconds) + + async def get_recent_correlation_events( + self, + host: str | None, + window_seconds: int, + scope_keys: list[str] | None = None, + ) -> list[CorrelationEventRecord]: + now_ts = datetime.now(timezone.utc).timestamp() + min_score = now_ts - window_seconds + + all_scope_keys = list(scope_keys or []) + if host: + host_key = f"host:{host.strip().lower().replace(' ', '_')}" + if host_key not in all_scope_keys: + all_scope_keys.append(host_key) + + result: list[CorrelationEventRecord] = [] + seen: set[str] = set() + + for scope_key in all_scope_keys: + key = self._correlation_key(scope_key) + await self.client.zremrangebyscore(key, 0, min_score) + raw_items = await self.client.zrangebyscore( + key, + min=min_score, + max=now_ts, + withscores=True, + ) + + for raw_member, score in raw_items: + member = raw_member.decode() if isinstance(raw_member, bytes) else str(raw_member) + payload = json.loads(member) + record = CorrelationEventRecord( + host=payload.get("host", ""), + event_id=payload.get("event_id"), + correlation_id=payload.get("correlation_id", ""), + kind=payload.get("kind", "unknown"), + severity=payload.get("severity"), + routing_class=payload.get("routing_class"), + fingerprint=payload.get("fingerprint"), + root_candidate=bool(payload.get("root_candidate", False)), + role=payload.get("role", "standalone"), + group_id=payload.get("group_id"), + parent_event_id=payload.get("parent_event_id"), + parent_correlation_id=payload.get("parent_correlation_id"), + timestamp=float(score), + tags=payload.get("tags") or {}, + service=payload.get("service"), + scope=payload.get("scope"), + component=payload.get("component"), + domain=payload.get("domain"), + ) + key_id = record.correlation_id or record.event_id or member + if key_id in seen: + continue + seen.add(key_id) + result.append(record) + + result.sort(key=lambda item: item.timestamp, reverse=True) + return result + + async def save_correlation_event( + self, + 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, + scope_keys: list[str] | None = None, + tags: dict[str, str] | None = None, + service: str | None = None, + scope: str | None = None, + component: str | None = None, + domain: str | None = None, + ) -> None: + now_ts = datetime.now(timezone.utc).timestamp() + payload = { + "host": host, + "event_id": event_id, + "correlation_id": correlation_id, + "kind": kind, + "severity": severity, + "routing_class": routing_class, + "fingerprint": fingerprint, + "root_candidate": root_candidate, + "role": role, + "group_id": group_id, + "parent_event_id": parent_event_id, + "parent_correlation_id": parent_correlation_id, + "tags": tags or {}, + "service": service, + "scope": scope, + "component": component, + "domain": domain, + } + raw = json.dumps(payload, ensure_ascii=False, sort_keys=True) + + all_scope_keys = list(scope_keys or []) + host_key = f"host:{host.strip().lower().replace(' ', '_')}" + if host_key not in all_scope_keys: + all_scope_keys.append(host_key) + + for scope_key in all_scope_keys: + key = self._correlation_key(scope_key) + await self.client.zadd(key, {raw: now_ts}) + await self.client.expire(key, self.fingerprint_ttl_seconds) + + @staticmethod + def _extract_phases(raw_members: Iterable[bytes | str]) -> list[str]: + phases: list[str] = [] + for member in raw_members: + value = member.decode() if isinstance(member, bytes) else str(member) + parts = value.split("|", 3) + if len(parts) >= 2: + phases.append(parts[1]) + return phases diff --git a/alert-processor/app/tag_utils.py b/alert-processor/app/tag_utils.py new file mode 100644 index 0000000..b1ddfe4 --- /dev/null +++ b/alert-processor/app/tag_utils.py @@ -0,0 +1,94 @@ +from __future__ import annotations + +import re +from typing import Any + + +def normalize_tag_dict(raw: Any) -> dict[str, str]: + if raw is None: + return {} + + if isinstance(raw, dict): + result: dict[str, str] = {} + for key, value in raw.items(): + if value is None: + continue + key_str = str(key).strip().lower() + value_str = str(value).strip() + if key_str and value_str: + result[key_str] = value_str + return result + + if isinstance(raw, list): + result: dict[str, str] = {} + for item in raw: + if not isinstance(item, dict): + continue + key = item.get("tag") or item.get("key") or item.get("name") + value = item.get("value") + if key is None or value is None: + continue + key_str = str(key).strip().lower() + value_str = str(value).strip() + if key_str and value_str: + result[key_str] = value_str + return result + + return {} + + +def merge_tags(*tag_sets: Any) -> dict[str, str]: + result: dict[str, str] = {} + for tag_set in tag_sets: + result.update(normalize_tag_dict(tag_set)) + return result + + +def safe_slug(value: str | None) -> str | None: + if not value: + return None + normalized = re.sub(r"[^a-zA-Z0-9._-]+", "_", value.strip().lower()).strip("_") + return normalized or None + + +def infer_domain_from_text(*candidates: Any) -> str | None: + for candidate in candidates: + if not candidate: + continue + text = str(candidate) + match = re.search(r"https?://([^/\s]+)", text, re.IGNORECASE) + if match: + return match.group(1).lower() + + match = re.search(r"([a-z0-9.-]+\.[a-z]{2,})(?:/|\s|$)", text, re.IGNORECASE) + if match: + return match.group(1).lower() + return None + + +def build_correlation_scopes(tags: dict[str, str], host: str | None, trigger_name: str | None) -> list[str]: + scopes: list[str] = [] + + scope = tags.get("scope") + service = tags.get("service") + component = tags.get("component") + domain = tags.get("domain") or infer_domain_from_text(trigger_name) + + if scope: + scopes.append(f"scope:{safe_slug(scope)}") + if service: + scopes.append(f"service:{safe_slug(service)}") + if domain: + scopes.append(f"domain:{safe_slug(domain)}") + if host: + scopes.append(f"host:{safe_slug(host)}") + if service and component: + scopes.append(f"service_component:{safe_slug(service)}:{safe_slug(component)}") + + seen: set[str] = set() + result: list[str] = [] + for item in scopes: + if item and item not in seen: + seen.add(item) + result.append(item) + return result diff --git a/alert-processor/app/worker.py b/alert-processor/app/worker.py new file mode 100644 index 0000000..4c8b688 --- /dev/null +++ b/alert-processor/app/worker.py @@ -0,0 +1,495 @@ +from __future__ import annotations + +import asyncio +import logging +import os +from datetime import datetime, timezone + +from redis.asyncio import Redis + +from app.audit_logger import AuditLogger +from app.config import settings +from app.correlation import CorrelationRegistry +from app.llm_remediation import LLMRemediationAdapter +from app.llm_triage import LLMTriageAdapter +from app.mail_notifier import MailNotifier +from app.matrix_notifier import MatrixNotifier +from app.matrix_token_manager import MatrixTokenManager +from app.notifications.dispatcher import NotificationDispatcher +from app.processor_service import ProcessorService +from app.queue_repo import RedisQueueRepository +from app.redis_repo import RedisStateRepository +from app.zabbix_client import ZabbixApiClient +from app.zabbix_enricher import ZabbixEnricher +from app.llm_correlation import LLMCorrelationAdapter + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", +) + +logger = logging.getLogger(__name__) + + +def _utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _worker_status_key() -> str: + return f"{settings.redis_key_prefix}:worker:status" + + +async def _publish_worker_status( + redis_client: Redis, + **fields: str | int | bool | None, +) -> None: + key = _worker_status_key() + + mapping: dict[str, str] = {} + for k, v in fields.items(): + if v is None: + continue + if isinstance(v, bool): + mapping[k] = "true" if v else "false" + else: + mapping[k] = str(v) + + mapping["heartbeat_at"] = _utc_now() + + if mapping: + await redis_client.hset(key, mapping=mapping) + + +async def _increment_worker_counter( + redis_client: Redis, + field_name: str, +) -> None: + key = _worker_status_key() + await redis_client.hincrby(key, field_name, 1) + await redis_client.hset(key, mapping={"heartbeat_at": _utc_now()}) + + +async def run_worker() -> None: + if not settings.redis_enabled: + raise RuntimeError("REDIS_ENABLED must be true for worker mode") + + if not settings.queue_enabled: + raise RuntimeError("QUEUE_ENABLED must be true for worker mode") + + redis_client = Redis.from_url( + settings.redis_url, + encoding="utf-8", + decode_responses=False, + ) + + queue_repo = RedisQueueRepository( + client=redis_client, + queue_name=settings.queue_name, + processing_name=settings.queue_processing_name, + deadletter_name=settings.queue_deadletter_name, + dedup_ttl_seconds=settings.queue_dedup_ttl_seconds, + ) + + await queue_repo.ping() + + audit_logger = None + if settings.audit_enabled: + audit_logger = AuditLogger( + client=redis_client, + key_prefix=settings.audit_key_prefix, + ttl_seconds=settings.audit_ttl_seconds, + max_stage_records=settings.audit_max_stage_records, + ) + + await _publish_worker_status( + redis_client, + state="starting", + started_at=_utc_now(), + pid=os.getpid(), + queue_name=settings.queue_name, + queue_processing_name=settings.queue_processing_name, + queue_deadletter_name=settings.queue_deadletter_name, + matrix_enabled=settings.matrix_enabled, + mail_enabled=settings.mail_enabled, + zabbix_api_enabled=settings.zabbix_api_enabled, + llm_enabled=settings.llm_enabled, + llm_triage_enabled=settings.llm_triage_enabled, + audit_enabled=settings.audit_enabled, + matrix_initialized=False, + mail_initialized=False, + zabbix_initialized=False, + llm_remediation_initialized=False, + llm_triage_initialized=False, + llm_correlation_initialized=False, + processed_count=0, + failed_count=0, + requeued_count=0, + deadletter_count=0, + ) + + if settings.queue_requeue_processing_on_startup: + moved = await queue_repo.requeue_processing_messages() + if moved: + logger.info("Requeued %s messages from processing list back to main queue", moved) + await _publish_worker_status( + redis_client, + processing_requeued_on_startup=moved, + last_requeue_at=_utc_now(), + ) + + redis_repo = RedisStateRepository( + client=redis_client, + key_prefix=settings.redis_key_prefix, + fingerprint_ttl_seconds=settings.redis_fingerprint_ttl_seconds, + event_ttl_seconds=settings.redis_event_ttl_seconds, + suppress_window_seconds=settings.suppress_window_seconds, + flap_window_seconds=settings.flap_window_seconds, + flap_threshold=settings.flap_threshold, + triage_cache_ttl_seconds=settings.llm_triage_cache_ttl_seconds, + ) + + matrix_token_manager = None + matrix_notifier = None + if settings.matrix_enabled: + matrix_token_manager = MatrixTokenManager( + token_endpoint=settings.matrix_oauth_token_endpoint, + client_id=settings.matrix_oauth_client_id, + client_secret=settings.matrix_oauth_client_secret or None, + initial_access_token=settings.matrix_access_token, + initial_refresh_token=settings.matrix_refresh_token, + initial_expires_in_seconds=settings.matrix_access_token_expires_in_seconds, + refresh_margin_seconds=settings.matrix_refresh_margin_seconds, + state_file=settings.matrix_token_state_file, + timeout_seconds=settings.matrix_request_timeout_seconds, + verify_tls=settings.matrix_verify_tls, + ) + matrix_token_manager.start_background_refresh() + + matrix_notifier = MatrixNotifier( + homeserver_url=settings.matrix_homeserver_url, + room_id=settings.matrix_room_id, + token_manager=matrix_token_manager, + message_type=settings.matrix_message_type, + timeout_seconds=settings.matrix_request_timeout_seconds, + verify_tls=settings.matrix_verify_tls, + ) + logger.info("Worker: Matrix notifier initialized") + await _publish_worker_status(redis_client, matrix_initialized=True) + + mail_notifier = None + if settings.mail_enabled: + mail_notifier = MailNotifier( + smtp_host=settings.mail_smtp_host, + smtp_port=settings.mail_smtp_port, + username=settings.mail_smtp_username, + password=settings.mail_smtp_password, + from_addr=settings.mail_from, + to_addr=settings.mail_to, + use_starttls=settings.mail_use_starttls, + use_tls=settings.mail_use_tls, + timeout_seconds=settings.mail_timeout_seconds, + ) + logger.info("Worker: Mail notifier initialized") + await _publish_worker_status(redis_client, mail_initialized=True) + + notification_dispatcher = NotificationDispatcher( + matrix_notifier=matrix_notifier, + mail_notifier=mail_notifier, + ) + + zabbix_api_client = None + zabbix_enricher = None + if settings.zabbix_api_enabled: + zabbix_api_client = ZabbixApiClient( + api_url=settings.zabbix_api_url, + api_token=settings.zabbix_api_token, + timeout_seconds=settings.zabbix_api_timeout_seconds, + verify_tls=settings.zabbix_api_verify_tls, + ) + zabbix_enricher = ZabbixEnricher( + client=zabbix_api_client, + web_url=settings.zabbix_web_url, + graph_period_hours=settings.zabbix_graph_period_hours, + graph_timezone=settings.zabbix_graph_timezone, + ) + logger.info("Worker: Zabbix enricher initialized") + await _publish_worker_status(redis_client, zabbix_initialized=True) + + llm_remediation_adapter = None + if settings.llm_enabled: + llm_remediation_adapter = LLMRemediationAdapter( + base_url=settings.llm_base_url, + model=settings.llm_model, + timeout_seconds=settings.llm_timeout_seconds, + verify_tls=settings.llm_verify_tls, + temperature=settings.llm_temperature, + max_steps=settings.llm_max_steps, + max_commands=settings.llm_max_commands, + ) + logger.info("Worker: LLM remediation adapter initialized") + await _publish_worker_status(redis_client, llm_remediation_initialized=True) + + llm_triage_adapter = None + if settings.llm_enabled and settings.llm_triage_enabled: + llm_triage_adapter = LLMTriageAdapter( + base_url=settings.llm_base_url, + model=settings.llm_model, + timeout_seconds=settings.llm_timeout_seconds, + verify_tls=settings.llm_verify_tls, + temperature=settings.llm_temperature, + ) + logger.info("Worker: LLM triage adapter initialized") + await _publish_worker_status(redis_client, llm_triage_initialized=True) + + llm_correlation_adapter = None + if settings.llm_enabled and settings.llm_correlation_enabled: + llm_correlation_adapter = LLMCorrelationAdapter( + base_url=settings.llm_base_url, + model=settings.llm_model, + timeout_seconds=settings.llm_timeout_seconds, + verify_tls=settings.llm_verify_tls, + temperature=settings.llm_temperature, + ) + logger.info("Worker: LLM correlation adapter initialized") + await _publish_worker_status(redis_client, llm_correlation_initialized=True) + + correlation_registry = None + if settings.correlation_enabled: + correlation_registry = CorrelationRegistry.from_yaml_files( + event_kind_rules_path=settings.correlation_kind_rules_path, + root_cause_map_path=settings.correlation_root_cause_path, + ) + logger.info("Worker: Correlation registry initialized") + + processor_service = ProcessorService( + redis_repo=redis_repo, + notification_dispatcher=notification_dispatcher, + zabbix_enricher=zabbix_enricher, + llm_remediation_adapter=llm_remediation_adapter, + llm_triage_adapter=llm_triage_adapter, + correlation_registry=correlation_registry, + audit_logger=audit_logger, + llm_correlation_adapter=llm_correlation_adapter, + ) + + logger.info("Worker started: queue=%s", settings.queue_name) + await _publish_worker_status(redis_client, state="running") + + try: + while True: + await _publish_worker_status(redis_client, state="idle") + + message = await queue_repo.claim_event( + timeout_seconds=settings.queue_block_timeout_seconds + ) + if message is None: + continue + + identity = queue_repo.build_processing_identity(message.envelope) + + await _publish_worker_status( + redis_client, + state="processing", + current_job_id=message.job_id, + current_identity=identity, + current_attempt=message.attempt, + current_correlation_id=message.envelope.event.correlation_id, + current_event_id=message.envelope.event.event_id or "", + ) + + if audit_logger is not None: + await audit_logger.log_worker_started( + envelope=message.envelope, + job_id=message.job_id, + attempt=message.attempt, + identity=identity, + ) + + if await queue_repo.is_processed(identity): + logger.info( + "Worker skipped already processed message: job_id=%s identity=%s", + message.job_id, + identity, + ) + await queue_repo.ack_message(message.raw_message) + + if audit_logger is not None: + await audit_logger.log_worker_outcome( + envelope=message.envelope, + state="duplicate_skipped", + attempt=message.attempt, + ) + + await _publish_worker_status( + redis_client, + state="idle", + last_skipped_job_id=message.job_id, + last_skipped_identity=identity, + current_job_id="", + current_identity="", + current_attempt="", + current_correlation_id="", + current_event_id="", + ) + continue + + try: + await processor_service.process(message.envelope) + await queue_repo.mark_processed(identity) + await queue_repo.ack_message(message.raw_message) + await _increment_worker_counter(redis_client, "processed_count") + + if audit_logger is not None: + await audit_logger.log_worker_outcome( + envelope=message.envelope, + state="processed", + attempt=message.attempt, + ) + + logger.info( + "Worker processed message successfully: job_id=%s identity=%s attempt=%s", + message.job_id, + identity, + message.attempt, + ) + + await _publish_worker_status( + redis_client, + state="idle", + last_processed_at=_utc_now(), + last_job_id=message.job_id, + last_identity=identity, + last_attempt=message.attempt, + last_correlation_id=message.envelope.event.correlation_id, + last_event_id=message.envelope.event.event_id or "", + last_error="", + current_job_id="", + current_identity="", + current_attempt="", + current_correlation_id="", + current_event_id="", + ) + except Exception as exc: + await _increment_worker_counter(redis_client, "failed_count") + + logger.exception( + "Worker failed to process message: job_id=%s identity=%s attempt=%s error=%s", + message.job_id, + identity, + message.attempt, + exc, + ) + + if audit_logger is not None: + await audit_logger.log_stage( + correlation_id=message.envelope.event.correlation_id, + event_id=message.envelope.event.event_id, + stage="worker_exception", + status="error", + details={ + "attempt": message.attempt, + "identity": identity, + "error": str(exc), + }, + ) + + await _publish_worker_status( + redis_client, + state="error", + last_error=str(exc), + last_error_at=_utc_now(), + last_failed_job_id=message.job_id, + last_failed_identity=identity, + last_failed_attempt=message.attempt, + ) + + if message.attempt + 1 < settings.queue_max_attempts: + await queue_repo.requeue_message(message, error=str(exc)) + await _increment_worker_counter(redis_client, "requeued_count") + + if audit_logger is not None: + await audit_logger.log_worker_outcome( + envelope=message.envelope, + state="requeued", + error=str(exc), + attempt=message.attempt + 1, + ) + + logger.warning( + "Worker requeued message: job_id=%s next_attempt=%s", + message.job_id, + message.attempt + 1, + ) + + await _publish_worker_status( + redis_client, + state="idle", + last_requeue_at=_utc_now(), + current_job_id="", + current_identity="", + current_attempt="", + current_correlation_id="", + current_event_id="", + ) + else: + await queue_repo.deadletter_message(message, error=str(exc)) + await _increment_worker_counter(redis_client, "deadletter_count") + + if audit_logger is not None: + await audit_logger.log_worker_outcome( + envelope=message.envelope, + state="deadletter", + error=str(exc), + attempt=message.attempt + 1, + ) + + logger.error( + "Worker sent message to deadletter: job_id=%s attempts=%s", + message.job_id, + message.attempt + 1, + ) + + await _publish_worker_status( + redis_client, + state="idle", + last_deadletter_at=_utc_now(), + current_job_id="", + current_identity="", + current_attempt="", + current_correlation_id="", + current_event_id="", + ) + finally: + await _publish_worker_status(redis_client, state="stopping") + + if zabbix_api_client is not None: + await zabbix_api_client.close() + + if llm_remediation_adapter is not None: + await llm_remediation_adapter.close() + + if llm_triage_adapter is not None: + await llm_triage_adapter.close() + + if llm_correlation_adapter is not None: + await llm_correlation_adapter.close() + + if matrix_notifier is not None: + await matrix_notifier.close() + + if matrix_token_manager is not None: + await matrix_token_manager.close() + + await _publish_worker_status(redis_client, state="stopped") + await redis_client.close() + + +def main() -> None: + try: + asyncio.run(run_worker()) + except KeyboardInterrupt: + logger.info("Worker stopped by user") + + +if __name__ == "__main__": + main() \ No newline at end of file diff --git a/alert-processor/app/worker_health.py b/alert-processor/app/worker_health.py new file mode 100644 index 0000000..b377374 --- /dev/null +++ b/alert-processor/app/worker_health.py @@ -0,0 +1,177 @@ +from __future__ import annotations + +from datetime import datetime, timezone + +from fastapi import FastAPI, HTTPException +from redis.asyncio import Redis + +from app.config import settings + + +app = FastAPI( + title="alert-worker-health", + version="1.0.0", + description="Reads worker runtime status from Redis and shows queue/module health.", +) + +redis_client: Redis | None = None + + +def _worker_status_key() -> str: + return f"{settings.redis_key_prefix}:worker:status" + + +def _to_bool(value: str | None) -> bool | None: + if value is None or value == "": + return None + return value.strip().lower() == "true" + + +def _to_int(value: str | None) -> int | None: + if value is None or value == "": + return None + try: + return int(value) + except ValueError: + return None + + +def _parse_iso(value: str | None) -> datetime | None: + if not value: + return None + try: + return datetime.fromisoformat(value) + except ValueError: + return None + + +def _is_worker_alive(heartbeat_at: str | None) -> bool: + hb = _parse_iso(heartbeat_at) + if hb is None: + return False + + now = datetime.now(timezone.utc) + delta = (now - hb).total_seconds() + return delta <= max(settings.queue_block_timeout_seconds * 3, 20) + + +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 + + +@app.on_event("startup") +async def startup_check() -> None: + global redis_client + + if not settings.redis_enabled: + raise RuntimeError("REDIS_ENABLED must be true for worker health service") + + redis_client = Redis.from_url( + settings.redis_url, + encoding="utf-8", + decode_responses=False, + ) + await redis_client.ping() + + +@app.on_event("shutdown") +async def shutdown_event() -> None: + global redis_client + + if redis_client is not None: + await redis_client.close() + + +@app.get("/health") +async def health() -> dict: + if redis_client is None: + raise HTTPException(status_code=503, detail="Redis client is not initialized") + + key = _worker_status_key() + raw_status = await redis_client.hgetall(key) + status = _decode_hash(raw_status) + + queue_len = await redis_client.llen(settings.queue_name) + processing_len = await redis_client.llen(settings.queue_processing_name) + deadletter_len = await redis_client.llen(settings.queue_deadletter_name) + + heartbeat_at = status.get("heartbeat_at") + worker_alive = _is_worker_alive(heartbeat_at) + worker_state = status.get("state") or "unknown" + + overall_status = "ok" + if not worker_alive: + overall_status = "degraded" + if deadletter_len > 0: + overall_status = "degraded" + if worker_state in {"error", "stopped"}: + overall_status = "degraded" + + return { + "status": overall_status, + "service": "alert-worker-health", + "redis_enabled": settings.redis_enabled, + "redis_connected": True, + "queue": { + "enabled": settings.queue_enabled, + "queue_name": settings.queue_name, + "processing_name": settings.queue_processing_name, + "deadletter_name": settings.queue_deadletter_name, + "queue_length": queue_len, + "processing_length": processing_len, + "deadletter_length": deadletter_len, + }, + "worker": { + "alive": worker_alive, + "state": worker_state, + "heartbeat_at": heartbeat_at, + "started_at": status.get("started_at"), + "pid": _to_int(status.get("pid")), + "current_job_id": status.get("current_job_id") or None, + "current_identity": status.get("current_identity") or None, + "current_attempt": _to_int(status.get("current_attempt")), + "last_job_id": status.get("last_job_id") or None, + "last_identity": status.get("last_identity") or None, + "last_attempt": _to_int(status.get("last_attempt")), + "last_processed_at": status.get("last_processed_at") or None, + "last_error": status.get("last_error") or None, + "last_error_at": status.get("last_error_at") or None, + "last_requeue_at": status.get("last_requeue_at") or None, + "last_deadletter_at": status.get("last_deadletter_at") or None, + "processed_count": _to_int(status.get("processed_count")) or 0, + "failed_count": _to_int(status.get("failed_count")) or 0, + "requeued_count": _to_int(status.get("requeued_count")) or 0, + "deadletter_count": _to_int(status.get("deadletter_count")) or 0, + }, + "modules": { + "matrix": { + "enabled": settings.matrix_enabled, + "initialized": _to_bool(status.get("matrix_initialized")), + }, + "mail": { + "enabled": settings.mail_enabled, + "initialized": _to_bool(status.get("mail_initialized")), + }, + "zabbix_api": { + "enabled": settings.zabbix_api_enabled, + "initialized": _to_bool(status.get("zabbix_initialized")), + }, + "llm_remediation": { + "enabled": settings.llm_enabled, + "initialized": _to_bool(status.get("llm_remediation_initialized")), + }, + "llm_triage": { + "enabled": settings.llm_enabled and settings.llm_triage_enabled, + "initialized": _to_bool(status.get("llm_triage_initialized")), + }, + "llm_correlation": { + "enabled": settings.llm_enabled and settings.llm_correlation_enabled, + "initialized": _to_bool(status.get("llm_correlation_initialized")), + }, + }, + } \ No newline at end of file