diff --git a/core/main.py b/core/main.py index 4449588..10a46ef 100644 --- a/core/main.py +++ b/core/main.py @@ -27,7 +27,8 @@ async def lifespan(app: FastAPI): @app.post("/alert") def receive_alert(payload: dict, background_tasks: BackgroundTasks) -> dict: incident = orchestrator.handle_alert(payload) - background_tasks.add_task(orchestrator.run_diagnostics, incident["incident_id"], payload) + if not incident.get("suppressed"): + background_tasks.add_task(orchestrator.run_diagnostics, incident["incident_id"], payload) return incident diff --git a/core/orchestrator.py b/core/orchestrator.py index 27917a0..e0429cf 100644 --- a/core/orchestrator.py +++ b/core/orchestrator.py @@ -1,5 +1,6 @@ import os import time +from collections import defaultdict from datetime import datetime, timezone from . import db from .services import git_client, llm_analyzer, vector_store, notifier, postmortem @@ -7,17 +8,54 @@ REPO_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # Sentinel root RUNBOOKS_DIR = os.path.join(REPO_PATH, "sandbox", "runbooks") +# ponytail: in-process sliding-window dedup; per-worker, resets on restart. +# Move to Postgres/Redis if you run multiple workers or need durability. +_DEDUP_WINDOW_S = float(os.getenv("SENTINEL_DEDUP_WINDOW_S", "300")) +_DEDUP_THRESHOLD = int(os.getenv("SENTINEL_DEDUP_THRESHOLD", "1")) # Nth hit in window fires +_recent_alerts: dict[str, list[float]] = defaultdict(list) # signature -> hit times +_open_signatures: dict[str, str] = {} # signature -> live incident_id + + +def _dedup_gate(signature: str) -> tuple[str, str | None]: + """'fire' (open a new incident), 'duplicate' (fold into an active one), + or 'suppress' (below the burst threshold).""" + now = time.monotonic() + hits = _recent_alerts[signature] + hits[:] = [t for t in hits if t >= now - _DEDUP_WINDOW_S] + hits.append(now) + + if signature in _open_signatures: + return "duplicate", _open_signatures[signature] + if len(hits) < _DEDUP_THRESHOLD: + return "suppress", None + return "fire", None + + +def _clear_signature(signature: str): + _open_signatures.pop(signature, None) + def handle_alert(alert_data: dict) -> dict: + signature = alert_data.get("error_signature", "") + decision, existing_id = _dedup_gate(signature) + if decision == "duplicate": + print(f"[dedup] alert folded into active incident {existing_id}") + return {"incident_id": existing_id, "status": "duplicate", "suppressed": True} + if decision == "suppress": + seen = len(_recent_alerts[signature]) + print(f"[dedup] alert suppressed ({seen}/{_DEDUP_THRESHOLD} in {int(_DEDUP_WINDOW_S)}s window)") + return {"incident_id": None, "status": "suppressed", "suppressed": True} + ts = datetime.now(timezone.utc).strftime("%Y_%m%d_%H%M%S") incident = { - "incident_id": f"inc_{ts}", + "incident_id": f"inc_{ts}_{os.urandom(2).hex()}", # suffix: same-second alerts must not collide on PK "status": "triggered", "trigger": alert_data, "diagnostics": None, "postmortem_draft_url": None, } db.save_incident(incident) + _open_signatures[signature] = incident["incident_id"] print(f"[orchestrator] {incident['incident_id']} created -> triggered") return incident @@ -61,6 +99,7 @@ def run_diagnostics(incident_id: str, alert_data: dict): db.update_diagnostics(incident_id, diagnostics) if degraded_reason: db.update_status(incident_id, "degraded") + _clear_signature(alert_data.get("error_signature", "")) # terminal: stop folding new alerts print(f"[orchestrator] {incident_id} -> degraded (terminal)") else: alert_dt = datetime.fromisoformat(alert_data["timestamp"].replace("Z", "+00:00")) @@ -76,6 +115,7 @@ def run_diagnostics(incident_id: str, alert_data: dict): except Exception as e: print(f"[orchestrator] {incident_id} diagnostics failed: {e}") db.update_status(incident_id, "triggered") # revert so it can be retried + _clear_signature(alert_data.get("error_signature", "")) # allow a retry alert through def resolve_incident(incident_id: str) -> dict: @@ -92,6 +132,7 @@ def resolve_incident(incident_id: str) -> dict: ) print(f"[orchestrator] {incident_id} -- postmortem LLM unavailable, degrading: {e}") db.update_postmortem(incident_id, draft) + _clear_signature((incident.get("trigger_data") or {}).get("error_signature", "")) # resolved: reopen to new alerts print(f"[orchestrator] {incident_id} -> resolved, postmortem generated") print(f"[metrics] {incident_id} resolve_to_postmortem_s={time.monotonic() - t0:.1f}") return db.get_incident(incident_id) diff --git a/core/tests/test_orchestrator.py b/core/tests/test_orchestrator.py index 1cce2b9..646fb07 100644 --- a/core/tests/test_orchestrator.py +++ b/core/tests/test_orchestrator.py @@ -1,5 +1,6 @@ # ponytail: mocks db so test runs without a live postgres connection from unittest.mock import patch +import core.orchestrator as orch from core.orchestrator import handle_alert _ALERT = { @@ -10,7 +11,13 @@ } +def _reset_dedup(): + orch._recent_alerts.clear() + orch._open_signatures.clear() + + def test_handle_alert(): + _reset_dedup() with patch("core.orchestrator.db.save_incident") as mock_save: result = handle_alert(_ALERT) assert result["status"] == "triggered" @@ -20,6 +27,41 @@ def test_handle_alert(): mock_save.assert_called_once() +def test_dedup_folds_repeat_alerts(): + """Second alert for a still-open signature must not open a new incident.""" + _reset_dedup() + with patch("core.orchestrator.db.save_incident") as mock_save: + first = handle_alert(_ALERT) + second = handle_alert(_ALERT) + assert first["status"] == "triggered" + assert second["status"] == "duplicate" + assert second["incident_id"] == first["incident_id"] + mock_save.assert_called_once() # only the first alert hit the DB + + # once the incident is cleared (resolved/degraded/failed), a new alert fires again + orch._clear_signature(_ALERT["error_signature"]) + with patch("core.orchestrator.db.save_incident"): + third = handle_alert(_ALERT) + assert third["status"] == "triggered" + assert third["incident_id"] != first["incident_id"] + + +def test_dedup_threshold_suppresses_below_burst(): + """With a threshold >1, alerts below the burst count are suppressed, not opened.""" + _reset_dedup() + with patch.object(orch, "_DEDUP_THRESHOLD", 3), \ + patch("core.orchestrator.db.save_incident") as mock_save: + r1 = handle_alert(_ALERT) + r2 = handle_alert(_ALERT) + r3 = handle_alert(_ALERT) + assert r1["status"] == "suppressed" + assert r2["status"] == "suppressed" + assert r3["status"] == "triggered" # 3rd hit crosses the threshold + mock_save.assert_called_once() + + if __name__ == "__main__": test_handle_alert() - print("✓ orchestrator test passed") + test_dedup_folds_repeat_alerts() + test_dedup_threshold_suppresses_below_burst() + print("[ok] orchestrator + dedup tests passed")