feat: record_metric() and alert state machine
This commit is contained in:
@@ -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")
|
||||
Reference in New Issue
Block a user