bvandeusenandClaude Opus 5.5
a6c24299a2
#5059 "Alert state machine: acknowledgement holds until resolve; N consecutive OK to resolve; editing, disabling or deleting a rule resets it". - The state machine is now next_step(), a pure function of one sample. An acknowledged alert stays acknowledged while breached; before, the next breaching sample re-fired it. Firing and acknowledged alerts resolve after consecutive_ok_required OK checks in a row (new rule column, default 1; rule form, rules list and MCP create/update_alert_rule clear_after_checks). - Changing what a rule watches (source, resource, metric, operator, threshold) or disabling it resets its alert to inactive; a firing or acknowledged one records a resolved event whose new note says why, and nobody is notified. Web and MCP edits share alerts.service.apply_rule_change. - Migration 0028: consecutive_ok_required, consecutive_ok_count, alert_events.note, and metric_value nullable (a reset has no sample). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
371 lines
13 KiB
Python
371 lines
13 KiB
Python
from __future__ import annotations
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from typing import TYPE_CHECKING
|
|
|
|
from sqlalchemy import and_, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from steward.models.alerts import (
|
|
AlertEvent,
|
|
AlertOperator,
|
|
AlertRule,
|
|
AlertState,
|
|
AlertStateEnum,
|
|
MaintenanceWindow,
|
|
)
|
|
from steward.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
|
|
|
|
|
|
def alert_extra_vars(rule: AlertRule, value: float) -> list[str]:
|
|
"""Ansible extra-vars describing the alert that fired, as `key=value` strings.
|
|
|
|
Injected into every alert-triggered playbook run so the playbook can target
|
|
the exact resource/metric that fired without templating. Pure function.
|
|
"""
|
|
return [
|
|
f"steward_alert_rule={rule.name}",
|
|
f"steward_alert_source={rule.source_module}",
|
|
f"steward_alert_resource={rule.resource_name}",
|
|
f"steward_alert_metric={rule.metric_name}",
|
|
f"steward_alert_value={value}",
|
|
f"steward_alert_threshold={rule.threshold}",
|
|
f"steward_alert_operator={rule.operator.value}",
|
|
]
|
|
|
|
|
|
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()
|
|
|
|
# Check active maintenance windows — skip rule evaluation if suppressed
|
|
mw_result = await session.execute(
|
|
select(MaintenanceWindow).where(
|
|
MaintenanceWindow.start_at <= now,
|
|
MaintenanceWindow.end_at >= now,
|
|
or_(
|
|
MaintenanceWindow.scope_source.is_(None),
|
|
and_(
|
|
MaintenanceWindow.scope_source == source_module,
|
|
or_(
|
|
MaintenanceWindow.scope_resource.is_(None),
|
|
MaintenanceWindow.scope_resource == resource_name,
|
|
),
|
|
),
|
|
),
|
|
).limit(1)
|
|
)
|
|
if mw_result.scalar_one_or_none() is not None:
|
|
return # suppressed by maintenance window
|
|
|
|
# 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
|
|
)
|
|
)
|
|
# Admin-configured Ansible action runs only on transition-to-firing.
|
|
if notif_type == "firing" and rule.ansible_action:
|
|
asyncio.create_task(
|
|
_run_ansible_action(_app, 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
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Step:
|
|
"""One sample's effect on an alert: the state after it, the two run
|
|
counters, and the event to record ("firing" / "resolved") if any."""
|
|
state: AlertStateEnum
|
|
fail_count: int
|
|
ok_count: int
|
|
event: str | None = None
|
|
|
|
|
|
def next_step(state: AlertStateEnum, fail_count: int, ok_count: int, breached: bool,
|
|
fail_required: int, ok_required: int) -> Step:
|
|
"""The alert state machine, as a pure function of one sample.
|
|
|
|
inactive → pending → firing when `fail_required` breaching checks run in a
|
|
row; a non-breaching check while pending goes back to inactive.
|
|
firing and acknowledged resolve (to inactive, with a resolved event) only
|
|
after `ok_required` non-breaching checks in a row; a breaching check
|
|
resets that run. An acknowledgement holds while the alert is still
|
|
breached: only a resolve clears it.
|
|
"""
|
|
fail_required, ok_required = max(1, fail_required), max(1, ok_required)
|
|
S = AlertStateEnum
|
|
if state in (S.firing, S.acknowledged):
|
|
if breached:
|
|
return Step(state, fail_count + 1, 0)
|
|
if ok_count + 1 >= ok_required:
|
|
return Step(S.inactive, 0, 0, "resolved")
|
|
return Step(state, fail_count, ok_count + 1)
|
|
# inactive, pending (and the transient resolved, read as inactive)
|
|
if not breached:
|
|
return Step(S.inactive, 0, 0)
|
|
fails = fail_count + 1 if state == S.pending else 1
|
|
if fails >= fail_required:
|
|
return Step(S.firing, fails, 0, "firing")
|
|
return Step(S.pending, fails, 0)
|
|
|
|
|
|
async def _evaluate_rule(
|
|
session: AsyncSession,
|
|
rule: AlertRule,
|
|
state: AlertState,
|
|
value: float,
|
|
now: datetime,
|
|
) -> list[tuple[str, str]]:
|
|
"""Apply one sample to a rule's state. Returns [(notif_type, event_id)]."""
|
|
from_state = state.state
|
|
step = next_step(from_state, state.consecutive_failure_count, state.consecutive_ok_count,
|
|
_is_breached(value, rule.operator, rule.threshold),
|
|
rule.consecutive_failures_required, rule.consecutive_ok_required)
|
|
state.consecutive_failure_count = step.fail_count
|
|
state.consecutive_ok_count = step.ok_count
|
|
notifications: list[tuple[str, str]] = []
|
|
if step.state != from_state:
|
|
state.state = step.state
|
|
state.last_transition_at = now
|
|
if step.state == AlertStateEnum.inactive:
|
|
state.acknowledged_by = None
|
|
state.acknowledged_at = None
|
|
if step.event:
|
|
event_id = _add_event(session, rule.id, from_state.value, step.event, value, now)
|
|
notifications.append((step.event, event_id))
|
|
await session.flush()
|
|
return notifications
|
|
|
|
|
|
async def reset_rule_state(session: AsyncSession, rule: AlertRule, note: str) -> None:
|
|
"""Put a rule's alert back to inactive because the rule itself changed
|
|
(what it watches, its threshold, or it was disabled). An alert that was
|
|
firing or acknowledged gets a resolved event carrying `note`; nobody is
|
|
notified, since nothing about the watched resource changed."""
|
|
state = (await session.execute(
|
|
select(AlertState).where(AlertState.rule_id == rule.id))).scalar_one_or_none()
|
|
if state is None:
|
|
return
|
|
now = datetime.now(timezone.utc)
|
|
if state.state in (AlertStateEnum.firing, AlertStateEnum.acknowledged):
|
|
_add_event(session, rule.id, state.state.value, "resolved", None, now, note=note)
|
|
if state.state != AlertStateEnum.inactive:
|
|
state.last_transition_at = now
|
|
state.state = AlertStateEnum.inactive
|
|
state.consecutive_failure_count = 0
|
|
state.consecutive_ok_count = 0
|
|
state.acknowledged_by = None
|
|
state.acknowledged_at = None
|
|
|
|
|
|
def _add_event(
|
|
session: AsyncSession,
|
|
rule_id: str,
|
|
from_state: str,
|
|
to_state: str,
|
|
value: float | None,
|
|
now: datetime,
|
|
note: str | None = None,
|
|
) -> 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,
|
|
note=note,
|
|
)
|
|
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 steward.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")
|
|
|
|
|
|
async def _run_ansible_action(
|
|
app: "Quart",
|
|
rule: AlertRule,
|
|
value: float,
|
|
event_id: str,
|
|
) -> None:
|
|
"""Run the rule's configured Ansible playbook (system-triggered).
|
|
|
|
Runs after the evaluation transaction commits (scheduled via create_task).
|
|
Creates a system-triggered AnsibleRun, links it to the firing AlertEvent,
|
|
then executes it with the static params plus injected alert context vars.
|
|
"""
|
|
from steward.ansible import executor, sources as src_module
|
|
from steward.models.ansible import AnsibleRun, AnsibleRunStatus
|
|
|
|
action = rule.ansible_action or {}
|
|
playbook = action.get("playbook")
|
|
inventory = action.get("inventory")
|
|
if not playbook or not inventory:
|
|
logger.error("Ansible action for rule %r missing playbook/inventory", rule.name)
|
|
return
|
|
|
|
try:
|
|
srcs = src_module.get_sources(app.config.get("ANSIBLE", {}))
|
|
except Exception:
|
|
logger.exception("Ansible action for rule %r: could not load sources", rule.name)
|
|
return
|
|
source = next((s for s in srcs if s["name"] == action.get("source")), None)
|
|
if source is None:
|
|
logger.error("Ansible action for rule %r: source %r not configured",
|
|
rule.name, action.get("source"))
|
|
return
|
|
|
|
# Static params + auto-injected alert context (all argv, never shell).
|
|
params: dict = {
|
|
"extra_vars": list(action.get("extra_vars") or []) + alert_extra_vars(rule, value),
|
|
}
|
|
if action.get("limit"):
|
|
params["limit"] = action["limit"]
|
|
if action.get("tags"):
|
|
params["tags"] = action["tags"]
|
|
if action.get("check"):
|
|
params["check"] = True
|
|
|
|
run_id = str(uuid.uuid4())
|
|
async with app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
db.add(AnsibleRun(
|
|
id=run_id,
|
|
playbook_path=playbook,
|
|
inventory_path=inventory,
|
|
source_name=source["name"],
|
|
triggered_by=None,
|
|
status=AnsibleRunStatus.running,
|
|
params=params,
|
|
))
|
|
event = (await db.execute(
|
|
select(AlertEvent).where(AlertEvent.id == event_id))).scalar_one_or_none()
|
|
if event:
|
|
event.ansible_run_id = run_id
|
|
|
|
await executor.start_run(app, run_id, playbook, inventory, source["path"], params)
|