diff --git a/steward/app.py b/steward/app.py index 21ce721..74c0389 100644 --- a/steward/app.py +++ b/steward/app.py @@ -344,9 +344,13 @@ def _register_core_tasks(app: Quart) -> None: from .core.cleanup import run_cleanup as _cleanup await _cleanup(app) - async def run_silence_check(): - from .core.alerts import check_silence - await check_silence(app) + async def run_alert_reconcile(): + from .core.alerts import check_silence, notify_after_maintenance + for step in (check_silence, notify_after_maintenance): + try: + await step(app) + except Exception: + app.logger.exception("alert_reconcile: %s failed", step.__name__) async def run_report_check(): from .core.reports import check_and_send @@ -372,10 +376,11 @@ def _register_core_tasks(app: Quart) -> None: interval_seconds=60, run_on_startup=True, ), - # Rules whose series has gone quiet get a visible no_data mark. + # Rules whose series has gone quiet get a visible no_data mark; alerts + # a maintenance window muted notify once it ends. ScheduledTask( - name="alert_silence_check", - coro_factory=run_silence_check, + name="alert_reconcile", + coro_factory=run_alert_reconcile, interval_seconds=60, run_on_startup=False, ), diff --git a/steward/core/alerts.py b/steward/core/alerts.py index 39b3ecd..06e2dc9 100644 --- a/steward/core/alerts.py +++ b/steward/core/alerts.py @@ -6,7 +6,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import TYPE_CHECKING -from sqlalchemy import select +from sqlalchemy import event as sa_event, select from sqlalchemy.ext.asyncio import AsyncSession from steward.models.alerts import ( @@ -92,6 +92,54 @@ def _in_window(w: MaintenanceWindow, source_module: str, resource_name: str) -> 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)) + + async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> None: """Write metric rows and evaluate the alert rules that watch them. @@ -103,8 +151,9 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non 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. Notification I/O is deferred via asyncio.create_task() so no - network calls occur while the transaction is open. + 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. """ if not rows: return @@ -147,25 +196,23 @@ async def record_metrics(session: AsyncSession, rows: list[PluginMetric]) -> Non continue samples = sorted(by_key[(rule.source_module, rule.resource_name, rule.metric_name)], key=lambda r: r.recorded_at) - muted = any(_in_window(w, rule.source_module, rule.resource_name) for w in windows) - resumed = _track_samples(session, rule, alert_state, samples, now) - if resumed and rule.notify_no_data and not muted and _app is not None: + 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 - asyncio.create_task( - _dispatch_notification(_app, "data_resumed", rule, value, event_id, note)) - if muted: - continue # suppressed by maintenance window + _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) + sample.recorded_at, note=muted) + if muted: + continue for notif_type, event_id in notifications: - if _app is not None: - asyncio.create_task( - _dispatch_notification(_app, 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: - asyncio.create_task( - _run_ansible_action(_app, rule, sample.value, event_id)) + _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) # A series is silent once it has gone this long without a sample, or three of @@ -219,11 +266,12 @@ def _duration(seconds: float) -> str: def _track_samples(session: AsyncSession, rule: AlertRule, state: AlertState, - samples: list[PluginMetric], now: datetime + 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.""" + 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: @@ -235,7 +283,8 @@ def _track_samples(session: AsyncSession, rule: AlertRule, state: AlertState, return None note = f"data resumed after {_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=note) + 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 @@ -301,8 +350,10 @@ async def _evaluate_rule( 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)].""" + """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), @@ -317,7 +368,8 @@ async def _evaluate_rule( 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) + 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 @@ -426,32 +478,17 @@ async def _run_ansible_action( ) -> 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. + 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 import executor, sources as src_module - from steward.models.ansible import AnsibleRun, AnsibleRunStatus + from steward.ansible.runner import trigger_run 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). + playbook = action.get("playbook") or "" + inventory = action.get("inventory") or "" + source_name = action.get("source") or "" params: dict = { "extra_vars": list(action.get("extra_vars") or []) + alert_extra_vars(rule, value), } @@ -461,25 +498,32 @@ async def _run_ansible_action( 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) + 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) - 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( + ev = (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) + if ev is None: + return + if run is not None: + ev.ansible_run_id = run.id + else: + why = f"Ansible action not started: {err}" + ev.note = f"{ev.note}; {why}" if ev.note else why async def check_silence(app: "Quart") -> None: @@ -521,10 +565,49 @@ async def check_silence(app: "Quart") -> None: note = f"no sample for {_duration((now - state.last_sample_at).total_seconds())}" if state.sample_interval_s: note += f" (usually every {_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=note) - if rule.notify_no_data and not any( - _in_window(w, rule.source_module, rule.resource_name) for w in windows): + 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: - asyncio.create_task(_dispatch_notification(app, "no_data", rule, None, event_id, note)) + _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)) diff --git a/steward/templates/alerts/maintenance.html b/steward/templates/alerts/maintenance.html index 4832abf..f2b941f 100644 --- a/steward/templates/alerts/maintenance.html +++ b/steward/templates/alerts/maintenance.html @@ -11,8 +11,9 @@

