75 lines
4.3 KiB
Python
75 lines
4.3 KiB
Python
"""Structured operational events with defensive redaction and retention."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import delete, select
|
|
|
|
from .config import settings
|
|
from .models import AgentEvent
|
|
|
|
EVENT_TYPES = {
|
|
"AGENT_STARTED", "AGENT_STOPPED", "HEARTBEAT", "PREFLIGHT_PASS", "PREFLIGHT_FAIL", "JOB_CLAIMED", "JOB_STARTED",
|
|
"JOB_COMPLETED", "JOB_FAILED", "PRODUCT_BLOCKED", "BLOCKER_RESOLVED", "DATASET_CHANGED", "COMPANY_CONTEXT_CHANGED",
|
|
"SCHEMA_CHANGED", "WRITE_PLANNED", "WRITE_COMPLETED", "WRITE_FAILED", "CONNECTION_ERROR", "RETRY", "VERSION_CHANGED",
|
|
"ADMIN_ACTION",
|
|
}
|
|
SEVERITIES = {"INFO", "WARNING", "ERROR", "CRITICAL"}
|
|
SECRET_RE = re.compile(r"(?i)(password|fb_password|token|api[_-]?key|secret|authorization|connection[_-]?string)\s*[:=]\s*[^,;\s]+")
|
|
|
|
|
|
def _redact(value):
|
|
if isinstance(value, dict):
|
|
return {str(key): "[REDACTED]" if str(key).lower() in {"password", "fb_password", "token", "api_key", "secret", "authorization", "connection_string"} else _redact(item) for key, item in value.items()}
|
|
if isinstance(value, list):
|
|
return [_redact(item) for item in value[:100]]
|
|
if isinstance(value, str):
|
|
return SECRET_RE.sub(lambda match: f"{match.group(1)}=[REDACTED]", value)
|
|
return value
|
|
|
|
|
|
def sanitize_event(message: str, context: dict | None = None) -> tuple[str, dict]:
|
|
safe_message = _redact(str(message or ""))[: settings.web_max_event_message]
|
|
safe_context = _redact(context or {})
|
|
encoded = json.dumps(safe_context, ensure_ascii=False, default=str)
|
|
if len(encoded.encode()) > settings.web_max_event_context_bytes:
|
|
encoded = encoded.encode()[: settings.web_max_event_context_bytes].decode("utf-8", errors="ignore")
|
|
safe_context = {"truncated": True, "payload": encoded}
|
|
return safe_message, safe_context
|
|
|
|
|
|
def record_event(session, *, tenant_id: int, installation_id: int, agent_id: int, severity: str, event_type: str, message: str, context: dict | None = None, job_id: int | None = None, product_code: str | None = None, blocker_id: int | None = None, is_audit: bool = False, occurred_at: datetime | None = None):
|
|
severity = severity.upper()
|
|
event_type = event_type.upper()
|
|
if severity not in SEVERITIES:
|
|
raise ValueError("invalid event severity")
|
|
if event_type not in EVENT_TYPES:
|
|
raise ValueError("invalid event type")
|
|
safe_message, safe_context = sanitize_event(message, context)
|
|
event = AgentEvent(tenant_id=tenant_id, installation_id=installation_id, agent_id=agent_id, severity=severity, event_type=event_type, message=safe_message, structured_context=safe_context, job_id=job_id, product_code=product_code, blocker_id=blocker_id, is_audit=is_audit, occurred_at=occurred_at or datetime.now(timezone.utc))
|
|
session.add(event)
|
|
return event
|
|
|
|
|
|
def purge_expired_events(session, now: datetime | None = None, retention_days: int | None = None) -> int:
|
|
cutoff = (now or datetime.now(timezone.utc)) - timedelta(days=retention_days if retention_days is not None else settings.agent_event_retention_days)
|
|
result = session.execute(delete(AgentEvent).where(AgentEvent.occurred_at < cutoff, AgentEvent.is_audit.is_(False), AgentEvent.severity != "CRITICAL"))
|
|
session.commit()
|
|
return result.rowcount or 0
|
|
|
|
|
|
def list_events(session, installation_db_id: int, *, page: int = 1, page_size: int = 50, severity: str | None = None, event_type: str | None = None, job_id: int | None = None, product_code: str | None = None, since: datetime | None = None, until: datetime | None = None, fetch_extra: bool = False):
|
|
page = max(1, page); page_size = min(100, max(1, page_size))
|
|
query = select(AgentEvent).where(AgentEvent.installation_id == installation_db_id)
|
|
for column, value in ((AgentEvent.severity, severity), (AgentEvent.event_type, event_type), (AgentEvent.job_id, job_id), (AgentEvent.product_code, product_code)):
|
|
if value is not None:
|
|
query = query.where(column == value)
|
|
if since: query = query.where(AgentEvent.occurred_at >= since)
|
|
if until: query = query.where(AgentEvent.occurred_at <= until)
|
|
query = query.order_by(AgentEvent.occurred_at.desc(), AgentEvent.id.desc())
|
|
rows = session.scalars(query.offset((page - 1) * page_size).limit(page_size + int(fetch_extra))).all()
|
|
return rows
|