| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| import time |
| from dataclasses import dataclass |
| from typing import Any, Dict, Iterable, Optional |
|
|
| import structlog |
|
|
| logger = structlog.get_logger("webhook_alerts") |
|
|
| try: |
| from app.utils.webhook_alerts import WebhookDispatcher |
| except ImportError: |
| WebhookDispatcher = None |
|
|
| |
| |
| _CONCERNING = {"risky", "unknown", "error"} |
|
|
|
|
| @dataclass |
| class RateAlert: |
| """Computed summary of one finished run, plus whether it breached.""" |
|
|
| total: int |
| risky: int |
| unknown: int |
| error: int |
| risky_rate: float |
| threshold: float |
| breached: bool |
|
|
|
|
| def check_risky_rate(rows: Iterable[Any], threshold: float = 0.5) -> Optional[RateAlert]: |
| """Compute the risky/unknown/error share of a run's rows. |
| |
| Rows may be dicts (model_dump() output in the Celery/sync paths) or objects |
| exposing ``.status`` (ValidationResult). Rows without a status are skipped. |
| Returns None when the monitor is disabled (threshold <= 0), when the |
| threshold is out of range (> 1), or when there is nothing to evaluate. |
| """ |
| if threshold <= 0 or threshold > 1: |
| return None |
|
|
| counts = {"risky": 0, "unknown": 0, "error": 0} |
| total = 0 |
| for row in rows: |
| status = row.get("status") if isinstance(row, dict) else getattr(row, "status", None) |
| if status is None: |
| continue |
| total += 1 |
| key = str(status).lower() |
| if key in _CONCERNING: |
| counts[key] += 1 |
|
|
| if total == 0: |
| return None |
|
|
| concerning = counts["risky"] + counts["unknown"] + counts["error"] |
| rate = concerning / total |
| return RateAlert( |
| total=total, |
| risky=counts["risky"], |
| unknown=counts["unknown"], |
| error=counts["error"], |
| risky_rate=rate, |
| threshold=threshold, |
| breached=rate >= threshold, |
| ) |
|
|
|
|
| def get_system_alert_template(metric_name: str, threshold: float, current_value: float, |
| run_summary: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: |
| """System-alert payload for a breached run-quality threshold. |
| |
| Mirrors the shape of webhook_alerts.get_system_alert_template and adds the |
| per-run breakdown so the receiver can diagnose without re-querying. |
| """ |
| payload = { |
| "sys_alert_id": f"sys_warn_{int(time.time())}", |
| "severity": "WARNING", |
| "system_status": "DEGRADED", |
| "metric_violated": metric_name, |
| "alert_conditions": { |
| "threshold_limit": threshold, |
| "active_value": current_value, |
| "deviation_percentage": ( |
| round(((current_value - threshold) / threshold * 100), 1) |
| if threshold > 0 else 100.0 |
| ), |
| }, |
| "triggered_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), |
| } |
| if run_summary is not None: |
| payload["run_summary"] = run_summary |
| return payload |
|
|
|
|
| async def maybe_alert_risky_rate( |
| rows: Iterable[Any], |
| webhook_url: Optional[str] = None, |
| job_id: str = "sync", |
| threshold: float = 0.5, |
| ) -> Optional[bool]: |
| """Check a finished run and fire the signed webhook when it breached. |
| |
| Returns True (dispatched), False (breached but delivery failed), None |
| (healthy or monitor disabled). Never raises: an alerting failure must not |
| turn a completed validation run into a failed one. |
| """ |
| try: |
| alert = check_risky_rate(rows, threshold=threshold) |
| except Exception as exc: |
| logger.warning("rate_check_failed", error=str(exc)) |
| return None |
|
|
| if alert is None or not alert.breached: |
| return None |
|
|
| run_summary = { |
| "job_id": job_id, |
| "total": alert.total, |
| "risky": alert.risky, |
| "unknown": alert.unknown, |
| "error": alert.error, |
| "risky_rate": round(alert.risky_rate, 4), |
| "threshold": alert.threshold, |
| } |
| payload = get_system_alert_template( |
| "bulk_risky_rate", alert.threshold, alert.risky_rate, run_summary=run_summary |
| ) |
|
|
| if not webhook_url or WebhookDispatcher is None: |
| logger.warning( |
| "bulk_risky_rate_breached_no_webhook", |
| **run_summary, |
| ) |
| return None |
|
|
| try: |
| ok = await WebhookDispatcher.dispatch( |
| webhook_url, "SYSTEM_ALERT", payload |
| ) |
| return bool(ok) |
| except Exception as exc: |
| logger.error("bulk_risky_rate_dispatch_failed", error=str(exc)) |
| return False |
|
|
