From 4d81a50d7ca4c1ae514066ce71e699e6dc387820 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Tue, 17 Mar 2026 19:46:18 -0400 Subject: [PATCH] feat: record_metric() and alert state machine --- fablednetmon/core/alerts.py | 240 ++++++++++++++++++++++++++++++++++++ 1 file changed, 240 insertions(+) create mode 100644 fablednetmon/core/alerts.py diff --git a/fablednetmon/core/alerts.py b/fablednetmon/core/alerts.py new file mode 100644 index 0000000..68784e9 --- /dev/null +++ b/fablednetmon/core/alerts.py @@ -0,0 +1,240 @@ +from __future__ import annotations +import asyncio +import logging +import uuid +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from fablednetmon.models.alerts import ( + AlertEvent, + AlertOperator, + AlertRule, + AlertState, + AlertStateEnum, +) +from fablednetmon.models.metrics import PluginMetric + +if TYPE_CHECKING: + from quart import Quart + +logger = logging.getLogger(__name__) + +_app: "Quart | None" = None + + +def init_alerts(app: "Quart") -> None: + """Store app reference for use in notification dispatch tasks.""" + global _app + _app = app + + +async def record_metric( + session: AsyncSession, + source_module: str, + resource_name: str, + metric_name: str, + value: float, +) -> None: + """Write a metric row and evaluate alert rules inline. + + Must be called inside an active transaction: + async with session.begin(): + await record_metric(session, ...) + + Notification I/O is deferred via asyncio.create_task() so no network + calls occur while the transaction is open. + """ + now = datetime.now(timezone.utc) + + # Write metric row + metric = PluginMetric( + source_module=source_module, + resource_name=resource_name, + metric_name=metric_name, + value=value, + recorded_at=now, + ) + session.add(metric) + await session.flush() + + # Load matching enabled rules + result = await session.execute( + select(AlertRule).where( + AlertRule.source_module == source_module, + AlertRule.resource_name == resource_name, + AlertRule.metric_name == metric_name, + AlertRule.enabled.is_(True), + ) + ) + rules = result.scalars().all() + + for rule in rules: + state_result = await session.execute( + select(AlertState).where(AlertState.rule_id == rule.id) + ) + alert_state = state_result.scalar_one_or_none() + if alert_state is None: + logger.warning(f"No alert_state row for rule {rule.id}, skipping") + continue + + notifications = await _evaluate_rule(session, rule, alert_state, value, now) + + for notif_type, event_id in notifications: + if _app is not None: + asyncio.create_task( + _dispatch_notification( + _app, notif_type, rule, value, event_id + ) + ) + + +def _is_breached(value: float, operator: AlertOperator, threshold: float) -> bool: + op = operator.value + if op == ">": + return value > threshold + if op == "<": + return value < threshold + if op == ">=": + return value >= threshold + if op == "<=": + return value <= threshold + if op == "==": + return value == threshold + if op == "!=": + return value != threshold + return False + + +async def _evaluate_rule( + session: AsyncSession, + rule: AlertRule, + state: AlertState, + value: float, + now: datetime, +) -> list[tuple[str, str]]: + """Evaluate a single rule against a metric value. Returns list of (notif_type, event_id).""" + breached = _is_breached(value, rule.operator, rule.threshold) + from_state = state.state + notifications: list[tuple[str, str]] = [] + + if state.state == AlertStateEnum.inactive: + if breached: + state.consecutive_failure_count = 1 + if state.consecutive_failure_count >= rule.consecutive_failures_required: + state.state = AlertStateEnum.firing + state.last_transition_at = now + event_id = _add_event(session, rule.id, from_state.value, "firing", value, now) + notifications.append(("firing", event_id)) + else: + state.state = AlertStateEnum.pending + state.last_transition_at = now + + elif state.state == AlertStateEnum.pending: + if breached: + state.consecutive_failure_count += 1 + if state.consecutive_failure_count >= rule.consecutive_failures_required: + state.state = AlertStateEnum.firing + state.last_transition_at = now + event_id = _add_event(session, rule.id, from_state.value, "firing", value, now) + notifications.append(("firing", event_id)) + else: + state.consecutive_failure_count = 0 + state.state = AlertStateEnum.inactive + state.last_transition_at = now + + elif state.state == AlertStateEnum.firing: + if not breached: + # RESOLVED is transient: record event, notify, then immediately go inactive + event_id = _add_event(session, rule.id, from_state.value, "resolved", value, now) + notifications.append(("resolved", event_id)) + state.consecutive_failure_count = 0 + state.state = AlertStateEnum.inactive + state.last_transition_at = now + + elif state.state == AlertStateEnum.acknowledged: + if not breached: + event_id = _add_event(session, rule.id, from_state.value, "resolved", value, now) + notifications.append(("resolved", event_id)) + state.consecutive_failure_count = 0 + state.state = AlertStateEnum.inactive + state.last_transition_at = now + state.acknowledged_by = None + state.acknowledged_at = None + else: + state.consecutive_failure_count += 1 + if state.consecutive_failure_count >= rule.consecutive_failures_required: + state.state = AlertStateEnum.firing + state.last_transition_at = now + event_id = _add_event(session, rule.id, from_state.value, "firing", value, now) + notifications.append(("firing", event_id)) + + await session.flush() + return notifications + + +def _add_event( + session: AsyncSession, + rule_id: str, + from_state: str, + to_state: str, + value: float, + now: datetime, +) -> str: + """Add an AlertEvent row and return its ID.""" + event_id = str(uuid.uuid4()) + event = AlertEvent( + id=event_id, + rule_id=rule_id, + from_state=from_state, + to_state=to_state, + metric_value=value, + transitioned_at=now, + ) + session.add(event) + return event_id + + +async def _dispatch_notification( + app: "Quart", + notif_type: str, + rule: AlertRule, + value: float, + event_id: str, +) -> None: + """Run after the transaction commits. Sends notifications and updates the event row.""" + from fablednetmon.core.notifications import dispatch_notifications + + alert_data = { + "rule_name": rule.name, + "state": notif_type.upper(), + "metric": rule.metric_name, + "value": value, + "threshold": rule.threshold, + "resource": rule.resource_name, + "source_module": rule.source_module, + "timestamp": datetime.now(timezone.utc).isoformat(), + } + + smtp_cfg = app.config.get("SMTP", {}) + webhook_cfg = app.config.get("WEBHOOK", {}) + results = await dispatch_notifications(smtp_cfg, webhook_cfg, alert_data) + + # Update event row with delivery results in a new transaction + async with app.db_sessionmaker() as db: + async with db.begin(): + event_result = await db.execute( + select(AlertEvent).where(AlertEvent.id == event_id) + ) + event = event_result.scalar_one_or_none() + if event: + email_r = results.get("email") + if email_r: + event.email_sent = email_r["sent"] + event.email_error = email_r.get("error") + webhook_r = results.get("webhook") + if webhook_r: + event.webhook_sent = webhook_r["sent"] + event.webhook_error = webhook_r.get("error")