from __future__ import annotations import asyncio import logging import uuid from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING from sqlalchemy import event as sa_event, select from sqlalchemy.ext.asyncio import AsyncSession from steward.models.alerts import ( AlertEvent, AlertOperator, AlertRule, AlertState, AlertStateEnum, MaintenanceWindow, ) from steward.core.formatting import human_duration from steward.models.metrics import PluginMetric, PluginMetricLatest 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) -> dict: """Ansible extra-vars describing the alert that fired. Injected into every alert-triggered playbook run so the playbook can target the exact resource/metric that fired without templating. They reach ansible-playbook in its JSON extra-vars file, not as `-e key=value`, which Ansible splits on whitespace: a rule named "High CPU on web" stays one value (#5157). Pure function. """ return { "steward_alert_rule": rule.name, "steward_alert_source": rule.source_module, "steward_alert_resource": rule.resource_name, "steward_alert_metric": rule.metric_name, "steward_alert_value": value, "steward_alert_threshold": rule.threshold, "steward_alert_operator": rule.operator.value, } def action_vars(lines) -> dict: """An alert action's stored `key=value` lines as a map, split at the first `=` so a value keeps its spaces, quotes and further `=`.""" out: dict = {} for line in lines or (): key, sep, val = str(line).partition("=") if sep and key.strip(): out[key.strip()] = val return out async def record_metric( session: AsyncSession, source_module: str, resource_name: str, metric_name: str, value: float, ) -> None: """One metric sample: record_metrics with a single row.""" await record_metrics(session, [PluginMetric( source_module=source_module, resource_name=resource_name, metric_name=metric_name, value=value)]) class MetricBatch: """Gathers one transaction's samples so they are recorded, and their alert rules evaluated, in a single record_metrics call: batch = MetricBatch() batch.add(source_module="unifi", resource_name="wan", metric_name="is_up", value=1.0) await batch.record(session) """ def __init__(self) -> None: self.rows: list[PluginMetric] = [] def add(self, source_module: str, resource_name: str, metric_name: str, value: float, recorded_at: datetime | None = None) -> None: self.rows.append(PluginMetric(source_module=source_module, resource_name=resource_name, metric_name=metric_name, value=float(value), recorded_at=recorded_at)) async def record(self, session: AsyncSession) -> None: rows, self.rows = self.rows, [] await record_metrics(session, rows) def _in_window(w: MaintenanceWindow, source_module: str, resource_name: str) -> bool: if w.scope_source is None: return True return w.scope_source == source_module and w.scope_resource in (None, resource_name) MUTED = "notification muted: maintenance window" RELEASED = "notified when the window ended" def _muted_by(windows, rule: AlertRule) -> str | None: """The note an event carries when a maintenance window holds back its notification, or None when no window covers the rule.""" w = next((w for w in windows if _in_window(w, rule.source_module, rule.resource_name)), None) return f"{MUTED} {w.name}" if w else None # Notification and action tasks in flight. Held here so the event loop's weak # references are not the only ones (a dropped task is silently never run). _background: set[asyncio.Task] = set() _AFTER_COMMIT = "steward_alert_after_commit" def _spawn(coro) -> None: task = asyncio.create_task(coro) _background.add(task) task.add_done_callback(_background.discard) def _after_commit(session: AsyncSession, fn, *args) -> None: """Run fn(app, *args) once the session's transaction commits, and never if it rolls back: a notification or a playbook run must describe what was actually stored.""" if _app is None: return info = session.info if _AFTER_COMMIT not in info: info[_AFTER_COMMIT] = [] app = _app def _committed(_sync_session) -> None: jobs, info[_AFTER_COMMIT] = info[_AFTER_COMMIT], [] for job_fn, job_args in jobs: _spawn(job_fn(app, *job_args)) def _rolled_back(_sync_session) -> None: info[_AFTER_COMMIT] = [] sa_event.listen(session.sync_session, "after_commit", _committed) sa_event.listen(session.sync_session, "after_rollback", _rolled_back) info[_AFTER_COMMIT].append((fn, args)) LATEST_CHUNK = 1000 async def _keep_latest(session: AsyncSession, rows: list[PluginMetric]) -> None: """Move each key's row in plugin_metric_latest to the newest sample in `rows`. A row only moves forward, so a backlog arriving after newer samples leaves it alone. Keys go in sorted order, so two batches sharing keys lock them in the same order and cannot deadlock.""" from sqlalchemy.dialects.postgresql import insert as pg_insert newest: dict[tuple[str, str, str], PluginMetric] = {} for row in rows: key = (row.source_module, row.resource_name, row.metric_name) if key not in newest or row.recorded_at > newest[key].recorded_at: newest[key] = row values = [{"source_module": k[0], "resource_name": k[1], "metric_name": k[2], "value": newest[k].value, "recorded_at": newest[k].recorded_at} for k in sorted(newest)] # Five parameters a row; chunks keep a statement well under asyncpg's 32767. for i in range(0, len(values), LATEST_CHUNK): stmt = pg_insert(PluginMetricLatest).values(values[i:i + LATEST_CHUNK]) await session.execute(stmt.on_conflict_do_update( index_elements=["source_module", "resource_name", "metric_name"], set_={"value": stmt.excluded.value, "recorded_at": stmt.excluded.recorded_at}, where=PluginMetricLatest.recorded_at <= stmt.excluded.recorded_at, )) async def record_metrics(session: AsyncSession, rows: list[PluginMetric], phases=None) -> None: """Write metric rows and evaluate the alert rules that watch them. Every ingest path comes through here, so a rule on any metric Steward records can fire. Must be called inside an active transaction: async with session.begin(): await record_metrics(session, rows) Maintenance windows, rules and their states are read once per batch, not once per row. Each rule sees its key's samples oldest first, so a backlog (an agent catching up after an outage) counts consecutive checks as they happened. A maintenance window holds back notifications and actions, not evaluation: the state still moves, and the event says it was muted. Notifications and actions start only once the transaction commits. `phases`, when given, is marked after each step (flush, rules, evaluate) so a slow caller can say which one its time went to. """ mark = phases.mark if phases is not None else (lambda _name: None) if not rows: return now = datetime.now(timezone.utc) for row in rows: if row.recorded_at is None: row.recorded_at = now session.add_all(rows) await session.flush() await _keep_latest(session, rows) mark("flush") by_key: dict[tuple[str, str, str], list[PluginMetric]] = {} for row in rows: by_key.setdefault((row.source_module, row.resource_name, row.metric_name), []).append(row) sources = {k[0] for k in by_key} metric_names = {k[2] for k in by_key} rules = (await session.execute( select(AlertRule).where( AlertRule.source_module.in_(sources), AlertRule.metric_name.in_(metric_names), AlertRule.enabled.is_(True), ) )).scalars().all() rules = [r for r in rules if (r.source_module, r.resource_name, r.metric_name) in by_key] if not rules: mark("rules") return windows = (await session.execute( select(MaintenanceWindow).where( MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now) )).scalars().all() states = {st.rule_id: st for st in (await session.execute( select(AlertState).where(AlertState.rule_id.in_([r.id for r in rules])) )).scalars()} mark("rules") for rule in rules: alert_state = states.get(rule.id) if alert_state is None: logger.warning("No alert_state row for rule %s, skipping", rule.id) continue samples = sorted(by_key[(rule.source_module, rule.resource_name, rule.metric_name)], key=lambda r: r.recorded_at) muted = _muted_by(windows, rule) resumed = _track_samples(session, rule, alert_state, samples, now, muted) if resumed and rule.notify_no_data and not muted: event_id, value, note = resumed _after_commit(session, _dispatch_notification, "data_resumed", rule, value, event_id, note) for sample in samples: notifications = await _evaluate_rule(session, rule, alert_state, sample.value, sample.recorded_at, note=muted) if muted: continue for notif_type, event_id in notifications: _after_commit(session, _dispatch_notification, notif_type, rule, sample.value, event_id) # Admin-configured Ansible action runs only on transition-to-firing. if notif_type == "firing" and rule.ansible_action: _after_commit(session, _run_ansible_action, rule, sample.value, event_id) mark("evaluate") # A series is silent once it has gone this long without a sample, or three of # its usual gaps if that is longer. Until its cadence is known (fewer than two # samples seen), it gets an hour: a monitor may check only every ten minutes. NO_DATA_FLOOR_S = 300.0 NO_DATA_GAPS = 3 NO_DATA_UNKNOWN_S = 3600.0 def next_interval(interval: float | None, gap: float) -> float: """A series' usual gap between samples, updated with one more gap. A smoothed average, so one late sample barely moves it. A gap longer than the silence threshold counts only as that threshold: an outage is not the series' cadence. A real change of cadence still gets learned, a step at a time (each step at most 1.5x). """ gap = max(0.0, gap) if interval is None or interval <= 0: return gap return 0.75 * interval + 0.25 * min(gap, NO_DATA_GAPS * interval) def no_data_after(interval: float | None) -> float: """Seconds without a sample before a series counts as silent.""" if not interval: return NO_DATA_UNKNOWN_S return max(NO_DATA_FLOOR_S, NO_DATA_GAPS * interval) def silent_since(last_sample_at: datetime | None, interval: float | None, watching_since: datetime, now: datetime) -> datetime | None: """When a series went silent, or None while it is still reporting. A series that has never had a sample counts from when the rule started watching it (created, changed or re-enabled). """ last = last_sample_at or watching_since limit = timedelta(seconds=no_data_after(interval)) return last + limit if now - last > limit else None def _track_samples(session: AsyncSession, rule: AlertRule, state: AlertState, samples: list[PluginMetric], now: datetime, muted: str | None = None ) -> tuple[str, float, str] | None: """Note a rule's new samples (oldest first) for silence tracking. When the series had been marked no_data, clear it and record a resume event, returned as (event_id, value, note) for the caller to notify. `muted` is added to the event's note when a maintenance window holds it back.""" for sample in samples: if state.last_sample_at is not None: if sample.recorded_at <= state.last_sample_at: continue # late or duplicate: not a gap gap = (sample.recorded_at - state.last_sample_at).total_seconds() state.sample_interval_s = next_interval(state.sample_interval_s, gap) state.last_sample_at = sample.recorded_at if state.no_data_since is None or not samples: return None note = f"data resumed after {human_duration((now - state.no_data_since).total_seconds())} silent" value = samples[-1].value event_id = _add_event(session, rule.id, "no_data", state.state.value, value, now, note=f"{note}; {muted}" if muted else note) state.no_data_since = None return event_id, value, note 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, note: str | None = None, ) -> list[tuple[str, str]]: """Apply one sample to a rule's state. Returns [(notif_type, event_id)]; an event it records carries `note`.""" 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, note=note) 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 or re-enabled), and start its silence watch over. 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) # The watch starts over now: silence is measured from here, against the # cadence of whatever the rule watches from now on. state.last_transition_at = now state.last_sample_at = None state.sample_interval_s = None state.no_data_since = None 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 | None, event_id: str, note: str | None = None, ) -> 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, "note": note, "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). Starts after the evaluation transaction commits. Goes through runner.trigger_run like every other run path, so the playbook and the inventory must be ones the source lists (ansible/guards.py). The firing event links to the run, or says why the action did not start. """ from steward.ansible.runner import IN_FLIGHT, trigger_run from steward.models.ansible import AnsibleRun if rule.id in _actions_starting: await _note_event(app, event_id, "Ansible action skipped: an earlier " "firing's action was still starting") return async with app.db_sessionmaker() as db: busy = (await db.execute( select(AnsibleRun.id) .join(AlertEvent, AlertEvent.ansible_run_id == AnsibleRun.id) .where(AlertEvent.rule_id == rule.id, AnsibleRun.status.in_(IN_FLIGHT)) .limit(1))).scalar_one_or_none() if busy: # One action run at a time per rule (#5158): a re-firing while the # last one is still going would run the same fix over it. await _note_event(app, event_id, f"Ansible action skipped: run {busy} " "from an earlier firing was still going") return _actions_starting.add(rule.id) try: await _start_ansible_action(app, rule, value, event_id, trigger_run) finally: _actions_starting.discard(rule.id) # Rules whose action is between the in-flight check and its run being linked # to the event, so two firings that close together can't both start one. _actions_starting: set[str] = set() async def _note_event(app: "Quart", event_id: str, why: str) -> None: async with app.db_sessionmaker() as db: async with db.begin(): ev = (await db.execute( select(AlertEvent).where(AlertEvent.id == event_id))).scalar_one_or_none() if ev is not None: ev.note = f"{ev.note}; {why}" if ev.note else why async def _start_ansible_action(app, rule, value, event_id, trigger_run) -> None: action = rule.ansible_action or {} playbook = action.get("playbook") or "" inventory = action.get("inventory") or "" source_name = action.get("source") or "" # One JSON extra-vars file; the firing's own vars win over the action's. params: dict = { "extra_vars_map": {**action_vars(action.get("extra_vars")), **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 scope = inventory if inventory.startswith("steward:") else f"repo:{source_name}:{inventory}" if not playbook or not inventory: run, err = None, "the action has no playbook or inventory" else: try: run, _source, err = await trigger_run( app, source_name=source_name, playbook_path=playbook, inventory_scope=scope, params=params, triggered_by=None, origin="alert", origin_id=rule.id) except Exception as exc: logger.exception("Ansible action for rule %r failed to start", rule.name) run, err = None, str(exc) if err: logger.error("Ansible action for rule %r not started: %s", rule.name, err) if run is None: await _note_event(app, event_id, f"Ansible action not started: {err}") return async with app.db_sessionmaker() as db: async with db.begin(): ev = (await db.execute( select(AlertEvent).where(AlertEvent.id == event_id))).scalar_one_or_none() if ev is not None: ev.ansible_run_id = run.id async def check_silence(app: "Quart") -> None: """Mark the enabled rules whose series has gone silent (no_data). Run every minute by the core scheduler. A rule is silent once its series has had no sample for no_data_after(its usual gap); one that never had a sample counts from when the rule started watching. The mark sits beside the state, so a firing alert that goes silent stays firing. Each silence is recorded once, as an event, and notified once (rule.notify_no_data, outside maintenance windows), after the transaction commits. """ now = datetime.now(timezone.utc) notify: list[tuple[AlertRule, str, str]] = [] async with app.db_sessionmaker() as session: async with session.begin(): rows = (await session.execute( select(AlertState, AlertRule) .join(AlertRule, AlertState.rule_id == AlertRule.id) .where(AlertRule.enabled.is_(True), AlertState.no_data_since.is_(None)) )).all() if not rows: return windows = (await session.execute( select(MaintenanceWindow).where( MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now) )).scalars().all() for state, rule in rows: watching_since = max(rule.created_at, state.last_transition_at) since = silent_since(state.last_sample_at, state.sample_interval_s, watching_since, now) if since is None: continue state.no_data_since = since if state.last_sample_at is None: note = (f"no sample yet, {human_duration((now - watching_since).total_seconds())}" " after the rule started watching") else: note = f"no sample for {human_duration((now - state.last_sample_at).total_seconds())}" if state.sample_interval_s: note += f" (usually every {human_duration(state.sample_interval_s)})" muted = _muted_by(windows, rule) event_id = _add_event(session, rule.id, state.state.value, "no_data", None, now, note=f"{note}; {muted}" if muted else note) if rule.notify_no_data and not muted: notify.append((rule, event_id, note)) for rule, event_id, note in notify: _spawn(_dispatch_notification(app, "no_data", rule, None, event_id, note)) async def notify_after_maintenance(app: "Quart") -> None: """Notify the alerts that started firing inside a maintenance window and are still firing now that no window covers them, and run their actions: a window delays what it mutes, it does not discard it. Run every minute by the core scheduler; each muted firing is released once.""" now = datetime.now(timezone.utc) release: list[tuple[AlertRule, str, float | None]] = [] async with app.db_sessionmaker() as session: async with session.begin(): rows = (await session.execute( select(AlertState, AlertRule) .join(AlertRule, AlertState.rule_id == AlertRule.id) .where(AlertRule.enabled.is_(True), AlertState.state == AlertStateEnum.firing) )).all() if not rows: return windows = (await session.execute( select(MaintenanceWindow).where( MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now) )).scalars().all() for state, rule in rows: if _muted_by(windows, rule): continue fired = (await session.execute( select(AlertEvent) .where(AlertEvent.rule_id == rule.id, AlertEvent.to_state == "firing") .order_by(AlertEvent.transitioned_at.desc()).limit(1) )).scalar_one_or_none() note = (fired.note or "") if fired else "" if MUTED not in note or RELEASED in note: continue fired.note = f"{note}; {RELEASED}" release.append((rule, fired.id, fired.metric_value)) for rule, event_id, value in release: _spawn(_dispatch_notification(app, "firing", rule, value, event_id)) if rule.ansible_action: _spawn(_run_ansible_action(app, rule, value, event_id))