From 5b05a91a15faf4cacf63e32a69b12b53eefae824 Mon Sep 17 00:00:00 2001 From: Alexander Zubarev Date: Thu, 6 Aug 2026 18:34:01 +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/llm_correlation.py | 214 ++++++++++++++++++++++ alert-processor/app/llm_remediation.py | 239 +++++++++++++++++++++++++ alert-processor/app/llm_triage.py | 162 +++++++++++++++++ alert-processor/app/mail_notifier.py | 77 ++++++++ alert-processor/app/main.py | 216 ++++++++++++++++++++++ 5 files changed, 908 insertions(+) create mode 100644 alert-processor/app/llm_correlation.py create mode 100644 alert-processor/app/llm_remediation.py create mode 100644 alert-processor/app/llm_triage.py create mode 100644 alert-processor/app/mail_notifier.py create mode 100644 alert-processor/app/main.py diff --git a/alert-processor/app/llm_correlation.py b/alert-processor/app/llm_correlation.py new file mode 100644 index 0000000..286c0ff --- /dev/null +++ b/alert-processor/app/llm_correlation.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +import json +import re +from dataclasses import dataclass + +import httpx + +from app.correlation import CorrelationEventRecord +from app.models import NotificationDecision, ProcessorForwardEnvelope + + +@dataclass +class LLMCorrelationResult: + ok: bool + role: str + kind: str + reason: str | None + confidence: str | None + parent_event_id: str | None = None + parent_correlation_id: str | None = None + suppress_child: bool = False + error: str | None = None + + +class LLMCorrelationAdapter: + def __init__( + self, + base_url: str, + model: str, + timeout_seconds: float = 45, + verify_tls: bool = True, + temperature: float = 0.1, + ) -> None: + self.base_url = base_url.rstrip("/") + self.model = model + self.temperature = temperature + + self.client = httpx.AsyncClient( + timeout=timeout_seconds, + verify=verify_tls, + ) + + async def close(self) -> None: + await self.client.aclose() + + async def generate( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + recent_events: list[CorrelationEventRecord], + ) -> LLMCorrelationResult: + prompt = self._build_prompt(envelope, decision, recent_events) + + response = await self.client.post( + f"{self.base_url}/api/generate", + json={ + "model": self.model, + "prompt": prompt, + "stream": False, + "format": "json", + "think": False, + "options": { + "temperature": self.temperature, + }, + }, + ) + response.raise_for_status() + + payload = response.json() + raw_text = payload.get("response", "") + parsed = self._parse_json_object(raw_text) + + role = self._normalize_role(parsed.get("role")) + kind = self._normalize_text(parsed.get("kind")) or "unknown" + reason = self._normalize_text(parsed.get("reason")) + confidence = self._normalize_confidence(parsed.get("confidence")) + parent_event_id = self._normalize_text(parsed.get("parent_event_id")) + parent_correlation_id = self._normalize_text(parsed.get("parent_correlation_id")) + suppress_child = bool(parsed.get("suppress_child", False)) + + if role == "child" and not (parent_event_id or parent_correlation_id): + return LLMCorrelationResult( + ok=False, + role="standalone", + kind=kind, + reason=reason, + confidence=confidence, + error="LLM correlation returned child without parent reference", + ) + + return LLMCorrelationResult( + ok=True, + role=role, + kind=kind, + reason=reason, + confidence=confidence, + parent_event_id=parent_event_id, + parent_correlation_id=parent_correlation_id, + suppress_child=suppress_child, + error=None, + ) + + def _build_prompt( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + recent_events: list[CorrelationEventRecord], + ) -> str: + event = envelope.event + zbx_context = event.zabbix_context or {} + + current_event = { + "event_id": event.event_id, + "correlation_id": event.correlation_id, + "severity": decision.severity, + "routing_class": decision.routing_class, + "host": event.host, + "service": zbx_context.get("service") or event.service, + "scope": zbx_context.get("scope"), + "component": zbx_context.get("component"), + "domain": zbx_context.get("domain"), + "trigger_name": event.trigger_name, + "value": event.value, + "tags": event.tags, + "item_key": zbx_context.get("item_key"), + "alert_scope": zbx_context.get("alert_scope"), + "correlation_scopes": zbx_context.get("correlation_scopes"), + } + + recent_payload = [ + { + "event_id": item.event_id, + "correlation_id": item.correlation_id, + "kind": item.kind, + "severity": item.severity, + "routing_class": item.routing_class, + "root_candidate": item.root_candidate, + "role": item.role, + "group_id": item.group_id, + "parent_event_id": item.parent_event_id, + "parent_correlation_id": item.parent_correlation_id, + "host": item.host, + "service": item.service, + "scope": item.scope, + "component": item.component, + "domain": item.domain, + "tags": item.tags, + } + for item in recent_events[-12:] + ] + + return ( + "You are a cautious AIOps correlation assistant.\n" + "/no_think\n" + "Decide whether the current alert is standalone, a root cause candidate, or a downstream child event.\n" + "Rules:\n" + "1. Output JSON only.\n" + "2. Allowed role values: standalone, root, child.\n" + "3. Use the current event and recent events in the same service/scope/domain context.\n" + "4. Prefer standalone when uncertain.\n" + "5. child role requires a parent_event_id or parent_correlation_id from recent_events.\n" + "6. Never suggest suppress_child for High or Disaster alerts.\n" + "7. suppress_child may be true only for role=child and only when confidence is high.\n" + "8. confidence must be one of: low, medium, high.\n" + "9. Synthetic website alerts may be children of recent app/container outages if service/scope/domain matches.\n\n" + "Return exactly this JSON schema:\n" + "{\n" + ' "role": "standalone|root|child",\n' + ' "kind": "short_kind_name_or_unknown",\n' + ' "reason": "short explanation",\n' + ' "confidence": "low|medium|high",\n' + ' "parent_event_id": "optional",\n' + ' "parent_correlation_id": "optional",\n' + ' "suppress_child": false\n' + "}\n\n" + f"Current event:\n{json.dumps(current_event, ensure_ascii=False, indent=2)}\n\n" + f"Recent events:\n{json.dumps(recent_payload, ensure_ascii=False, indent=2)}" + ) + + @staticmethod + def _parse_json_object(raw_text: str) -> dict: + text = raw_text.strip() + try: + return json.loads(text) + except json.JSONDecodeError: + pass + + fenced = re.search(r"\{.*\}", text, flags=re.DOTALL) + if fenced: + return json.loads(fenced.group(0)) + + raise ValueError(f"Unable to parse LLM JSON response: {raw_text}") + + @staticmethod + def _normalize_role(value: object) -> str: + role = str(value or "standalone").strip().lower() + if role not in {"standalone", "root", "child"}: + return "standalone" + return role + + @staticmethod + def _normalize_confidence(value: object) -> str: + confidence = str(value or "low").strip().lower() + if confidence not in {"low", "medium", "high"}: + return "low" + return confidence + + @staticmethod + def _normalize_text(value: object) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None diff --git a/alert-processor/app/llm_remediation.py b/alert-processor/app/llm_remediation.py new file mode 100644 index 0000000..a46e86b --- /dev/null +++ b/alert-processor/app/llm_remediation.py @@ -0,0 +1,239 @@ +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from typing import Any + +import httpx + +from app.models import NotificationDecision, ProcessorForwardEnvelope + + +@dataclass +class RemediationResult: + ok: bool + summary: str | None + steps: list[str] + commands: list[str] + error: str | None = None + + +class LLMRemediationAdapter: + def __init__( + self, + base_url: str, + model: str, + timeout_seconds: float = 45, + verify_tls: bool = True, + temperature: float = 0.1, + max_steps: int = 4, + max_commands: int = 4, + ) -> None: + self.base_url = base_url.rstrip("/") + self.model = model + self.temperature = temperature + self.max_steps = max_steps + self.max_commands = max_commands + + self.client = httpx.AsyncClient( + timeout=timeout_seconds, + verify=verify_tls, + ) + + async def close(self) -> None: + await self.client.aclose() + + async def generate( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + ) -> RemediationResult: + prompt = self._build_prompt(envelope, decision) + + response = await self.client.post( + f"{self.base_url}/api/generate", + json={ + "model": self.model, + "prompt": prompt, + "stream": False, + "format": "json", + "think": False, + "options": { + "temperature": self.temperature, + }, + }, + ) + response.raise_for_status() + + data = response.json() + raw_text = data.get("response", "") + + parsed = self._parse_json_object(raw_text) + + summary = self._normalize_text(parsed.get("summary")) + steps = self._normalize_list(parsed.get("steps"), self.max_steps) + commands = self._normalize_list(parsed.get("commands"), self.max_commands) + + # Пост-фильтрация команд + alert_scope = ( + envelope.event.zabbix_context.get("alert_scope") + if envelope.event.zabbix_context + else "unknown" + ) + commands = self._post_filter_commands(commands, alert_scope=alert_scope) + + if not summary and not steps and not commands: + return RemediationResult( + ok=False, + summary=None, + steps=[], + commands=[], + error="LLM returned empty remediation payload", + ) + + return RemediationResult( + ok=True, + summary=summary, + steps=steps, + commands=commands, + error=None, + ) + + def _build_prompt( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + ) -> str: + event = envelope.event + zbx_context = event.zabbix_context or {} + + context = { + "severity": decision.severity, + "reason": decision.reason, + "event_phase": decision.event_phase, + "host": event.host, + "service": event.service, + "trigger_name": event.trigger_name, + "fingerprint": decision.fingerprint, + "repeat_count": decision.repeat_count, + "item_id": event.item_id, + "trigger_id": event.trigger_id, + "event_id": event.event_id, + "value": event.value, + "tags": event.tags, + "zabbix_context": zbx_context, + "alert_scope": zbx_context.get("alert_scope", "unknown"), + } + + return ( + "You are a cautious SRE assistant. /no_think\n" + "Generate remediation guidance for a Zabbix alert.\n" + "Rules:\n" + "1. Output JSON only.\n" + "2. Do not suggest destructive commands.\n" + "3. Commands must be diagnostic or read-only.\n" + "4. Keep it concise.\n" + "5. If this is a recovery event, return empty remediation.\n" + "6. Do not assume that host name indicates container name.\n" + "7. If alert_scope is host_os, diagnose the Linux host, not a container.\n" + "8. Only suggest docker/container commands when alert_scope is container.\n" + "9. Prefer generic Linux diagnostic commands for host_os alerts.\n" + "10. Avoid restart, rm, kill -9, stop, prune, delete, drop, truncate, format.\n\n" + "Return exactly this JSON schema:\n" + "{\n" + ' "summary": "short explanation",\n' + ' "steps": ["step1", "step2"],\n' + ' "commands": ["cmd1", "cmd2"]\n' + "}\n\n" + f"Alert context:\n{json.dumps(context, ensure_ascii=False, indent=2)}" + ) + + @staticmethod + def _parse_json_object(raw_text: str) -> dict[str, Any]: + text = raw_text.strip() + + try: + return json.loads(text) + except json.JSONDecodeError: + pass + + fenced = re.search(r"\{.*\}", text, flags=re.DOTALL) + if fenced: + return json.loads(fenced.group(0)) + + raise ValueError(f"Unable to parse LLM JSON response: {raw_text}") + + @staticmethod + def _normalize_text(value: Any) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None + + @staticmethod + def _normalize_list(value: Any, limit: int) -> list[str]: + if value is None: + return [] + + if isinstance(value, list): + result = [] + for item in value: + text = str(item).strip() + if text: + result.append(text) + return result[:limit] + + text = str(value).strip() + return [text] if text else [] + + @staticmethod + def _post_filter_commands(commands: list[str], alert_scope: str) -> list[str]: + destructive_patterns = ( + r"\brm\b", + r"\bmv\b", + r"\bshutdown\b", + r"\breboot\b", + r"\bpoweroff\b", + r"\bhalt\b", + r"\bmkfs\b", + r"\bfdisk\b", + r"\bdd\b", + r"\bkill\s+-9\b", + r"\bdocker\s+rm\b", + r"\bdocker\s+stop\b", + r"\bdocker\s+restart\b", + r"\bdocker\s+system\s+prune\b", + r"\bkubectl\s+delete\b", + r"\btruncate\b", + r"\bdrop\b", + ) + + container_patterns = ( + r"\bdocker\b", + r"\bpodman\b", + r"\bkubectl\b", + r"\bcrictl\b", + ) + + filtered: list[str] = [] + + for cmd in commands: + normalized = cmd.strip() + if not normalized: + continue + + lowered = normalized.lower() + + # Убираем деструктивные команды + if any(re.search(pattern, lowered) for pattern in destructive_patterns): + continue + + # Если это host_os, выбрасываем container-специфичные команды + if alert_scope == "host_os": + if any(re.search(pattern, lowered) for pattern in container_patterns): + continue + + filtered.append(normalized) + + return filtered \ No newline at end of file diff --git a/alert-processor/app/llm_triage.py b/alert-processor/app/llm_triage.py new file mode 100644 index 0000000..f491f80 --- /dev/null +++ b/alert-processor/app/llm_triage.py @@ -0,0 +1,162 @@ +from __future__ import annotations + +import json +import re +from dataclasses import dataclass +from typing import Any + +import httpx + +from app.models import NotificationDecision, ProcessorForwardEnvelope + + +@dataclass +class TriageResult: + ok: bool + verdict: str + classification: str | None + reason: str | None + error: str | None = None + + +class LLMTriageAdapter: + def __init__( + self, + base_url: str, + model: str, + timeout_seconds: float = 45, + verify_tls: bool = True, + temperature: float = 0.1, + ) -> None: + self.base_url = base_url.rstrip("/") + self.model = model + self.temperature = temperature + + self.client = httpx.AsyncClient( + timeout=timeout_seconds, + verify=verify_tls, + ) + + async def close(self) -> None: + await self.client.aclose() + + async def generate( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + ) -> TriageResult: + prompt = self._build_prompt(envelope, decision) + + response = await self.client.post( + f"{self.base_url}/api/generate", + json={ + "model": self.model, + "prompt": prompt, + "stream": False, + "format": "json", + "think": False, + "options": { + "temperature": self.temperature, + }, + }, + ) + response.raise_for_status() + + data = response.json() + raw_text = data.get("response", "") + parsed = self._parse_json_object(raw_text) + + verdict = self._normalize_verdict(parsed.get("verdict")) + classification = self._normalize_text(parsed.get("classification")) + reason = self._normalize_text(parsed.get("reason")) + + if verdict not in {"notify", "suppress", "hold"}: + return TriageResult( + ok=False, + verdict="hold", + classification=None, + reason=None, + error=f"Invalid triage verdict: {verdict}", + ) + + return TriageResult( + ok=True, + verdict=verdict, + classification=classification, + reason=reason, + error=None, + ) + + def _build_prompt( + self, + envelope: ProcessorForwardEnvelope, + decision: NotificationDecision, + ) -> str: + event = envelope.event + zbx_context = event.zabbix_context or {} + + context = { + "severity": decision.severity, + "reason": decision.reason, + "event_phase": decision.event_phase, + "host": event.host, + "service": event.service, + "trigger_name": event.trigger_name, + "fingerprint": decision.fingerprint, + "repeat_count": decision.repeat_count, + "item_id": event.item_id, + "trigger_id": event.trigger_id, + "event_id": event.event_id, + "value": event.value, + "tags": event.tags, + "zabbix_context": zbx_context, + "alert_scope": zbx_context.get("alert_scope", "unknown"), + } + + return ( + "You are a cautious AIOps triage assistant. /no_think\n" + "Decide whether a low-severity Zabbix alert should notify, suppress, or stay on hold.\n" + "Rules:\n" + "1. Output JSON only.\n" + "2. Allowed verdict values: notify, suppress, hold.\n" + "3. notify = actionable low-severity event worth sending to operator now.\n" + "4. suppress = obvious noise, repetition, or insignificant deviation.\n" + "5. hold = uncertain, informational, or not enough evidence.\n" + "6. Never escalate to email; low-severity notify means Matrix only.\n" + "7. Be conservative.\n\n" + "Return exactly this JSON schema:\n" + "{\n" + ' "verdict": "notify|suppress|hold",\n' + ' "classification": "actionable|noise|informational|unknown",\n' + ' "reason": "short explanation"\n' + "}\n\n" + f"Alert context:\n{json.dumps(context, ensure_ascii=False, indent=2)}" + ) + + @staticmethod + def _parse_json_object(raw_text: str) -> dict[str, Any]: + text = raw_text.strip() + + try: + return json.loads(text) + except json.JSONDecodeError: + pass + + fenced = re.search(r"\{.*\}", text, flags=re.DOTALL) + if fenced: + return json.loads(fenced.group(0)) + + raise ValueError(f"Unable to parse LLM JSON response: {raw_text}") + + @staticmethod + def _normalize_verdict(value: Any) -> str: + if value is None: + return "hold" + return str(value).strip().lower() + + @staticmethod + def _normalize_text(value: Any) -> str | None: + if value is None: + return None + text = str(value).strip() + return text or None \ No newline at end of file diff --git a/alert-processor/app/mail_notifier.py b/alert-processor/app/mail_notifier.py new file mode 100644 index 0000000..c2c52b8 --- /dev/null +++ b/alert-processor/app/mail_notifier.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +from dataclasses import dataclass +from email.message import EmailMessage +from pathlib import Path + +import aiosmtplib + + +@dataclass +class MailSendResult: + ok: bool + error: str | None + + +class MailNotifier: + def __init__( + self, + smtp_host: str, + smtp_port: int, + username: str, + password: str, + from_addr: str, + to_addr: str, + use_starttls: bool = True, + use_tls: bool = False, + timeout_seconds: float = 15, + ) -> None: + self.smtp_host = smtp_host + self.smtp_port = smtp_port + self.username = username + self.password = password + self.from_addr = from_addr + self.to_addr = to_addr + self.use_starttls = use_starttls + self.use_tls = use_tls + self.timeout_seconds = timeout_seconds + + async def send_message( + self, + subject: str, + body: str, + attachments: list[str] | None = None, + ) -> MailSendResult: + message = EmailMessage() + message["From"] = self.from_addr + message["To"] = self.to_addr + message["Subject"] = subject + message.set_content(body) + + for attachment in attachments or []: + path = Path(attachment) + if not path.exists() or not path.is_file(): + continue + + data = path.read_bytes() + message.add_attachment( + data, + maintype="image", + subtype="png", + filename=path.name, + ) + + try: + await aiosmtplib.send( + message, + hostname=self.smtp_host, + port=self.smtp_port, + username=self.username, + password=self.password, + start_tls=self.use_starttls, + use_tls=self.use_tls, + timeout=self.timeout_seconds, + ) + return MailSendResult(ok=True, error=None) + except Exception as exc: + return MailSendResult(ok=False, error=str(exc)) \ No newline at end of file diff --git a/alert-processor/app/main.py b/alert-processor/app/main.py new file mode 100644 index 0000000..ce2d746 --- /dev/null +++ b/alert-processor/app/main.py @@ -0,0 +1,216 @@ +from __future__ import annotations + +import logging + +from fastapi import FastAPI, Header, HTTPException, Query, status +from redis.asyncio import Redis + +from app.audit_logger import AuditLogger +from app.config import settings +from app.models import IngestAck, ProcessorForwardEnvelope +from app.queue_repo import RedisQueueRepository + + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(name)s %(message)s", +) + +logger = logging.getLogger(__name__) + +app = FastAPI( + title="alert-processor-ingest", + version="2.1.0", + description="Ingress endpoint that queues normalized events for async processing.", +) + +redis_client: Redis | None = None +queue_repo: RedisQueueRepository | None = None +audit_logger: AuditLogger | None = None + + +def _extract_token( + x_internal_token: str | None, + authorization: str | None, +) -> str | None: + if x_internal_token: + return x_internal_token.strip() + + if authorization: + auth = authorization.strip() + if auth.lower().startswith("bearer "): + return auth[7:].strip() + return auth + + return None + + +def _validate_token( + x_internal_token: str | None, + authorization: str | None, +) -> None: + if not settings.require_internal_api_token: + return + + provided = _extract_token(x_internal_token, authorization) + if not provided or provided != settings.internal_api_token: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid or missing internal API token", + ) + + +@app.on_event("startup") +async def startup_check() -> None: + global redis_client, queue_repo, audit_logger + + if settings.require_internal_api_token and not settings.internal_api_token: + raise RuntimeError( + "INTERNAL_API_TOKEN is required, but not set. " + "Set it in environment variables or in .env file." + ) + + if not settings.redis_enabled: + raise RuntimeError("REDIS_ENABLED must be true for async queue mode") + + if not settings.queue_enabled: + raise RuntimeError("QUEUE_ENABLED must be true for async queue 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() + + 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, + ) + + logger.info( + "Async ingest initialized: redis_url=%s queue_name=%s processing_name=%s deadletter_name=%s audit_enabled=%s", + settings.redis_url, + settings.queue_name, + settings.queue_processing_name, + settings.queue_deadletter_name, + settings.audit_enabled, + ) + + +@app.on_event("shutdown") +async def shutdown_event() -> None: + global redis_client + + if redis_client is not None: + await redis_client.close() + logger.info("Redis connection closed") + + +@app.get("/health") +async def health() -> dict[str, str | bool | int]: + return { + "status": "ok", + "service": settings.app_name, + "mode": "async_ingest", + "redis_enabled": settings.redis_enabled, + "redis_connected": redis_client is not None, + "queue_enabled": settings.queue_enabled, + "queue_configured": queue_repo is not None, + "queue_name": settings.queue_name, + "queue_processing_name": settings.queue_processing_name, + "queue_deadletter_name": settings.queue_deadletter_name, + "audit_enabled": settings.audit_enabled, + "audit_configured": audit_logger is not None, + } + + +@app.post( + "/internal/events", + response_model=IngestAck, + status_code=status.HTTP_202_ACCEPTED, +) +async def receive_internal_event( + envelope: ProcessorForwardEnvelope, + x_internal_token: str | None = Header(default=None), + authorization: str | None = Header(default=None), +) -> IngestAck: + _validate_token(x_internal_token, authorization) + + if queue_repo is None: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Queue repository is not initialized", + ) + + event = envelope.event + job_id = await queue_repo.enqueue_event(envelope) + + if audit_logger is not None: + await audit_logger.log_ingest_queued( + envelope=envelope, + job_id=job_id, + queue_name=settings.queue_name, + ) + + logger.info( + "Queued event for async processing: job_id=%s correlation_id=%s event_id=%s severity=%s host=%s trigger=%s", + job_id, + event.correlation_id, + event.event_id, + event.severity, + event.host, + event.trigger_name, + ) + + return IngestAck( + accepted=True, + correlation_id=event.correlation_id, + queued=True, + job_id=job_id, + message="Event queued for async processing", + ) + + +@app.get("/audit/events/{correlation_id}") +async def get_audit_by_correlation(correlation_id: str) -> dict: + if audit_logger is None: + raise HTTPException(status_code=503, detail="Audit logger is not enabled") + + payload = await audit_logger.get_event_audit(correlation_id=correlation_id) + if payload is None: + raise HTTPException(status_code=404, detail="Audit record not found") + + return payload + + +@app.get("/audit/by-event/{event_id}") +async def get_audit_by_event_id(event_id: str) -> dict: + if audit_logger is None: + raise HTTPException(status_code=503, detail="Audit logger is not enabled") + + payload = await audit_logger.get_event_audit(event_id=event_id) + if payload is None: + raise HTTPException(status_code=404, detail="Audit record not found") + + return payload + + +@app.get("/audit/recent") +async def get_recent_audit( + limit: int = Query(default=20, ge=1, le=100), +) -> list[dict]: + if audit_logger is None: + raise HTTPException(status_code=503, detail="Audit logger is not enabled") + + return await audit_logger.list_recent(limit=limit) \ No newline at end of file