"""Alertmanager service definition.""" from __future__ import annotations from typing import Any from media_library_viewer_api.integrations.base import ( SecretField, ServiceBaseUrl, ServiceConfigBase, ServiceDefinition, WidgetConfigBase, widget_kind, ) class AlertmanagerConfig(ServiceConfigBase): """Non-secret Alertmanager connection config.""" base_url: ServiceBaseUrl timeout_seconds: int = 5 class AlertmanagerAlertsWidgetConfig(WidgetConfigBase): """Active-alerts summary for an Alertmanager instance.""" severity_filter: str | None = None def summarize_alerts( alerts: list[dict[str, Any]], *, severity_filter: str | None = None, ) -> dict[str, Any]: """Build a UI-friendly summary from an Alertmanager ``/api/v1/alerts`` list. Reshapes the raw alert objects into a stable summary (``total``, ``by_severity``, top-50 ``alerts``). When ``severity_filter`` is given, only alerts whose ``labels.severity`` matches are counted. """ by_severity: dict[str, int] = {} open_alerts: list[dict[str, Any]] = [] for alert in alerts: labels = alert.get("labels") or {} annotations = alert.get("annotations") or {} severity = labels.get("severity", "unknown") if severity_filter and severity != severity_filter: continue by_severity[severity] = by_severity.get(severity, 0) + 1 open_alerts.append( { "name": labels.get("alertname", "unknown"), "severity": severity, "category": labels.get("category", ""), "job_name": labels.get("job_name", labels.get("job", "")), "summary": annotations.get("summary", ""), "description": annotations.get("description", ""), "active_since": alert.get("startsAt"), "state": alert.get("status", "firing"), "labels": labels, } ) open_alerts.sort(key=lambda a: (a["severity"] not in {"critical", "warning"}, a["severity"], a["name"])) return { "total": len(open_alerts), "by_severity": by_severity, "alerts": open_alerts[:50], } DEFINITION = ServiceDefinition( service_type="alertmanager", name="Alertmanager", description="Alertmanager alerts and status.", config_model=AlertmanagerConfig, secret_fields=[ SecretField(key="api_key", label="API key", helper="Optional bearer token"), ], widget_kinds=[ widget_kind( kind="active_alerts", name="Active alerts", description="Firing alerts summary from Alertmanager.", model_cls=AlertmanagerAlertsWidgetConfig, default_config={}, refresh_interval_ms=30_000, ), ], )