"""Alert tools: what is firing, what fired before, the rules, and maintenance windows.""" from __future__ import annotations from datetime import datetime, timedelta, timezone from sqlalchemy import select from steward.alerts import service as alert_svc from steward.core import outbox from steward.mcp._context import current_caller, mcp_app from steward.models.alerts import AlertOperator, AlertRule, AlertState, MaintenanceWindow from steward.models.users import UserRole from ._audit import audit_mcp from ._shape import ( age_s, bounded, clamp_limit, db_session, iso, lookback, now_utc, pick, rnd, ) def _condition(rule: AlertRule) -> str: n = rule.consecutive_failures_required or 1 ok = rule.consecutive_ok_required or 1 tail = f" for {n} consecutive checks" if n > 1 else "" clear = f"; clears after {ok} OK checks" if ok > 1 else "" return f"{rule.metric_name} {rule.operator.value} {rule.threshold:g}{tail}{clear}" def _action(rule: AlertRule) -> dict | None: """The playbook a rule runs when it fires (alerts/routes.py _parse_ansible_action).""" a = rule.ansible_action if not a: return None return {k: a.get(k) for k in ("source", "playbook", "inventory", "limit", "tags", "extra_vars", "check") if a.get(k) is not None} async def list_alerts() -> dict: """Alerts that need attention now: firing, pending, acknowledged but still breaching, or gone silent (no_data). pending: the condition holds but not yet for the rule's required number of consecutive checks. firing: it has. acknowledged: someone has seen it and it still holds. no_data: the metric the rule watches has stopped reporting (no sample for three of its usual gaps, at least 5 minutes); a firing alert that goes silent stays "firing" with `no_data_since` set. Each carries the rule's condition, the resource and metric it watches, `since` (its latest state change) and `last_sample_at`. Firing, unacknowledged alerts come first. An alert that cleared is not here; get_alert_history shows the transitions. """ async with db_session() as db: active = await alert_svc.active_alerts(db) names = await alert_svc.usernames(db, (s.acknowledged_by for s, _ in active)) order = {"firing": 0, "no_data": 1, "pending": 2, "acknowledged": 3} rows = [ { "rule_id": rule.id, "rule_name": rule.name, "state": alert_svc.alert_label(state), "no_data_since": iso(state.no_data_since), "last_sample_at": iso(state.last_sample_at), "source_module": rule.source_module, "resource": rule.resource_name, "metric": rule.metric_name, "condition": _condition(rule), "since": iso(state.last_transition_at), "age_s": age_s(state.last_transition_at), "acknowledged_by": names.get(state.acknowledged_by) if state.acknowledged_by else None, "acknowledged_at": iso(state.acknowledged_at), } for state, rule in active ] rows.sort(key=lambda r: (order.get(r["state"], 4), -(r["age_s"] or 0))) return {"alerts": rows, "total": len(rows)} async def get_alert_history(rule: str = "", range: str = "7d", limit: int = 50) -> dict: """State transitions of alert rules, newest first: when each fired, cleared or was acked. rule: one rule's name or id (list_alert_rules shows them); empty for every rule. range: "1h", "6h", "24h", "7d" (default) or "30d". limit: default 50, max 500. Each transition carries the metric value that caused it and whether the email/webhook notification went out (with the error if not), plus the id of any Ansible run the rule's action triggered. `email` is the email's delivery: state "held" (gathering other alerts; `at` is when it sends), "sending", "retrying" (`attempts`, next try `at`, last `error`), "sent" or "failed" (`error` names each reader not reached); null when no email was queued for it. `by` names the person who made it happen (an acknowledgement, or a rule edit that reset the alert); null when a reading or Steward itself did. """ since, key = lookback(range) limit = clamp_limit(limit, 50, 500) async with db_session() as db: rules = await alert_svc.list_rules(db) chosen = pick(rule, rules, kind="alert rule", list_tool="list_alert_rules") if rule else None rows = await alert_svc.history(db, since, chosen.id if chosen else None, limit) names = await alert_svc.usernames(db, (e.actor_id for e, _ in rows)) emails = await outbox.email_states(db, [e.id for e, _ in rows]) events = [ { "rule_id": r.id, "rule_name": r.name, "from_state": e.from_state, "to_state": e.to_state, "metric_value": rnd(e.metric_value), "note": e.note, "by": names.get(e.actor_id, e.actor_id), "at": iso(e.transitioned_at), "age_s": age_s(e.transitioned_at), "email_sent": e.email_sent, "email_error": e.email_error, "email": ({**emails[e.id], "at": iso(emails[e.id]["at"])} if e.id in emails else None), "webhook_sent": e.webhook_sent, "webhook_error": e.webhook_error, "ansible_run_id": e.ansible_run_id, } for e, r in rows ] out = {"range": key, "rule": chosen.name if chosen else None} out.update(bounded(events, limit, key="events", more="Pass a larger `limit` (max 500), a shorter `range`, or one `rule`.")) if out["truncated"]: # Fetched limit+1 rows to detect the cut; the true total isn't known. out["total"] = None out["more"] = f"More than {limit} transitions in range. Pass a larger `limit` (max 500), a shorter `range`, or one `rule`." return out async def list_alert_rules(enabled_only: bool = False, limit: int = 200) -> dict: """Every alert rule: what it watches, its condition, current state and automatic action. enabled_only: true to leave out disabled rules. limit: default 200, max 500. `state` is the rule's current state: "inactive" (condition not met), "pending", "firing", "acknowledged", "resolved" (cleared since it last fired), "no_data" (inactive and its metric has stopped reporting), or null if it has no state row. `no_data_since` is set while its metric is silent, whatever the state. `ansible_action` is the playbook the rule runs when it fires, if any (`check: true` = dry run). `covers` is "resource" for a rule on one resource, or "all_hosts" / "group" (with `group`) for a rule covering hosts: it has no state of its own, `hosts` names the hosts it covers, and each host's rule is listed with `part_of` naming it. """ limit = clamp_limit(limit, 200, 500) async with db_session() as db: rules = await alert_svc.list_rules(db) states = {s.rule_id: s for s in (await db.execute(select(AlertState))).scalars()} if enabled_only: rules = [r for r in rules if r.enabled] names = {r.id: r.name for r in rules} hosts: dict[str, list[str]] = {} for r in rules: if r.parent_id: hosts.setdefault(r.parent_id, []).append(r.resource_name) rows = [ { "rule_id": r.id, "rule_name": r.name, "enabled": r.enabled, "source_module": r.source_module, "resource": r.resource_name or None, **_covers(r, hosts.get(r.id, []), names.get(r.parent_id)), "metric": r.metric_name, "condition": _condition(r), "state": alert_svc.alert_label(states[r.id]) if r.id in states else None, "state_since": iso(states[r.id].last_transition_at) if r.id in states else None, "no_data_since": iso(states[r.id].no_data_since) if r.id in states else None, "ansible_action": _action(r), } for r in rules ] return bounded(rows, limit, key="rules") async def list_maintenance_windows(include_past: bool = False, limit: int = 50) -> dict: """Maintenance windows, during which matching alerts are held back from notifying. include_past: true to include windows that have ended (default: current and upcoming only). limit: default 50, max 500. `scope` is what a window silences: a source module and/or one resource, or everything when both are null. `active` is true while it is in effect. """ limit = clamp_limit(limit, 50, 500) now = now_utc() async with db_session() as db: windows = await alert_svc.list_maintenance_windows(db) rows = [ { "window_id": w.id, "name": w.name, "start_at": iso(w.start_at), "end_at": iso(w.end_at), "active": w.start_at <= now <= w.end_at, "scope": {"source_module": w.scope_source, "resource": w.scope_resource}, } for w in windows if include_past or w.end_at >= now ] return bounded(rows, limit, key="windows") # ── changes ────────────────────────────────────────────────────────────────── async def get_alert_rule_options(source_module: str = "") -> dict: """What an alert rule can watch: the choices create_alert_rule accepts. source_module: empty lists every source and its metrics; a source's name also lists its resources (hosts, containers, monitors… whatever that source reports under). A monitor reports under its type (icmp/tcp/dns/http) with its name as the resource: is_up (1 up, 0 down) and response_ms. """ async with db_session() as db: sources = alert_svc.source_modules() if not source_module: return {"sources": {src: await alert_svc.metrics_for(db, src) for src in sources}, "operators": [o.value for o in AlertOperator]} if source_module not in sources: raise ValueError(f"source_module {source_module!r} is not one of " f"{', '.join(sources)}.") return {"source_module": source_module, "metrics": await alert_svc.metrics_for(db, source_module), "resources": await alert_svc.resources_for(db, source_module), "operators": [o.value for o in AlertOperator]} async def _rule(db, rule: str) -> AlertRule: return pick(rule, await alert_svc.list_rules(db), kind="alert rule", list_tool="list_alert_rules") def _covers(r: AlertRule, hosts: list[str], parent_name: str | None) -> dict: """What a rule covers (#5464): the hosts under a covering rule, or the covering rule a host's rule is part of.""" out: dict = {"covers": r.covers} if r.covers == "group": out["group"] = r.cover_group if r.covers != "resource": out["hosts"] = sorted(hosts, key=str.lower) if r.parent_id: out["part_of"] = parent_name return out async def _rule_row(db, r: AlertRule) -> dict: hosts = [c.resource_name for c in (await db.execute( select(AlertRule).where(AlertRule.parent_id == r.id))).scalars()] if r.covers != "resource" else [] parent = await db.get(AlertRule, r.parent_id) if r.parent_id else None return {"rule_id": r.id, "rule_name": r.name, "enabled": r.enabled, "source_module": r.source_module, "resource": r.resource_name or None, **_covers(r, hosts, parent.name if parent else None), "metric": r.metric_name, "condition": _condition(r), "notify_no_data": r.notify_no_data, "ansible_action": _action(r)} async def _not_a_child(db, r: AlertRule) -> None: """A host's rule under a covering rule is changed through its parent.""" if r.parent_id: parent = await db.get(AlertRule, r.parent_id) raise alert_svc.PartOfCoveringRule(r, parent.name if parent else "?") async def create_alert_rule(name: str, source_module: str, resource: str, metric: str, operator: str, threshold: float, consecutive_checks: int = 1, clear_after_checks: int = 1, notify_no_data: bool = True, enabled: bool = True, covers: str = "resource", group: str = "") -> dict: """Add an alert rule: notify when `metric` of `resource` crosses `threshold`. source_module, resource, metric: get_alert_rule_options lists the valid choices; a resource must have reported (or, for a monitor, exist). covers: "resource" (default) watches `resource`. "all_hosts" covers every host, "group" every host in the inventory group `group`; pass `resource` as "". Steward then keeps one rule per host, named ": ", and adds one within a minute for a host that starts reporting or joins the group. Only for host_agent. operator: ">", "<", ">=", "<=", "==" or "!=". consecutive_checks: how many checks in a row must breach before it fires (default 1); higher values ride out a single blip. clear_after_checks: how many checks in a row must be OK before a firing or acknowledged alert resolves (default 1). An acknowledgement holds until then, however long the breach lasts. notify_no_data: also notify when the metric stops reporting and when it resumes (default true). The silence shows in list_alerts either way. "Agent offline" is source host_agent, metric seconds_since_seen, "> 180". Notifications go to the channels set in Steward's Settings. The playbook a rule runs when it fires is set separately (set_alert_action, admin). Recorded in the audit log as " via MCP". """ caller = current_caller() async with db_session() as db, db.begin(): fields = await alert_svc.rule_fields( db, name=name, source_module=source_module, resource_name=resource, metric_name=metric, operator=operator, threshold=threshold, consecutive_failures_required=consecutive_checks, consecutive_ok_required=clear_after_checks, notify_no_data=notify_no_data, covers=covers, cover_group=group) r = await alert_svc.add_rule(db, fields, created_by=caller.user_id, enabled=enabled) row = await _rule_row(db, r) await audit_mcp("alert_rule.created", "alert_rule", r.name, {"rule_id": r.id, "source": r.source_module, "resource": r.resource_name or alert_svc.covers_text(r), "metric": r.metric_name}) return row async def update_alert_rule(rule: str, name: str | None = None, source_module: str | None = None, resource: str | None = None, metric: str | None = None, operator: str | None = None, threshold: float | None = None, consecutive_checks: int | None = None, clear_after_checks: int | None = None, notify_no_data: bool | None = None, enabled: bool | None = None, group: str | None = None) -> dict: """Change an alert rule. Only the arguments you pass change; the rest stay. rule: its name or id (list_alert_rules). Pass enabled=false to silence it without deleting it. The changed rule is validated as create_alert_rule validates a new one. Changing what it watches (source, resource, metric, operator, threshold), or disabling or re-enabling it, puts its alert back to inactive; a firing alert records a resolved event saying why. Recorded in the audit log. A rule covering hosts passes every change on to its per-host rules; `group` moves a group-covering rule to another inventory group. What a rule covers is fixed once it exists. A host's rule under a covering rule is changed through that rule, not alone. """ async with db_session() as db, db.begin(): r = await _rule(db, rule) await _not_a_child(db, r) fields = await alert_svc.rule_fields( db, name=r.name if name is None else name, source_module=r.source_module if source_module is None else source_module, resource_name=r.resource_name if resource is None else resource, metric_name=r.metric_name if metric is None else metric, operator=r.operator.value if operator is None else operator, threshold=r.threshold if threshold is None else threshold, consecutive_failures_required=(r.consecutive_failures_required if consecutive_checks is None else consecutive_checks), consecutive_ok_required=(r.consecutive_ok_required if clear_after_checks is None else clear_after_checks), notify_no_data=r.notify_no_data if notify_no_data is None else notify_no_data, covers=r.covers, cover_group=(r.cover_group or "") if group is None else group) await alert_svc.apply_rule_change(db, r, fields, enabled=enabled, actor_id=current_caller().user_id) if r.covers != "resource": await alert_svc.reconcile_children(db, r, actor_id=current_caller().user_id) row = await _rule_row(db, r) changed = sorted(k for k, v in {"name": name, "source_module": source_module, "resource": resource, "metric": metric, "operator": operator, "threshold": threshold, "consecutive_checks": consecutive_checks, "clear_after_checks": clear_after_checks, "notify_no_data": notify_no_data, "enabled": enabled, "group": group}.items() if v is not None) await audit_mcp("alert_rule.updated", "alert_rule", r.name, {"rule_id": r.id, "changed": changed}) return row async def delete_alert_rule(rule: str) -> dict: """Delete an alert rule and its history. To silence it but keep it, update_alert_rule(enabled=false). Deleting a rule that covers hosts deletes its per-host rules and their history too. Admin only; recorded in the audit log.""" async with db_session() as db, db.begin(): r = await _rule(db, rule) await _not_a_child(db, r) await db.delete(r) await audit_mcp("alert_rule.deleted", "alert_rule", r.name, {"rule_id": r.id}) return {"deleted": r.name, "rule_id": r.id} async def acknowledge_alert(rule: str) -> dict: """Acknowledge a firing alert: you have seen it. It stays "acknowledged" while the condition holds and resolves when it clears. rule: the firing rule's name or id (list_alerts). Recorded in the audit log. """ caller = current_caller() async with db_session() as db, db.begin(): r = await _rule(db, rule) try: state = await alert_svc.acknowledge(db, r.id, caller.user_id) except LookupError: raise ValueError(f"{r.name} has never been evaluated, so there is nothing to " "acknowledge.") from None await audit_mcp("alert_rule.acknowledged", "alert_rule", r.name, {"rule_id": r.id}) return {"rule_id": r.id, "rule_name": r.name, "state": state.state.value, "acknowledged_at": iso(state.acknowledged_at)} async def send_test_alert(rule: str) -> dict: """Send a TEST notification for a rule now, to check it reaches someone: email to everyone who gets alerts (each in their own zone) and the webhook, the same channels a real alert of this rule would use. Nothing fires and the rule's history is unchanged. rule: the rule's name or id (list_alert_rules). Returns `ok` (something was sent and nothing failed) and one line per channel saying what happened. Recorded in the audit log. """ async with db_session() as db: r = await _rule(db, rule) sent = await alert_svc.send_test(mcp_app(), r.id) await audit_mcp("alert_rule.tested", "alert_rule", r.name, {"rule_id": r.id, "ok": sent["ok"], "lines": sent["lines"]}) return {"rule_id": r.id, "rule_name": r.name, "ok": sent["ok"], "result": sent["lines"], "email": sent["email"], "webhook": sent["webhook"]} def _when(value: str, what: str) -> datetime: try: dt = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) except ValueError: raise ValueError(f"{what} {value!r} is not an ISO-8601 time, e.g. 2026-10-06T22:00:00Z.") from None return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) def _window_row(w: MaintenanceWindow, now: datetime) -> dict: return {"window_id": w.id, "name": w.name, "start_at": iso(w.start_at), "end_at": iso(w.end_at), "active": w.start_at <= now <= w.end_at, "scope": {"source_module": w.scope_source, "resource": w.scope_resource}} async def create_maintenance_window(name: str, duration_minutes: int = 0, end_at: str = "", start_at: str = "", source_module: str = "", resource: str = "") -> dict: """Hold back alert notifications while you work on something. Give the end as duration_minutes OR end_at (ISO-8601; a time with no zone is UTC). start_at defaults to now. source_module, resource: limit the window to one source, or to one resource of it; both empty silences everything. Alerts still change state during a window, so list_alerts shows them; only notifications are held. Recorded in the audit log. """ now = now_utc() start = _when(start_at, "start_at") if (start_at or "").strip() else now if bool(duration_minutes) == bool((end_at or "").strip()): raise ValueError("give exactly one of duration_minutes or end_at.") if duration_minutes: if duration_minutes < 0: raise ValueError("duration_minutes must be positive.") end = start + timedelta(minutes=duration_minutes) else: end = _when(end_at, "end_at") fields = alert_svc.window_fields(name=name, start_at=start, end_at=end, scope_source=source_module, scope_resource=resource) w = MaintenanceWindow(created_by=current_caller().user_id, **fields) async with db_session() as db, db.begin(): db.add(w) await audit_mcp("maintenance_window.created", "maintenance_window", w.name, {"window_id": w.id, "scope": alert_svc.window_scope(w), "start_at": iso(w.start_at), "end_at": iso(w.end_at)}) return _window_row(w, now) async def end_maintenance_window(window: str) -> dict: """End a maintenance window now, so notifications resume. window: its name or id (list_maintenance_windows). A window in progress ends now and stays on record; one that has not started yet is removed. Recorded in the audit log. """ now = now_utc() async with db_session() as db, db.begin(): windows = await alert_svc.list_maintenance_windows(db) w = pick(window, windows, kind="maintenance window", list_tool="list_maintenance_windows") if w.end_at <= now: raise ValueError(f"{w.name} already ended at {iso(w.end_at)}.") if w.start_at > now: await db.delete(w) outcome = "cancelled" else: w.end_at = now outcome = "ended" await audit_mcp(f"maintenance_window.{outcome}", "maintenance_window", w.name, {"window_id": w.id}) return {**_window_row(w, now), "outcome": outcome} def register(register_tool) -> None: for fn in (list_alerts, get_alert_history, list_alert_rules, list_maintenance_windows, get_alert_rule_options): register_tool(fn, role=UserRole.viewer, read_only=True) for fn in (create_alert_rule, update_alert_rule, acknowledge_alert, send_test_alert, create_maintenance_window, end_maintenance_window): register_tool(fn, role=UserRole.operator, read_only=False) register_tool(delete_alert_rule, role=UserRole.admin, read_only=False)