Загрузить файлы в «alert-processor/app»
This commit is contained in:
@@ -0,0 +1,220 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import mimetypes
|
||||
import uuid
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
|
||||
import httpx
|
||||
|
||||
from app.matrix_token_manager import MatrixTokenManager
|
||||
|
||||
|
||||
@dataclass
|
||||
class MatrixSendResult:
|
||||
ok: bool
|
||||
status_code: int | None
|
||||
event_id: str | None
|
||||
error: str | None
|
||||
content_uri: str | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class MatrixUploadResult:
|
||||
ok: bool
|
||||
status_code: int | None
|
||||
content_uri: str | None
|
||||
error: str | None
|
||||
|
||||
|
||||
class MatrixNotifier:
|
||||
def __init__(
|
||||
self,
|
||||
homeserver_url: str,
|
||||
room_id: str,
|
||||
token_manager: MatrixTokenManager,
|
||||
message_type: str = "m.notice",
|
||||
timeout_seconds: float = 10,
|
||||
verify_tls: bool = True,
|
||||
) -> None:
|
||||
self.homeserver_url = homeserver_url.rstrip("/")
|
||||
self.room_id = room_id
|
||||
self.token_manager = token_manager
|
||||
self.message_type = message_type
|
||||
self.client = httpx.AsyncClient(
|
||||
timeout=timeout_seconds,
|
||||
verify=verify_tls,
|
||||
)
|
||||
|
||||
async def close(self) -> None:
|
||||
await self.client.aclose()
|
||||
|
||||
async def send_message(self, body: str) -> MatrixSendResult:
|
||||
payload = {
|
||||
"msgtype": self.message_type,
|
||||
"body": body,
|
||||
}
|
||||
return await self._send_room_message(payload)
|
||||
|
||||
async def send_image(
|
||||
self,
|
||||
file_path: str,
|
||||
body: str | None = None,
|
||||
) -> MatrixSendResult:
|
||||
path = Path(file_path)
|
||||
if not path.exists() or not path.is_file():
|
||||
return MatrixSendResult(
|
||||
ok=False,
|
||||
status_code=None,
|
||||
event_id=None,
|
||||
error=f"Image file not found: {file_path}",
|
||||
content_uri=None,
|
||||
)
|
||||
|
||||
mime_type, _ = mimetypes.guess_type(path.name)
|
||||
mime_type = mime_type or "image/png"
|
||||
data = path.read_bytes()
|
||||
|
||||
upload_result = await self._upload_bytes(
|
||||
filename=path.name,
|
||||
data=data,
|
||||
content_type=mime_type,
|
||||
)
|
||||
if not upload_result.ok:
|
||||
return MatrixSendResult(
|
||||
ok=False,
|
||||
status_code=upload_result.status_code,
|
||||
event_id=None,
|
||||
error=upload_result.error,
|
||||
content_uri=None,
|
||||
)
|
||||
|
||||
payload = {
|
||||
"msgtype": "m.image",
|
||||
"body": body or path.name,
|
||||
"url": upload_result.content_uri,
|
||||
"info": {
|
||||
"mimetype": mime_type,
|
||||
"size": len(data),
|
||||
},
|
||||
}
|
||||
|
||||
send_result = await self._send_room_message(payload)
|
||||
send_result.content_uri = upload_result.content_uri
|
||||
return send_result
|
||||
|
||||
async def _send_room_message(self, payload: dict) -> MatrixSendResult:
|
||||
txn_id = str(uuid.uuid4())
|
||||
room_id_encoded = quote(self.room_id, safe="")
|
||||
url = (
|
||||
f"{self.homeserver_url}"
|
||||
f"/_matrix/client/v3/rooms/{room_id_encoded}"
|
||||
f"/send/m.room.message/{txn_id}"
|
||||
)
|
||||
|
||||
try:
|
||||
response = await self._request_with_refresh(
|
||||
method="PUT",
|
||||
url=url,
|
||||
json=payload,
|
||||
headers={"Content-Type": "application/json"},
|
||||
)
|
||||
data = response.json() if response.content else {}
|
||||
|
||||
if response.is_success:
|
||||
return MatrixSendResult(
|
||||
ok=True,
|
||||
status_code=response.status_code,
|
||||
event_id=data.get("event_id"),
|
||||
error=None,
|
||||
content_uri=None,
|
||||
)
|
||||
|
||||
return MatrixSendResult(
|
||||
ok=False,
|
||||
status_code=response.status_code,
|
||||
event_id=None,
|
||||
error=str(data),
|
||||
content_uri=None,
|
||||
)
|
||||
except Exception as exc:
|
||||
return MatrixSendResult(
|
||||
ok=False,
|
||||
status_code=None,
|
||||
event_id=None,
|
||||
error=str(exc),
|
||||
content_uri=None,
|
||||
)
|
||||
|
||||
async def _upload_bytes(
|
||||
self,
|
||||
filename: str,
|
||||
data: bytes,
|
||||
content_type: str,
|
||||
) -> MatrixUploadResult:
|
||||
url = f"{self.homeserver_url}/_matrix/media/v3/upload"
|
||||
|
||||
try:
|
||||
response = await self._request_with_refresh(
|
||||
method="POST",
|
||||
url=url,
|
||||
params={"filename": filename},
|
||||
content=data,
|
||||
headers={"Content-Type": content_type},
|
||||
)
|
||||
payload = response.json() if response.content else {}
|
||||
|
||||
if response.is_success:
|
||||
return MatrixUploadResult(
|
||||
ok=True,
|
||||
status_code=response.status_code,
|
||||
content_uri=payload.get("content_uri"),
|
||||
error=None,
|
||||
)
|
||||
|
||||
return MatrixUploadResult(
|
||||
ok=False,
|
||||
status_code=response.status_code,
|
||||
content_uri=None,
|
||||
error=str(payload),
|
||||
)
|
||||
except Exception as exc:
|
||||
return MatrixUploadResult(
|
||||
ok=False,
|
||||
status_code=None,
|
||||
content_uri=None,
|
||||
error=str(exc),
|
||||
)
|
||||
|
||||
async def _request_with_refresh(
|
||||
self,
|
||||
method: str,
|
||||
url: str,
|
||||
**kwargs,
|
||||
) -> httpx.Response:
|
||||
response = await self._request_once(method, url, **kwargs)
|
||||
|
||||
if response.status_code == 401:
|
||||
await self.token_manager.refresh_if_needed(force=True)
|
||||
response = await self._request_once(method, url, **kwargs)
|
||||
|
||||
return response
|
||||
|
||||
async def _request_once(
|
||||
self,
|
||||
method: str,
|
||||
url: str,
|
||||
**kwargs,
|
||||
) -> httpx.Response:
|
||||
token = await self.token_manager.get_access_token()
|
||||
|
||||
headers = dict(kwargs.pop("headers", {}) or {})
|
||||
headers["Authorization"] = f"Bearer {token}"
|
||||
|
||||
return await self.client.request(
|
||||
method=method,
|
||||
url=url,
|
||||
headers=headers,
|
||||
**kwargs,
|
||||
)
|
||||
@@ -0,0 +1,168 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from dataclasses import asdict, dataclass
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class MatrixTokenState:
|
||||
access_token: str
|
||||
refresh_token: str
|
||||
expires_at: float | None = None
|
||||
|
||||
|
||||
class MatrixTokenManager:
|
||||
def __init__(
|
||||
self,
|
||||
token_endpoint: str,
|
||||
client_id: str,
|
||||
client_secret: str | None,
|
||||
initial_access_token: str,
|
||||
initial_refresh_token: str,
|
||||
initial_expires_in_seconds: int | None,
|
||||
refresh_margin_seconds: int,
|
||||
state_file: str,
|
||||
timeout_seconds: float = 10,
|
||||
verify_tls: bool = True,
|
||||
) -> None:
|
||||
self.token_endpoint = token_endpoint
|
||||
self.client_id = client_id
|
||||
self.client_secret = client_secret or None
|
||||
self.refresh_margin_seconds = refresh_margin_seconds
|
||||
self.state_path = Path(state_file)
|
||||
self._lock = asyncio.Lock()
|
||||
self._stop_event = asyncio.Event()
|
||||
self._refresh_task: asyncio.Task | None = None
|
||||
self._refresh_failures = 0
|
||||
|
||||
self.client = httpx.AsyncClient(
|
||||
timeout=timeout_seconds,
|
||||
verify=verify_tls,
|
||||
)
|
||||
|
||||
state = self._load_state()
|
||||
if state is None:
|
||||
expires_at = None
|
||||
if initial_expires_in_seconds and initial_expires_in_seconds > 0:
|
||||
expires_at = time.time() + initial_expires_in_seconds
|
||||
|
||||
state = MatrixTokenState(
|
||||
access_token=initial_access_token,
|
||||
refresh_token=initial_refresh_token,
|
||||
expires_at=expires_at,
|
||||
)
|
||||
self._save_state(state)
|
||||
|
||||
self._state = state
|
||||
|
||||
def _load_state(self) -> MatrixTokenState | None:
|
||||
if not self.state_path.exists():
|
||||
return None
|
||||
|
||||
raw = json.loads(self.state_path.read_text(encoding="utf-8"))
|
||||
return MatrixTokenState(
|
||||
access_token=raw["access_token"],
|
||||
refresh_token=raw["refresh_token"],
|
||||
expires_at=raw.get("expires_at"),
|
||||
)
|
||||
|
||||
def _save_state(self, state: MatrixTokenState) -> None:
|
||||
self.state_path.write_text(
|
||||
json.dumps(asdict(state), ensure_ascii=False, indent=2),
|
||||
encoding="utf-8",
|
||||
)
|
||||
|
||||
def _needs_refresh(self) -> bool:
|
||||
if not self._state.refresh_token:
|
||||
return False
|
||||
if self._state.expires_at is None:
|
||||
return False
|
||||
return time.time() >= (self._state.expires_at - self.refresh_margin_seconds)
|
||||
|
||||
async def get_access_token(self) -> str:
|
||||
await self.refresh_if_needed()
|
||||
return self._state.access_token
|
||||
|
||||
async def refresh_if_needed(self, force: bool = False) -> MatrixTokenState:
|
||||
async with self._lock:
|
||||
if not force and not self._needs_refresh():
|
||||
return self._state
|
||||
|
||||
data = {
|
||||
"grant_type": "refresh_token",
|
||||
"refresh_token": self._state.refresh_token,
|
||||
"client_id": self.client_id,
|
||||
}
|
||||
if self.client_secret:
|
||||
data["client_secret"] = self.client_secret
|
||||
|
||||
response = await self.client.post(
|
||||
self.token_endpoint,
|
||||
data=data,
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
response.raise_for_status()
|
||||
payload = response.json()
|
||||
|
||||
access_token = payload["access_token"]
|
||||
refresh_token = payload.get("refresh_token", self._state.refresh_token)
|
||||
expires_in = payload.get("expires_in")
|
||||
expires_at = None
|
||||
if expires_in is not None:
|
||||
expires_at = time.time() + int(expires_in)
|
||||
|
||||
self._state = MatrixTokenState(
|
||||
access_token=access_token,
|
||||
refresh_token=refresh_token,
|
||||
expires_at=expires_at,
|
||||
)
|
||||
self._save_state(self._state)
|
||||
self._refresh_failures = 0
|
||||
|
||||
logger.info("Matrix OAuth token refreshed successfully")
|
||||
return self._state
|
||||
|
||||
def start_background_refresh(self) -> None:
|
||||
if self._refresh_task is None:
|
||||
self._refresh_task = asyncio.create_task(self._refresh_loop())
|
||||
|
||||
async def _refresh_loop(self) -> None:
|
||||
while not self._stop_event.is_set():
|
||||
try:
|
||||
await self.refresh_if_needed()
|
||||
except Exception as exc:
|
||||
self._refresh_failures += 1
|
||||
logger.exception("Matrix token background refresh failed: %s", exc)
|
||||
|
||||
if self._refresh_failures > 0:
|
||||
sleep_for = min(300, 30 * self._refresh_failures)
|
||||
else:
|
||||
sleep_for = 30
|
||||
if self._state.expires_at is not None:
|
||||
remaining = int(self._state.expires_at - time.time() - self.refresh_margin_seconds)
|
||||
sleep_for = max(5, min(60, remaining if remaining > 0 else 5))
|
||||
|
||||
try:
|
||||
await asyncio.wait_for(self._stop_event.wait(), timeout=sleep_for)
|
||||
except asyncio.TimeoutError:
|
||||
pass
|
||||
|
||||
async def close(self) -> None:
|
||||
self._stop_event.set()
|
||||
|
||||
if self._refresh_task is not None:
|
||||
self._refresh_task.cancel()
|
||||
try:
|
||||
await self._refresh_task
|
||||
except asyncio.CancelledError:
|
||||
pass
|
||||
|
||||
await self.client.aclose()
|
||||
@@ -0,0 +1,101 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
|
||||
class NormalizedEvent(BaseModel):
|
||||
source: str = "zabbix"
|
||||
correlation_id: str
|
||||
received_at: datetime
|
||||
remote_addr: str | None = None
|
||||
|
||||
event_id: str | None = None
|
||||
problem_id: str | None = None
|
||||
event_type: str = "problem"
|
||||
timestamp: datetime | None = None
|
||||
|
||||
severity: str | None = None
|
||||
severity_code: int | None = None
|
||||
|
||||
host: str | None = None
|
||||
host_id: str | None = None
|
||||
service: str | None = None
|
||||
|
||||
trigger_name: str | None = None
|
||||
trigger_id: str | None = None
|
||||
item_id: str | None = None
|
||||
|
||||
value: str | None = None
|
||||
opdata: str | None = None
|
||||
zabbix_url: str | None = None
|
||||
|
||||
event_url: str | None = None
|
||||
graph_url: str | None = None
|
||||
graph_image_path: str | None = None
|
||||
zabbix_context: dict[str, Any] = Field(default_factory=dict)
|
||||
|
||||
tags: dict[str, str] = Field(default_factory=dict)
|
||||
raw_payload: dict[str, Any]
|
||||
|
||||
|
||||
class ProcessorForwardEnvelope(BaseModel):
|
||||
event: NormalizedEvent
|
||||
|
||||
|
||||
class NotificationDecision(BaseModel):
|
||||
notify: bool
|
||||
severity: str | None = None
|
||||
channels: list[str] = Field(default_factory=list)
|
||||
reason: str
|
||||
routing_class: str
|
||||
fingerprint: str | None = None
|
||||
repeat_count: int = 1
|
||||
suppressed: bool = False
|
||||
suppress_reason: str | None = None
|
||||
event_phase: str = "problem"
|
||||
open_incident_found: bool = False
|
||||
recovered_from_severity: str | None = None
|
||||
flap_detected: bool = False
|
||||
flap_event_count: int = 0
|
||||
flap_reason: str | None = None
|
||||
|
||||
triage_applied: bool = False
|
||||
triage_source: str | None = None
|
||||
triage_verdict: str | None = None
|
||||
triage_reason: str | None = None
|
||||
triage_classification: str | None = None
|
||||
|
||||
correlation_applied: bool = False
|
||||
correlation_role: str | None = None
|
||||
correlation_group_id: str | None = None
|
||||
correlation_kind: str | None = None
|
||||
correlation_reason: str | None = None
|
||||
correlation_source: str | None = None
|
||||
correlation_confidence: str | None = None
|
||||
parent_event_id: str | None = None
|
||||
parent_correlation_id: str | None = None
|
||||
root_cause_candidate: bool = False
|
||||
correlated_event_count: int = 0
|
||||
|
||||
llm_enriched: bool = False
|
||||
remediation_summary: str | None = None
|
||||
remediation_steps: list[str] = Field(default_factory=list)
|
||||
remediation_commands: list[str] = Field(default_factory=list)
|
||||
|
||||
|
||||
class ProcessorAck(BaseModel):
|
||||
accepted: bool
|
||||
correlation_id: str
|
||||
decision: NotificationDecision
|
||||
message: str
|
||||
|
||||
|
||||
class IngestAck(BaseModel):
|
||||
accepted: bool
|
||||
correlation_id: str
|
||||
queued: bool
|
||||
job_id: str
|
||||
message: str
|
||||
@@ -0,0 +1,269 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from app.models import NormalizedEvent, NotificationDecision
|
||||
|
||||
|
||||
HIGH_SEVERITIES = {"high", "disaster"}
|
||||
AVERAGE_SEVERITIES = {"average"}
|
||||
LOW_SEVERITIES = {"warning", "information", "not classified", "not_classified", "info"}
|
||||
|
||||
RECOVERY_EVENT_TYPES = {"recovery", "resolved", "resolve", "ok", "clear", "closed"}
|
||||
|
||||
|
||||
def normalize_severity(value: str | None) -> str | None:
|
||||
if value is None:
|
||||
return None
|
||||
return value.strip().lower()
|
||||
|
||||
|
||||
def normalize_event_phase(event_type: str | None, value: str | None = None) -> str:
|
||||
raw = (event_type or "").strip().lower()
|
||||
if raw in RECOVERY_EVENT_TYPES:
|
||||
return "recovery"
|
||||
|
||||
if not raw and str(value or "").strip() == "0":
|
||||
return "recovery"
|
||||
|
||||
return "problem"
|
||||
|
||||
|
||||
def evaluate_event(
|
||||
event: NormalizedEvent,
|
||||
fingerprint: str,
|
||||
repeat_count: int,
|
||||
) -> NotificationDecision:
|
||||
severity = normalize_severity(event.severity)
|
||||
|
||||
if severity in HIGH_SEVERITIES:
|
||||
return NotificationDecision(
|
||||
notify=True,
|
||||
severity=event.severity,
|
||||
channels=["matrix", "mail"],
|
||||
reason="Severity is High/Disaster: mandatory notification",
|
||||
routing_class="high_priority",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="problem",
|
||||
)
|
||||
|
||||
if severity in AVERAGE_SEVERITIES:
|
||||
return NotificationDecision(
|
||||
notify=True,
|
||||
severity=event.severity,
|
||||
channels=["matrix"],
|
||||
reason="Severity is Average: notify by deterministic baseline policy",
|
||||
routing_class="average_priority",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="problem",
|
||||
)
|
||||
|
||||
if severity in LOW_SEVERITIES:
|
||||
return NotificationDecision(
|
||||
notify=False,
|
||||
severity=event.severity,
|
||||
channels=[],
|
||||
reason="Severity is Warning or lower: held for triage/suppress pipeline",
|
||||
routing_class="low_priority",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="problem",
|
||||
)
|
||||
|
||||
return NotificationDecision(
|
||||
notify=False,
|
||||
severity=event.severity,
|
||||
channels=[],
|
||||
reason="Unknown severity: conservative hold until policy is expanded",
|
||||
routing_class="unknown_priority",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="problem",
|
||||
)
|
||||
|
||||
|
||||
def build_recovery_decision(
|
||||
fingerprint: str,
|
||||
repeat_count: int,
|
||||
open_incident_found: bool,
|
||||
previous_severity: str | None,
|
||||
previous_channels: list[str] | None,
|
||||
previous_routing_class: str | None,
|
||||
) -> NotificationDecision:
|
||||
severity_norm = normalize_severity(previous_severity)
|
||||
|
||||
if not open_incident_found:
|
||||
return NotificationDecision(
|
||||
notify=False,
|
||||
severity=previous_severity,
|
||||
channels=[],
|
||||
reason="Recovery received but no matching open incident was found",
|
||||
routing_class="recovery_ignored",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="recovery",
|
||||
open_incident_found=False,
|
||||
recovered_from_severity=previous_severity,
|
||||
)
|
||||
|
||||
if previous_routing_class == "triage_low_priority_notify":
|
||||
return NotificationDecision(
|
||||
notify=True,
|
||||
severity=previous_severity,
|
||||
channels=previous_channels or ["matrix"],
|
||||
reason="Recovery matched a triaged low-severity incident",
|
||||
routing_class="recovery_triage_low_priority",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="recovery",
|
||||
open_incident_found=True,
|
||||
recovered_from_severity=previous_severity,
|
||||
)
|
||||
|
||||
if severity_norm in HIGH_SEVERITIES:
|
||||
return NotificationDecision(
|
||||
notify=True,
|
||||
severity=previous_severity,
|
||||
channels=previous_channels or ["matrix", "mail"],
|
||||
reason="Recovery matched an open High/Disaster incident",
|
||||
routing_class="recovery_high_priority",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="recovery",
|
||||
open_incident_found=True,
|
||||
recovered_from_severity=previous_severity,
|
||||
)
|
||||
|
||||
if severity_norm in AVERAGE_SEVERITIES:
|
||||
return NotificationDecision(
|
||||
notify=True,
|
||||
severity=previous_severity,
|
||||
channels=previous_channels or ["matrix"],
|
||||
reason="Recovery matched an open Average incident",
|
||||
routing_class="recovery_average_priority",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="recovery",
|
||||
open_incident_found=True,
|
||||
recovered_from_severity=previous_severity,
|
||||
)
|
||||
|
||||
return NotificationDecision(
|
||||
notify=False,
|
||||
severity=previous_severity,
|
||||
channels=[],
|
||||
reason="Recovery matched a low-priority incident: no notification by baseline policy",
|
||||
routing_class="recovery_low_priority",
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
event_phase="recovery",
|
||||
open_incident_found=True,
|
||||
recovered_from_severity=previous_severity,
|
||||
)
|
||||
|
||||
|
||||
def decision_supports_suppress(decision: NotificationDecision) -> bool:
|
||||
return decision.routing_class == "average_priority" and decision.notify
|
||||
|
||||
|
||||
def decision_supports_flap_suppress(decision: NotificationDecision) -> bool:
|
||||
return decision.routing_class in {
|
||||
"average_priority",
|
||||
"recovery_average_priority",
|
||||
} and decision.notify
|
||||
|
||||
|
||||
def is_low_severity_problem_candidate(
|
||||
decision: NotificationDecision,
|
||||
) -> bool:
|
||||
return decision.event_phase == "problem" and decision.routing_class == "low_priority"
|
||||
|
||||
|
||||
def apply_suppress_window(
|
||||
decision: NotificationDecision,
|
||||
ttl_seconds: int,
|
||||
) -> NotificationDecision:
|
||||
return decision.model_copy(
|
||||
update={
|
||||
"notify": False,
|
||||
"channels": [],
|
||||
"suppressed": True,
|
||||
"suppress_reason": f"Suppress window active, {ttl_seconds}s remaining",
|
||||
"reason": "Duplicate Average event suppressed by re-notify window",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def apply_flap_suppress(
|
||||
decision: NotificationDecision,
|
||||
event_count: int,
|
||||
window_seconds: int,
|
||||
) -> NotificationDecision:
|
||||
return decision.model_copy(
|
||||
update={
|
||||
"notify": False,
|
||||
"channels": [],
|
||||
"suppressed": True,
|
||||
"flap_detected": True,
|
||||
"flap_event_count": event_count,
|
||||
"flap_reason": f"Flapping detected: {event_count} phase changes within {window_seconds}s",
|
||||
"reason": "Average event suppressed because the series is flapping",
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def apply_low_severity_triage(
|
||||
decision: NotificationDecision,
|
||||
verdict: str,
|
||||
classification: str | None,
|
||||
reason: str | None,
|
||||
source: str,
|
||||
) -> NotificationDecision:
|
||||
verdict_norm = (verdict or "hold").strip().lower()
|
||||
classification_norm = (classification or "unknown").strip().lower()
|
||||
reason_text = (reason or "").strip() or "LLM triage applied"
|
||||
|
||||
common = {
|
||||
"triage_applied": True,
|
||||
"triage_source": source,
|
||||
"triage_verdict": verdict_norm,
|
||||
"triage_reason": reason_text,
|
||||
"triage_classification": classification_norm,
|
||||
}
|
||||
|
||||
if verdict_norm == "notify":
|
||||
return decision.model_copy(
|
||||
update={
|
||||
**common,
|
||||
"notify": True,
|
||||
"suppressed": False,
|
||||
"suppress_reason": None,
|
||||
"channels": ["matrix"],
|
||||
"routing_class": "triage_low_priority_notify",
|
||||
"reason": f"LLM triage marked low-severity event as actionable: {reason_text}",
|
||||
}
|
||||
)
|
||||
|
||||
if verdict_norm == "suppress":
|
||||
return decision.model_copy(
|
||||
update={
|
||||
**common,
|
||||
"notify": False,
|
||||
"suppressed": True,
|
||||
"channels": [],
|
||||
"routing_class": "triage_low_priority_suppressed",
|
||||
"reason": "LLM triage suppressed low-severity event",
|
||||
"suppress_reason": reason_text,
|
||||
}
|
||||
)
|
||||
|
||||
return decision.model_copy(
|
||||
update={
|
||||
**common,
|
||||
"notify": False,
|
||||
"suppressed": False,
|
||||
"channels": [],
|
||||
"routing_class": "triage_low_priority_hold",
|
||||
"reason": f"LLM triage kept low-severity event on hold: {reason_text}",
|
||||
}
|
||||
)
|
||||
@@ -0,0 +1,744 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from app.audit_logger import AuditLogger
|
||||
from app.config import settings
|
||||
from app.correlation import (
|
||||
CorrelationAssessment,
|
||||
CorrelationRegistry,
|
||||
apply_correlation_to_decision,
|
||||
assess_correlation,
|
||||
)
|
||||
from app.fingerprint import build_fingerprint
|
||||
from app.llm_correlation import LLMCorrelationAdapter, LLMCorrelationResult
|
||||
from app.llm_remediation import LLMRemediationAdapter
|
||||
from app.llm_triage import LLMTriageAdapter
|
||||
from app.models import NotificationDecision, ProcessorForwardEnvelope
|
||||
from app.notifications.dispatcher import NotificationDispatcher
|
||||
from app.policy import (
|
||||
apply_flap_suppress,
|
||||
apply_low_severity_triage,
|
||||
apply_suppress_window,
|
||||
build_recovery_decision,
|
||||
decision_supports_flap_suppress,
|
||||
decision_supports_suppress,
|
||||
evaluate_event,
|
||||
is_low_severity_problem_candidate,
|
||||
normalize_event_phase,
|
||||
)
|
||||
from app.redis_repo import FlapState, OpenIncidentState, RedisStateRepository
|
||||
from app.zabbix_enricher import ZabbixEnricher
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _confidence_rank(value: str | None) -> int:
|
||||
normalized = (value or "").strip().lower()
|
||||
if normalized == "high":
|
||||
return 3
|
||||
if normalized == "medium":
|
||||
return 2
|
||||
if normalized == "low":
|
||||
return 1
|
||||
return 0
|
||||
|
||||
|
||||
def _is_high_or_disaster(severity: str | None) -> bool:
|
||||
value = (severity or "").strip().lower()
|
||||
return value in {"high", "disaster"}
|
||||
|
||||
|
||||
class ProcessorService:
|
||||
def __init__(
|
||||
self,
|
||||
redis_repo: RedisStateRepository,
|
||||
notification_dispatcher: NotificationDispatcher,
|
||||
zabbix_enricher: ZabbixEnricher | None = None,
|
||||
llm_remediation_adapter: LLMRemediationAdapter | None = None,
|
||||
llm_triage_adapter: LLMTriageAdapter | None = None,
|
||||
correlation_registry: CorrelationRegistry | None = None,
|
||||
audit_logger: AuditLogger | None = None,
|
||||
llm_correlation_adapter: LLMCorrelationAdapter | None = None,
|
||||
) -> None:
|
||||
self.redis_repo = redis_repo
|
||||
self.notification_dispatcher = notification_dispatcher
|
||||
self.zabbix_enricher = zabbix_enricher
|
||||
self.llm_remediation_adapter = llm_remediation_adapter
|
||||
self.llm_triage_adapter = llm_triage_adapter
|
||||
self.correlation_registry = correlation_registry
|
||||
self.audit_logger = audit_logger
|
||||
self.llm_correlation_adapter = llm_correlation_adapter
|
||||
|
||||
async def process(self, envelope: ProcessorForwardEnvelope) -> NotificationDecision:
|
||||
event = envelope.event
|
||||
fingerprint = build_fingerprint(event)
|
||||
repeat_count = 1
|
||||
event_phase = normalize_event_phase(event.event_type, event.value)
|
||||
|
||||
async def audit(stage: str, status: str = "ok", details: dict | None = None) -> None:
|
||||
if self.audit_logger is not None:
|
||||
await self.audit_logger.log_stage(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
stage=stage,
|
||||
status=status,
|
||||
details=details,
|
||||
)
|
||||
|
||||
flap_state = FlapState(
|
||||
active=False,
|
||||
event_count=0,
|
||||
phases=[],
|
||||
window_seconds=settings.flap_window_seconds,
|
||||
)
|
||||
|
||||
fp_state = await self.redis_repo.update_fingerprint_state(
|
||||
fingerprint=fingerprint,
|
||||
event=event,
|
||||
)
|
||||
repeat_count = fp_state.count
|
||||
|
||||
await self.redis_repo.save_event_snapshot(
|
||||
event=event,
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
)
|
||||
|
||||
await audit(
|
||||
"state_updated",
|
||||
details={
|
||||
"fingerprint": fingerprint,
|
||||
"repeat_count": repeat_count,
|
||||
"event_phase": event_phase,
|
||||
},
|
||||
)
|
||||
|
||||
if settings.flap_enabled:
|
||||
flap_state = await self.redis_repo.record_phase_transition(
|
||||
fingerprint=fingerprint,
|
||||
event_phase=event_phase,
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
)
|
||||
await audit(
|
||||
"flap_evaluated",
|
||||
details={
|
||||
"flap_active": flap_state.active,
|
||||
"flap_event_count": flap_state.event_count,
|
||||
"window_seconds": flap_state.window_seconds,
|
||||
},
|
||||
)
|
||||
|
||||
baseline_low_severity_candidate = False
|
||||
|
||||
if event_phase == "recovery":
|
||||
open_incident: OpenIncidentState | None = await self.redis_repo.get_open_incident(
|
||||
fingerprint
|
||||
)
|
||||
|
||||
decision = build_recovery_decision(
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
open_incident_found=open_incident is not None,
|
||||
previous_severity=open_incident.severity if open_incident else None,
|
||||
previous_channels=open_incident.channels if open_incident else None,
|
||||
previous_routing_class=open_incident.routing_class if open_incident else None,
|
||||
)
|
||||
|
||||
if (
|
||||
flap_state.active
|
||||
and settings.flap_enabled
|
||||
and settings.flap_apply_to_average
|
||||
and decision_supports_flap_suppress(decision)
|
||||
):
|
||||
decision = apply_flap_suppress(
|
||||
decision=decision,
|
||||
event_count=flap_state.event_count,
|
||||
window_seconds=flap_state.window_seconds,
|
||||
)
|
||||
|
||||
await self.redis_repo.clear_suppress_window(fingerprint)
|
||||
await self.redis_repo.clear_open_incident(fingerprint)
|
||||
|
||||
else:
|
||||
decision = evaluate_event(
|
||||
event=event,
|
||||
fingerprint=fingerprint,
|
||||
repeat_count=repeat_count,
|
||||
)
|
||||
|
||||
if (
|
||||
flap_state.active
|
||||
and settings.flap_enabled
|
||||
and settings.flap_apply_to_average
|
||||
and decision_supports_flap_suppress(decision)
|
||||
):
|
||||
decision = apply_flap_suppress(
|
||||
decision=decision,
|
||||
event_count=flap_state.event_count,
|
||||
window_seconds=flap_state.window_seconds,
|
||||
)
|
||||
elif (
|
||||
settings.suppress_enabled
|
||||
and settings.suppress_apply_to_average
|
||||
and decision_supports_suppress(decision)
|
||||
):
|
||||
suppress_state = await self.redis_repo.get_suppress_state(fingerprint)
|
||||
|
||||
if suppress_state.active:
|
||||
decision = apply_suppress_window(
|
||||
decision=decision,
|
||||
ttl_seconds=suppress_state.ttl_seconds,
|
||||
)
|
||||
elif decision.notify:
|
||||
await self.redis_repo.activate_suppress_window(fingerprint)
|
||||
|
||||
baseline_low_severity_candidate = (
|
||||
settings.llm_triage_enabled
|
||||
and self.llm_triage_adapter is not None
|
||||
and is_low_severity_problem_candidate(decision)
|
||||
and not decision.suppressed
|
||||
)
|
||||
|
||||
await audit(
|
||||
"baseline_policy_applied",
|
||||
details={
|
||||
"notify": decision.notify,
|
||||
"suppressed": decision.suppressed,
|
||||
"routing_class": decision.routing_class,
|
||||
"reason": decision.reason,
|
||||
},
|
||||
)
|
||||
|
||||
should_enrich = self.zabbix_enricher is not None
|
||||
if settings.zabbix_enrich_only_notify:
|
||||
should_enrich = should_enrich and (
|
||||
(decision.notify and not decision.suppressed) or baseline_low_severity_candidate
|
||||
)
|
||||
|
||||
if should_enrich and self.zabbix_enricher is not None:
|
||||
try:
|
||||
await self.zabbix_enricher.enrich_event(event)
|
||||
await audit(
|
||||
"zabbix_enrichment",
|
||||
details={
|
||||
"event_url": event.event_url,
|
||||
"graph_url": event.graph_url,
|
||||
"graph_image_path": event.graph_image_path,
|
||||
"zabbix_context_keys": sorted(list((event.zabbix_context or {}).keys())),
|
||||
},
|
||||
)
|
||||
except Exception as exc:
|
||||
await audit(
|
||||
"zabbix_enrichment",
|
||||
status="error",
|
||||
details={"error": str(exc)},
|
||||
)
|
||||
logger.exception(
|
||||
"Zabbix enrichment failed: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
exc,
|
||||
)
|
||||
|
||||
if baseline_low_severity_candidate and self.llm_triage_adapter is not None:
|
||||
triage_applied = False
|
||||
|
||||
cached = await self.redis_repo.get_triage_cache(
|
||||
fingerprint=fingerprint,
|
||||
severity=decision.severity,
|
||||
)
|
||||
if cached is not None:
|
||||
decision = apply_low_severity_triage(
|
||||
decision=decision,
|
||||
verdict=cached.verdict,
|
||||
classification=cached.classification,
|
||||
reason=cached.reason,
|
||||
source="cache",
|
||||
)
|
||||
triage_applied = True
|
||||
|
||||
await audit(
|
||||
"llm_triage",
|
||||
details={
|
||||
"source": "cache",
|
||||
"verdict": cached.verdict,
|
||||
"classification": cached.classification,
|
||||
"reason": cached.reason,
|
||||
},
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"LLM triage reused from cache: correlation_id=%s event_id=%s verdict=%s classification=%s ttl=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
cached.verdict,
|
||||
cached.classification,
|
||||
cached.ttl_seconds,
|
||||
)
|
||||
|
||||
if not triage_applied:
|
||||
try:
|
||||
triage = await self.llm_triage_adapter.generate(envelope, decision)
|
||||
if triage.ok:
|
||||
decision = apply_low_severity_triage(
|
||||
decision=decision,
|
||||
verdict=triage.verdict,
|
||||
classification=triage.classification,
|
||||
reason=triage.reason,
|
||||
source="llm",
|
||||
)
|
||||
await self.redis_repo.save_triage_cache(
|
||||
fingerprint=fingerprint,
|
||||
severity=decision.severity,
|
||||
verdict=triage.verdict,
|
||||
classification=triage.classification,
|
||||
reason=triage.reason,
|
||||
source="llm",
|
||||
)
|
||||
|
||||
await audit(
|
||||
"llm_triage",
|
||||
details={
|
||||
"source": "llm",
|
||||
"verdict": triage.verdict,
|
||||
"classification": triage.classification,
|
||||
"reason": triage.reason,
|
||||
},
|
||||
)
|
||||
else:
|
||||
await audit(
|
||||
"llm_triage",
|
||||
status="error",
|
||||
details={"error": triage.error},
|
||||
)
|
||||
logger.warning(
|
||||
"LLM triage returned no usable content: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
triage.error,
|
||||
)
|
||||
except Exception as exc:
|
||||
await audit(
|
||||
"llm_triage",
|
||||
status="error",
|
||||
details={"error": str(exc)},
|
||||
)
|
||||
logger.exception(
|
||||
"LLM triage failed: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
exc,
|
||||
)
|
||||
|
||||
if (
|
||||
settings.correlation_enabled
|
||||
and self.correlation_registry is not None
|
||||
and decision.event_phase == "problem"
|
||||
):
|
||||
try:
|
||||
zbx_context = event.zabbix_context or {}
|
||||
scope_keys = list(zbx_context.get("correlation_scopes") or [])
|
||||
|
||||
recent = await self.redis_repo.get_recent_correlation_events(
|
||||
host=event.host,
|
||||
scope_keys=scope_keys,
|
||||
window_seconds=settings.correlation_window_seconds,
|
||||
)
|
||||
|
||||
deterministic_assessment = assess_correlation(
|
||||
envelope=envelope,
|
||||
decision=decision,
|
||||
recent_events=recent,
|
||||
registry=self.correlation_registry,
|
||||
)
|
||||
|
||||
chosen_assessment: CorrelationAssessment | None = (
|
||||
deterministic_assessment if deterministic_assessment.applied else None
|
||||
)
|
||||
|
||||
strong_deterministic = (
|
||||
deterministic_assessment.applied
|
||||
and deterministic_assessment.role in {"root", "child"}
|
||||
)
|
||||
|
||||
if (
|
||||
self.llm_correlation_adapter is not None
|
||||
and settings.llm_correlation_enabled
|
||||
and not strong_deterministic
|
||||
):
|
||||
try:
|
||||
llm_result = await self.llm_correlation_adapter.generate(
|
||||
envelope=envelope,
|
||||
decision=decision,
|
||||
recent_events=recent,
|
||||
)
|
||||
|
||||
llm_assessment = self._build_llm_correlation_assessment(
|
||||
llm_result=llm_result,
|
||||
decision=decision,
|
||||
event=envelope.event,
|
||||
)
|
||||
|
||||
if llm_assessment is not None:
|
||||
chosen_assessment = llm_assessment
|
||||
await audit(
|
||||
"llm_correlation",
|
||||
details={
|
||||
"role": llm_assessment.role,
|
||||
"kind": llm_assessment.kind,
|
||||
"reason": llm_assessment.reason,
|
||||
"confidence": llm_assessment.confidence,
|
||||
"parent_event_id": llm_assessment.parent_event_id,
|
||||
"parent_correlation_id": llm_assessment.parent_correlation_id,
|
||||
"suppress_child": llm_assessment.suppress_child,
|
||||
"scope_keys": scope_keys,
|
||||
},
|
||||
)
|
||||
else:
|
||||
await audit(
|
||||
"llm_correlation",
|
||||
status="skipped",
|
||||
details={
|
||||
"reason": llm_result.reason,
|
||||
"confidence": llm_result.confidence,
|
||||
"error": llm_result.error,
|
||||
"scope_keys": scope_keys,
|
||||
},
|
||||
)
|
||||
except Exception as exc:
|
||||
await audit(
|
||||
"llm_correlation",
|
||||
status="error",
|
||||
details={"error": str(exc), "scope_keys": scope_keys},
|
||||
)
|
||||
logger.exception(
|
||||
"LLM correlation fallback failed: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
exc,
|
||||
)
|
||||
|
||||
if chosen_assessment is not None:
|
||||
decision = apply_correlation_to_decision(
|
||||
decision=decision,
|
||||
assessment=chosen_assessment,
|
||||
)
|
||||
|
||||
await self.redis_repo.save_correlation_event(
|
||||
host=event.host or "",
|
||||
event_id=event.event_id,
|
||||
correlation_id=event.correlation_id,
|
||||
kind=chosen_assessment.kind,
|
||||
severity=decision.severity,
|
||||
routing_class=decision.routing_class,
|
||||
fingerprint=decision.fingerprint,
|
||||
root_candidate=chosen_assessment.root_cause_candidate,
|
||||
role=chosen_assessment.role,
|
||||
group_id=chosen_assessment.group_id,
|
||||
parent_event_id=chosen_assessment.parent_event_id,
|
||||
parent_correlation_id=chosen_assessment.parent_correlation_id,
|
||||
scope_keys=scope_keys,
|
||||
tags=event.tags,
|
||||
service=zbx_context.get("service") or event.service,
|
||||
scope=zbx_context.get("scope"),
|
||||
component=zbx_context.get("component"),
|
||||
domain=zbx_context.get("domain"),
|
||||
)
|
||||
|
||||
await audit(
|
||||
"correlation",
|
||||
details={
|
||||
"applied": chosen_assessment.applied,
|
||||
"role": chosen_assessment.role,
|
||||
"kind": chosen_assessment.kind,
|
||||
"group_id": chosen_assessment.group_id,
|
||||
"parent_event_id": chosen_assessment.parent_event_id,
|
||||
"parent_correlation_id": chosen_assessment.parent_correlation_id,
|
||||
"root_cause_candidate": chosen_assessment.root_cause_candidate,
|
||||
"correlated_event_count": chosen_assessment.correlated_event_count,
|
||||
"reason": chosen_assessment.reason,
|
||||
"suppress_child": chosen_assessment.suppress_child,
|
||||
"source": chosen_assessment.source,
|
||||
"confidence": chosen_assessment.confidence,
|
||||
"scope_keys": scope_keys,
|
||||
},
|
||||
)
|
||||
|
||||
except Exception as exc:
|
||||
await audit(
|
||||
"correlation",
|
||||
status="error",
|
||||
details={"error": str(exc)},
|
||||
)
|
||||
logger.exception(
|
||||
"Correlation baseline failed: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
exc,
|
||||
)
|
||||
|
||||
if (
|
||||
decision.event_phase == "problem"
|
||||
and decision.notify
|
||||
and not decision.suppressed
|
||||
and decision.routing_class in {
|
||||
"high_priority",
|
||||
"average_priority",
|
||||
"triage_low_priority_notify",
|
||||
}
|
||||
):
|
||||
await self.redis_repo.upsert_open_incident(
|
||||
fingerprint=fingerprint,
|
||||
event=event,
|
||||
decision=decision,
|
||||
)
|
||||
|
||||
if (
|
||||
self.llm_remediation_adapter is not None
|
||||
and decision.event_phase == "problem"
|
||||
and decision.notify
|
||||
and not decision.suppressed
|
||||
):
|
||||
try:
|
||||
remediation = await self.llm_remediation_adapter.generate(envelope, decision)
|
||||
if remediation.ok:
|
||||
decision = decision.model_copy(
|
||||
update={
|
||||
"llm_enriched": True,
|
||||
"remediation_summary": remediation.summary,
|
||||
"remediation_steps": remediation.steps,
|
||||
"remediation_commands": remediation.commands,
|
||||
}
|
||||
)
|
||||
await audit(
|
||||
"llm_remediation",
|
||||
details={
|
||||
"summary": remediation.summary,
|
||||
"steps_count": len(remediation.steps),
|
||||
"commands_count": len(remediation.commands),
|
||||
},
|
||||
)
|
||||
else:
|
||||
await audit(
|
||||
"llm_remediation",
|
||||
status="error",
|
||||
details={"error": remediation.error},
|
||||
)
|
||||
logger.warning(
|
||||
"LLM remediation returned no usable content: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
remediation.error,
|
||||
)
|
||||
except Exception as exc:
|
||||
await audit(
|
||||
"llm_remediation",
|
||||
status="error",
|
||||
details={"error": str(exc)},
|
||||
)
|
||||
logger.exception(
|
||||
"LLM remediation failed: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
exc,
|
||||
)
|
||||
|
||||
if self.audit_logger is not None:
|
||||
await self.audit_logger.log_decision(
|
||||
envelope=envelope,
|
||||
decision=decision,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
"Processed event: correlation_id=%s event_id=%s event_phase=%s severity=%s "
|
||||
"host=%s trigger=%s fingerprint=%s repeat_count=%s "
|
||||
"open_incident_found=%s flap_detected=%s flap_event_count=%s "
|
||||
"notify=%s suppressed=%s channels=%s reason=%s "
|
||||
"triage_applied=%s triage_source=%s triage_verdict=%s triage_classification=%s triage_reason=%s "
|
||||
"correlation_applied=%s correlation_role=%s correlation_kind=%s correlation_group_id=%s parent_event_id=%s root_cause_candidate=%s correlated_event_count=%s correlation_reason=%s correlation_source=%s correlation_confidence=%s "
|
||||
"suppress_reason=%s flap_reason=%s recovered_from_severity=%s "
|
||||
"event_url=%s graph_url=%s llm_enriched=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
decision.event_phase,
|
||||
event.severity,
|
||||
event.host,
|
||||
event.trigger_name,
|
||||
fingerprint,
|
||||
repeat_count,
|
||||
decision.open_incident_found,
|
||||
decision.flap_detected,
|
||||
decision.flap_event_count,
|
||||
decision.notify,
|
||||
decision.suppressed,
|
||||
",".join(decision.channels),
|
||||
decision.reason,
|
||||
decision.triage_applied,
|
||||
decision.triage_source,
|
||||
decision.triage_verdict,
|
||||
decision.triage_classification,
|
||||
decision.triage_reason,
|
||||
decision.correlation_applied,
|
||||
decision.correlation_role,
|
||||
decision.correlation_kind,
|
||||
decision.correlation_group_id,
|
||||
decision.parent_event_id,
|
||||
decision.root_cause_candidate,
|
||||
decision.correlated_event_count,
|
||||
decision.correlation_reason,
|
||||
decision.correlation_source,
|
||||
decision.correlation_confidence,
|
||||
decision.suppress_reason,
|
||||
decision.flap_reason,
|
||||
decision.recovered_from_severity,
|
||||
event.event_url,
|
||||
event.graph_url,
|
||||
decision.llm_enriched,
|
||||
)
|
||||
|
||||
dispatch_report = await self.notification_dispatcher.dispatch(
|
||||
envelope=envelope,
|
||||
decision=decision,
|
||||
)
|
||||
|
||||
delivery_payload = {
|
||||
"attempted": dispatch_report.attempted,
|
||||
"matrix_attempted": dispatch_report.matrix_attempted,
|
||||
"matrix_sent": dispatch_report.matrix_sent,
|
||||
"matrix_event_id": dispatch_report.matrix_event_id,
|
||||
"matrix_error": dispatch_report.matrix_error,
|
||||
"matrix_image_attempted": dispatch_report.matrix_image_attempted,
|
||||
"matrix_image_sent": dispatch_report.matrix_image_sent,
|
||||
"matrix_image_event_id": dispatch_report.matrix_image_event_id,
|
||||
"matrix_image_mxc_uri": dispatch_report.matrix_image_mxc_uri,
|
||||
"matrix_image_error": dispatch_report.matrix_image_error,
|
||||
"mail_attempted": dispatch_report.mail_attempted,
|
||||
"mail_sent": dispatch_report.mail_sent,
|
||||
"mail_error": dispatch_report.mail_error,
|
||||
"errors": dispatch_report.errors,
|
||||
}
|
||||
|
||||
if self.audit_logger is not None:
|
||||
await self.audit_logger.log_delivery(
|
||||
envelope=envelope,
|
||||
delivery_payload=delivery_payload,
|
||||
)
|
||||
|
||||
if dispatch_report.matrix_attempted:
|
||||
if dispatch_report.matrix_sent:
|
||||
logger.info(
|
||||
"Matrix notification sent: correlation_id=%s event_id=%s matrix_event_id=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
dispatch_report.matrix_event_id,
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
"Matrix notification failed: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
dispatch_report.matrix_error,
|
||||
)
|
||||
|
||||
if dispatch_report.matrix_image_attempted:
|
||||
if dispatch_report.matrix_image_sent:
|
||||
logger.info(
|
||||
"Matrix graph image sent: correlation_id=%s event_id=%s matrix_event_id=%s content_uri=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
dispatch_report.matrix_image_event_id,
|
||||
dispatch_report.matrix_image_mxc_uri,
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
"Matrix graph image failed: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
dispatch_report.matrix_image_error,
|
||||
)
|
||||
|
||||
if dispatch_report.mail_attempted:
|
||||
if dispatch_report.mail_sent:
|
||||
logger.info(
|
||||
"Mail notification sent: correlation_id=%s event_id=%s recipient=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
settings.mail_to,
|
||||
)
|
||||
else:
|
||||
logger.error(
|
||||
"Mail notification failed: correlation_id=%s event_id=%s error=%s",
|
||||
event.correlation_id,
|
||||
event.event_id,
|
||||
dispatch_report.mail_error,
|
||||
)
|
||||
|
||||
return decision
|
||||
|
||||
def _build_llm_correlation_assessment(
|
||||
self,
|
||||
llm_result: LLMCorrelationResult,
|
||||
decision: NotificationDecision,
|
||||
event,
|
||||
) -> CorrelationAssessment | None:
|
||||
if not llm_result.ok:
|
||||
return None
|
||||
|
||||
min_confidence_rank = _confidence_rank(settings.llm_correlation_min_confidence)
|
||||
actual_confidence_rank = _confidence_rank(llm_result.confidence)
|
||||
|
||||
role = llm_result.role
|
||||
confidence = llm_result.confidence or "low"
|
||||
|
||||
if role in {"root", "child"} and actual_confidence_rank < min_confidence_rank:
|
||||
return None
|
||||
|
||||
if role == "child" and not (llm_result.parent_event_id or llm_result.parent_correlation_id):
|
||||
return None
|
||||
|
||||
suppress_child = False
|
||||
if (
|
||||
role == "child"
|
||||
and llm_result.suppress_child
|
||||
and actual_confidence_rank >= _confidence_rank("high")
|
||||
and not _is_high_or_disaster(decision.severity)
|
||||
):
|
||||
suppress_child = True
|
||||
|
||||
if role == "root":
|
||||
group_id = event.event_id or event.correlation_id
|
||||
return CorrelationAssessment(
|
||||
applied=True,
|
||||
role="root",
|
||||
kind=llm_result.kind,
|
||||
group_id=group_id,
|
||||
reason=llm_result.reason,
|
||||
parent_event_id=None,
|
||||
parent_correlation_id=None,
|
||||
root_cause_candidate=True,
|
||||
correlated_event_count=0,
|
||||
suppress_child=False,
|
||||
source="llm",
|
||||
confidence=confidence,
|
||||
)
|
||||
|
||||
if role == "child":
|
||||
group_id = llm_result.parent_event_id or llm_result.parent_correlation_id
|
||||
return CorrelationAssessment(
|
||||
applied=True,
|
||||
role="child",
|
||||
kind=llm_result.kind,
|
||||
group_id=group_id,
|
||||
reason=llm_result.reason,
|
||||
parent_event_id=llm_result.parent_event_id,
|
||||
parent_correlation_id=llm_result.parent_correlation_id,
|
||||
root_cause_candidate=False,
|
||||
correlated_event_count=0,
|
||||
suppress_child=suppress_child,
|
||||
source="llm",
|
||||
confidence=confidence,
|
||||
)
|
||||
|
||||
return None
|
||||
Reference in New Issue
Block a user