- During an active window, alert rule evaluation is suppressed for the matching scope. - Metrics are still recorded — only notifications are silenced. + During an active window, alerts in its scope still change state and show on the Alerts page, + but send no notifications and run no Ansible actions. Each event a window mutes says so in + the alert's history. An alert that is still firing when the window ends notifies then, and runs its action.

{% if windows %} diff --git a/steward/templates/alerts/maintenance_form.html b/steward/templates/alerts/maintenance_form.html index 841c03a..7870ecc 100644 --- a/steward/templates/alerts/maintenance_form.html +++ b/steward/templates/alerts/maintenance_form.html @@ -50,7 +50,7 @@

- Leave blank to suppress all resources within the selected source. + Leave blank to mute all resources within the selected source.

@@ -64,9 +64,9 @@
How scope works
-
All alerts — every alert rule is suppressed
-
By source — e.g. source=ping suppresses all ping alerts
-
By source + resource — e.g. ping/my-server suppresses only that host's ping alerts
+
All alerts — every alert rule's notifications are muted
+
By source — e.g. source=icmp mutes all ping alerts
+
By source + resource — e.g. icmp/my-server mutes only that host's ping alerts
diff --git a/tests/core/test_alert_no_data.py b/tests/core/test_alert_no_data.py index 148c99e..418d61b 100644 --- a/tests/core/test_alert_no_data.py +++ b/tests/core/test_alert_no_data.py @@ -96,3 +96,14 @@ def test_a_sample_after_silence_clears_it_and_records_the_resume(): assert value == 42.0 and note == "data resumed after 15m silent" [event] = session.added assert (event.id, event.from_state, event.to_state) == (event_id, "no_data", "firing") + + +def test_muted_by_names_the_window_that_covers_the_rule(): + from steward.core.alerts import _muted_by + rule = NS(source_module="host_agent", resource_name="nas") + windows = [NS(name="other", scope_source="docker", scope_resource=None), + NS(name="nas work", scope_source="host_agent", scope_resource="nas")] + assert _muted_by(windows, rule) == "notification muted: maintenance window nas work" + assert _muted_by(windows[:1], rule) is None + assert _muted_by([NS(name="all", scope_source=None, scope_resource=None)], rule) == ( + "notification muted: maintenance window all") diff --git a/tests/integration/test_alert_ansible_action.py b/tests/integration/test_alert_ansible_action.py index a72f5d1..e41db95 100644 --- a/tests/integration/test_alert_ansible_action.py +++ b/tests/integration/test_alert_ansible_action.py @@ -48,7 +48,8 @@ def test_firing_triggers_ansible_action(app, tmp_path): - debug: msg: "FIRED-{{ steward_alert_metric }}" """)) - (tmp_path / "inv.ini").write_text("localhost ansible_connection=local\n") + # inventory.ini: an inventory name the source lists, as trigger_run's guard requires. + (tmp_path / "inventory.ini").write_text("localhost ansible_connection=local\n") # An ansible source the action resolves against. app.config["ANSIBLE"] = { @@ -57,7 +58,7 @@ def test_firing_triggers_ansible_action(app, tmp_path): uid = str(uuid.uuid4()) rule_id = str(uuid.uuid4()) - action = {"source": "t", "playbook": "fire.yml", "inventory": "inv.ini"} + action = {"source": "t", "playbook": "fire.yml", "inventory": "inventory.ini"} async def _go(): async with app.db_sessionmaker() as s: @@ -82,9 +83,13 @@ def test_firing_triggers_ansible_action(app, tmp_path): async with s.begin(): await record_metric(s, "host_agent", "srv1", "cpu_pct", 95.0) - # Let the fire-and-forget dispatch + action tasks complete. - pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] - await asyncio.gather(*pending, return_exceptions=True) + # Let the dispatch, the action and the run it starts finish. The run's + # task is created by the action's task, so wait until none are left. + loop = asyncio.get_running_loop() + deadline = loop.time() + 120 + while pending := [t for t in asyncio.all_tasks() if t is not asyncio.current_task()]: + assert loop.time() < deadline, f"tasks still running: {pending}" + await asyncio.wait(pending, timeout=deadline - loop.time()) async with app.db_sessionmaker() as s: run = (await s.execute( diff --git a/tests/integration/test_alert_dispatch.py b/tests/integration/test_alert_dispatch.py new file mode 100644 index 0000000..ed2e18d --- /dev/null +++ b/tests/integration/test_alert_dispatch.py @@ -0,0 +1,184 @@ +"""Integration (#5063): against real Postgres, notifications and actions start +only after the evaluating transaction commits; a maintenance window mutes them +without stopping the state from moving, and an alert still firing when the +window ends notifies then; an alert action is held to what its source lists.""" +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +import pytest + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +async def _rule(app, tag, **extra): + from steward.alerts import service as alert_svc + from steward.models.metrics import PluginMetric + from steward.models.users import User, UserRole + res = f"box-{tag}" + async with app.db_sessionmaker() as s: + async with s.begin(): + user = User(username=f"dp-{tag}", email=f"dp-{tag}@test", password_hash="x", + role=UserRole.admin) + s.add(user) + s.add(PluginMetric(source_module="host_agent", resource_name=res, + metric_name="cpu_pct", value=1.0)) + await s.flush() + fields = await alert_svc.rule_fields( + s, name=f"hot {tag}", source_module="host_agent", resource_name=res, + metric_name="cpu_pct", operator=">", threshold=90) + rule = await alert_svc.add_rule(s, fields, created_by=user.id, **extra) + return rule.id, res, user.id + + +async def _events(app, rule_id): + from sqlalchemy import select + from steward.models.alerts import AlertEvent + async with app.db_sessionmaker() as s: + rows = (await s.execute(select(AlertEvent).where(AlertEvent.rule_id == rule_id) + .order_by(AlertEvent.transitioned_at))).scalars().all() + return [(e.from_state, e.to_state, e.note, e.ansible_run_id) for e in rows] + + +async def _settle(): + for _ in range(5): + await asyncio.sleep(0) + + +def test_a_rolled_back_evaluation_notifies_nobody(app): + from steward.core import alerts + + sent = [] + + async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None): + sent.append((notif_type, value)) + + async def _go(): + rule_id, res, _ = await _rule(app, uuid.uuid4().hex[:8]) + + class Boom(Exception): + pass + + with pytest.raises(Boom): + async with app.db_sessionmaker() as s: + async with s.begin(): + await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0) + raise Boom + await _settle() + assert sent == [] and await _events(app, rule_id) == [] + + async with app.db_sessionmaker() as s: + async with s.begin(): + await alerts.record_metric(s, "host_agent", res, "cpu_pct", 96.0) + await _settle() + assert sent == [] # not before the commit + await _settle() + assert sent == [("firing", 96.0)] + + with patch.object(alerts, "_dispatch_notification", fake_dispatch): + asyncio.run(_go()) + + +def test_maintenance_mutes_but_the_state_moves_and_notifies_when_it_ends(app): + from sqlalchemy import select, update + from steward.core import alerts + from steward.models.alerts import AlertState, AlertStateEnum, MaintenanceWindow + + sent = [] + + async def fake_dispatch(app_, notif_type, rule, value, event_id, note=None): + sent.append((notif_type, value)) + + async def _go(): + tag = uuid.uuid4().hex[:8] + rule_id, res, user_id = await _rule(app, tag) + now = datetime.now(timezone.utc) + async with app.db_sessionmaker() as s: + async with s.begin(): + window = MaintenanceWindow( + name=f"patching {tag}", start_at=now - timedelta(minutes=1), + end_at=now + timedelta(hours=1), scope_source="host_agent", + scope_resource=res, created_by=user_id) + s.add(window) + window_id = window.id + + async with app.db_sessionmaker() as s: + async with s.begin(): + await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0) + await _settle() + async with app.db_sessionmaker() as s: + st = (await s.execute(select(AlertState).where( + AlertState.rule_id == rule_id))).scalar_one() + assert st.state == AlertStateEnum.firing # evaluated inside the window + [(frm, to, note, _)] = await _events(app, rule_id) + assert (frm, to, note) == ("inactive", "firing", + f"notification muted: maintenance window patching {tag}") + assert sent == [] + + await alerts.notify_after_maintenance(app) # window still open: held + await _settle() + assert sent == [] + + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute(update(MaintenanceWindow).where(MaintenanceWindow.id == window_id) + .values(end_at=datetime.now(timezone.utc))) + await alerts.notify_after_maintenance(app) + await alerts.notify_after_maintenance(app) # released once + await _settle() + assert sent == [("firing", 95.0)] + [(_, _, note, _)] = await _events(app, rule_id) + assert note.endswith("; notified when the window ended") + + with patch.object(alerts, "_dispatch_notification", fake_dispatch): + asyncio.run(_go()) + + +def test_an_alert_action_outside_its_source_is_refused(app, tmp_path): + from steward.core import alerts + from steward.models.ansible import AnsibleRun + from sqlalchemy import func, select + + (tmp_path / "fix.yml").write_text("- hosts: all\n tasks: []\n") + (tmp_path / "inventory.ini").write_text("localhost\n") + app.config["ANSIBLE"] = {"sources": [{"name": "t", "type": "local", "path": str(tmp_path)}]} + + async def quiet(*args, **kwargs): + pass + + async def _go(): + tag = uuid.uuid4().hex[:8] + rule_id, res, _ = await _rule( + app, tag, ansible_action={"source": "t", "playbook": "../../etc/evil.yml", + "inventory": "inventory.ini"}) + async with app.db_sessionmaker() as s: + before = (await s.execute(select(func.count()).select_from(AnsibleRun))).scalar_one() + async with app.db_sessionmaker() as s: + async with s.begin(): + await alerts.record_metric(s, "host_agent", res, "cpu_pct", 95.0) + loop = asyncio.get_running_loop() + deadline = loop.time() + 30 + while alerts._background: + assert loop.time() < deadline, "alert tasks did not finish" + await asyncio.wait(set(alerts._background), timeout=deadline - loop.time()) + [(frm, to, note, run_id)] = await _events(app, rule_id) + assert (frm, to, run_id) == ("inactive", "firing", None) + assert note == "Ansible action not started: Source 't' has no playbook '../../etc/evil.yml'" + async with app.db_sessionmaker() as s: + after = (await s.execute(select(func.count()).select_from(AnsibleRun))).scalar_one() + assert after == before + + with patch.object(alerts, "_dispatch_notification", quiet): + asyncio.run(_go())