Загрузить файлы в «alert-processor/app»
This commit is contained in:
@@ -0,0 +1,330 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
from redis.asyncio import Redis
|
||||
|
||||
from app.models import NotificationDecision, ProcessorForwardEnvelope
|
||||
|
||||
|
||||
def _utc_now() -> str:
|
||||
return datetime.now(timezone.utc).isoformat()
|
||||
|
||||
|
||||
class AuditLogger:
|
||||
def __init__(
|
||||
self,
|
||||
client: Redis,
|
||||
key_prefix: str,
|
||||
ttl_seconds: int = 604800,
|
||||
max_stage_records: int = 200,
|
||||
) -> None:
|
||||
self.client = client
|
||||
self.key_prefix = key_prefix
|
||||
self.ttl_seconds = ttl_seconds
|
||||
self.max_stage_records = max_stage_records
|
||||
|
||||
def _journal_key(self, correlation_id: str) -> str:
|
||||
return f"{self.key_prefix}:journal:{correlation_id}"
|
||||
|
||||
def _stages_key(self, correlation_id: str) -> str:
|
||||
return f"{self.key_prefix}:stages:{correlation_id}"
|
||||
|
||||
def _event_index_key(self, event_id: str) -> str:
|
||||
return f"{self.key_prefix}:event:{event_id}"
|
||||
|
||||
def _recent_key(self) -> str:
|
||||
return f"{self.key_prefix}:recent"
|
||||
|
||||
async def log_ingest_queued(
|
||||
self,
|
||||
envelope: ProcessorForwardEnvelope,
|
||||
job_id: str,
|
||||
queue_name: str,
|
||||
) -> None:
|
||||
event = envelope.event
|
||||
correlation_id = event.correlation_id
|
||||
event_id = event.event_id or ""
|
||||
|
||||
await self._upsert_journal(
|
||||
correlation_id=correlation_id,
|
||||
event_id=event_id or None,
|
||||
fields={
|
||||
"state": "queued",
|
||||
"job_id": job_id,
|
||||
"queue_name": queue_name,
|
||||
"correlation_id": correlation_id,
|
||||
"event_id": event_id,
|
||||
"host": event.host or "",
|
||||
"service": event.service or "",
|
||||
"trigger_name": event.trigger_name or "",
|
||||
"severity": event.severity or "",
|
||||
"queued_at": _utc_now(),
|
||||
"updated_at": _utc_now(),
|
||||
},
|
||||
)
|
||||
|
||||
await self.log_stage(
|
||||
correlation_id=correlation_id,
|
||||
event_id=event.event_id,
|
||||
stage="ingest_queued",
|
||||
status="ok",
|
||||
details={
|
||||
"job_id": job_id,
|
||||
"queue_name": queue_name,
|
||||
},
|
||||
)
|
||||
|
||||
async def log_worker_started(
|
||||
self,
|
||||
envelope: ProcessorForwardEnvelope,
|
||||
job_id: str,
|
||||
attempt: int,
|
||||
identity: str,
|
||||
) -> None:
|
||||
event = envelope.event
|
||||
await self._upsert_journal(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
fields={
|
||||
"state": "processing",
|
||||
"worker_started_at": _utc_now(),
|
||||
"worker_attempt": attempt,
|
||||
"processing_identity": identity,
|
||||
"updated_at": _utc_now(),
|
||||
},
|
||||
)
|
||||
|
||||
await self.log_stage(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
stage="worker_started",
|
||||
status="ok",
|
||||
details={
|
||||
"job_id": job_id,
|
||||
"attempt": attempt,
|
||||
"identity": identity,
|
||||
},
|
||||
)
|
||||
|
||||
async def log_stage(
|
||||
self,
|
||||
correlation_id: str,
|
||||
event_id: str | None,
|
||||
stage: str,
|
||||
status: str = "ok",
|
||||
details: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
payload = {
|
||||
"ts": _utc_now(),
|
||||
"stage": stage,
|
||||
"status": status,
|
||||
"details": details or {},
|
||||
}
|
||||
raw = json.dumps(payload, ensure_ascii=False)
|
||||
|
||||
stages_key = self._stages_key(correlation_id)
|
||||
await self.client.rpush(stages_key, raw)
|
||||
await self.client.ltrim(stages_key, -self.max_stage_records, -1)
|
||||
await self.client.expire(stages_key, self.ttl_seconds)
|
||||
|
||||
await self._upsert_journal(
|
||||
correlation_id=correlation_id,
|
||||
event_id=event_id,
|
||||
fields={
|
||||
"last_stage": stage,
|
||||
"last_stage_status": status,
|
||||
"updated_at": _utc_now(),
|
||||
},
|
||||
)
|
||||
|
||||
async def log_decision(
|
||||
self,
|
||||
envelope: ProcessorForwardEnvelope,
|
||||
decision: NotificationDecision,
|
||||
) -> None:
|
||||
event = envelope.event
|
||||
await self._upsert_journal(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
fields={
|
||||
"state": "decision_made",
|
||||
"decision_at": _utc_now(),
|
||||
"notify": str(decision.notify).lower(),
|
||||
"suppressed": str(decision.suppressed).lower(),
|
||||
"routing_class": decision.routing_class or "",
|
||||
"decision_reason": decision.reason or "",
|
||||
"decision_json": json.dumps(
|
||||
decision.model_dump(mode="json"),
|
||||
ensure_ascii=False,
|
||||
),
|
||||
"updated_at": _utc_now(),
|
||||
},
|
||||
)
|
||||
|
||||
await self.log_stage(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
stage="decision_made",
|
||||
status="ok",
|
||||
details={
|
||||
"notify": decision.notify,
|
||||
"suppressed": decision.suppressed,
|
||||
"routing_class": decision.routing_class,
|
||||
"reason": decision.reason,
|
||||
},
|
||||
)
|
||||
|
||||
async def log_delivery(
|
||||
self,
|
||||
envelope: ProcessorForwardEnvelope,
|
||||
delivery_payload: dict[str, Any],
|
||||
) -> None:
|
||||
event = envelope.event
|
||||
|
||||
await self._upsert_journal(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
fields={
|
||||
"state": "delivered",
|
||||
"delivery_at": _utc_now(),
|
||||
"delivery_json": json.dumps(delivery_payload, ensure_ascii=False),
|
||||
"updated_at": _utc_now(),
|
||||
},
|
||||
)
|
||||
|
||||
await self.log_stage(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
stage="delivery_completed",
|
||||
status="ok",
|
||||
details=delivery_payload,
|
||||
)
|
||||
|
||||
async def log_worker_outcome(
|
||||
self,
|
||||
envelope: ProcessorForwardEnvelope,
|
||||
state: str,
|
||||
error: str | None = None,
|
||||
attempt: int | None = None,
|
||||
) -> None:
|
||||
event = envelope.event
|
||||
|
||||
fields = {
|
||||
"state": state,
|
||||
"updated_at": _utc_now(),
|
||||
}
|
||||
if error:
|
||||
fields["last_error"] = error
|
||||
fields["last_error_at"] = _utc_now()
|
||||
if attempt is not None:
|
||||
fields["worker_attempt"] = attempt
|
||||
|
||||
await self._upsert_journal(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
fields=fields,
|
||||
)
|
||||
|
||||
await self.log_stage(
|
||||
correlation_id=event.correlation_id,
|
||||
event_id=event.event_id,
|
||||
stage=f"worker_{state}",
|
||||
status="error" if error else "ok",
|
||||
details={
|
||||
"attempt": attempt,
|
||||
"error": error,
|
||||
},
|
||||
)
|
||||
|
||||
async def get_event_audit(
|
||||
self,
|
||||
correlation_id: str | None = None,
|
||||
event_id: str | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
resolved_correlation = correlation_id
|
||||
|
||||
if not resolved_correlation and event_id:
|
||||
value = await self.client.get(self._event_index_key(event_id))
|
||||
if value is None:
|
||||
return None
|
||||
resolved_correlation = value.decode() if isinstance(value, bytes) else str(value)
|
||||
|
||||
if not resolved_correlation:
|
||||
return None
|
||||
|
||||
journal_raw = await self.client.hgetall(self._journal_key(resolved_correlation))
|
||||
if not journal_raw:
|
||||
return None
|
||||
|
||||
stages_raw = await self.client.lrange(self._stages_key(resolved_correlation), 0, -1)
|
||||
|
||||
journal = self._decode_hash(journal_raw)
|
||||
stages = [json.loads(item.decode() if isinstance(item, bytes) else str(item)) for item in stages_raw]
|
||||
|
||||
if journal.get("decision_json"):
|
||||
try:
|
||||
journal["decision"] = json.loads(journal["decision_json"])
|
||||
except Exception:
|
||||
journal["decision"] = None
|
||||
|
||||
if journal.get("delivery_json"):
|
||||
try:
|
||||
journal["delivery"] = json.loads(journal["delivery_json"])
|
||||
except Exception:
|
||||
journal["delivery"] = None
|
||||
|
||||
return {
|
||||
"journal": journal,
|
||||
"stages": stages,
|
||||
}
|
||||
|
||||
async def list_recent(self, limit: int = 20) -> list[dict[str, Any]]:
|
||||
correlation_ids_raw = await self.client.zrevrange(self._recent_key(), 0, max(limit - 1, 0))
|
||||
results: list[dict[str, Any]] = []
|
||||
|
||||
for raw in correlation_ids_raw:
|
||||
correlation_id = raw.decode() if isinstance(raw, bytes) else str(raw)
|
||||
audit = await self.get_event_audit(correlation_id=correlation_id)
|
||||
if audit is not None:
|
||||
results.append(audit)
|
||||
|
||||
return results
|
||||
|
||||
async def _upsert_journal(
|
||||
self,
|
||||
correlation_id: str,
|
||||
event_id: str | None,
|
||||
fields: dict[str, Any],
|
||||
) -> None:
|
||||
journal_key = self._journal_key(correlation_id)
|
||||
|
||||
mapping: dict[str, str] = {}
|
||||
for key, value in fields.items():
|
||||
if value is None:
|
||||
continue
|
||||
mapping[key] = str(value)
|
||||
|
||||
await self.client.hset(journal_key, mapping=mapping)
|
||||
await self.client.expire(journal_key, self.ttl_seconds)
|
||||
|
||||
if event_id:
|
||||
await self.client.set(
|
||||
self._event_index_key(event_id),
|
||||
correlation_id,
|
||||
ex=self.ttl_seconds,
|
||||
)
|
||||
|
||||
await self.client.zadd(self._recent_key(), {correlation_id: datetime.now(timezone.utc).timestamp()})
|
||||
await self.client.expire(self._recent_key(), self.ttl_seconds)
|
||||
|
||||
@staticmethod
|
||||
def _decode_hash(raw: dict[bytes, bytes]) -> dict[str, str]:
|
||||
result: dict[str, str] = {}
|
||||
for k, v in raw.items():
|
||||
key = k.decode() if isinstance(k, bytes) else str(k)
|
||||
value = v.decode() if isinstance(v, bytes) else str(v)
|
||||
result[key] = value
|
||||
return result
|
||||
Reference in New Issue
Block a user