feat(alerts): alert history page — transitions, deliveries, mutes and actions (#5066)
CI / lint (push) Successful in 3s
CI / integration (push) Successful in 50s
CI / unit (push) Successful in 1m7s
CI / publish (push) Successful in 1m4s

#5066 "Alert history page". /alerts/history lists every alert transition
in a range (1h–30d, the shared range picker), newest first, all rules or
one (?rule=). Each row: from → to, the value against the rule's condition
("no data" for a silent one), email and webhook sent or failed with the
error, "muted · <window>" (and "notified when the window ended") taken out
of the event note, other notes, and a link to the Ansible run an action
started. Range and rule changes swap only the rows (HTMX). Viewer role.

The Alerts page links each rule's history and the whole history. The query
moves to alert_svc.history, which MCP get_alert_history now uses too.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-06 14:06:04 -04:00
co-authored by Claude Opus 5.5
parent c9f5c049db
commit dc22a35e9d
8 changed files with 302 additions and 12 deletions
+45
View File
@@ -5,6 +5,7 @@ from steward.core.audit import log_audit
from sqlalchemy import select
from steward.auth.middleware import require_role
from steward.alerts import service as alert_svc
from steward.core.time_range import parse_range
from steward.ansible import sources as src_module
from steward.models.alerts import AlertRule, AlertOperator, MaintenanceWindow
from steward.models.users import UserRole
@@ -101,6 +102,50 @@ async def list_alerts():
return await render_template("alerts/list.html", active=active, rules=rules, operators=_OPERATORS)
HISTORY_LIMIT = 200
def _history_row(event, rule) -> dict:
"""One transition as the history page shows it."""
from steward.core.alerts import MUTED, RELEASED
notes, muted_by, released = [], None, False
for part in (event.note or "").split("; "):
if part.startswith(MUTED):
muted_by = part[len(MUTED):].strip() or "a maintenance window"
elif part == RELEASED:
released = True
elif part:
notes.append(part)
value = "no data" if event.metric_value is None else f"{event.metric_value:g}"
return {
"at": event.transitioned_at, "rule": rule,
"from_state": event.from_state, "to_state": event.to_state,
"value": value, "condition": f"{rule.operator.value} {rule.threshold:g}",
"notes": notes, "muted_by": muted_by, "released": released,
"deliveries": [(label, sent, error) for label, sent, error in (
("email", event.email_sent, event.email_error),
("webhook", event.webhook_sent, event.webhook_error)) if sent is not None],
"ansible_run_id": event.ansible_run_id,
}
@alerts_bp.get("/history")
@require_role(UserRole.viewer)
async def alert_history():
"""Every alert transition in a range, newest first; one rule's with ?rule=."""
since, range_key = parse_range(request.args.get("range") or "7d")
async with current_app.db_sessionmaker() as db:
rules = await alert_svc.list_rules(db)
chosen = next((r for r in rules if r.id == request.args.get("rule")), None)
rows = await alert_svc.history(db, since, chosen.id if chosen else None, HISTORY_LIMIT)
ctx = {"events": [_history_row(e, r) for e, r in rows[:HISTORY_LIMIT]],
"truncated": len(rows) > HISTORY_LIMIT, "limit": HISTORY_LIMIT,
"rules": rules, "chosen": chosen, "current_range": range_key}
if request.headers.get("HX-Request"):
return await render_template("alerts/_history_rows.html", **ctx)
return await render_template("alerts/history.html", **ctx)
@alerts_bp.get("/api/rule_fields")
@require_role(UserRole.viewer)
async def rule_fields_api():
+16 -1
View File
@@ -12,7 +12,7 @@ from datetime import datetime, timezone
from sqlalchemy import or_, select
from steward.models.alerts import (
AlertOperator, AlertRule, AlertState, AlertStateEnum, MaintenanceWindow,
AlertEvent, AlertOperator, AlertRule, AlertState, AlertStateEnum, MaintenanceWindow,
)
from steward.models.metrics import PluginMetric
from steward.models.monitors import MONITOR_TYPE_VALUES, Monitor
@@ -89,6 +89,21 @@ async def list_rules(db) -> list[AlertRule]:
return list((await db.execute(select(AlertRule).order_by(AlertRule.name))).scalars())
async def history(db, since: datetime, rule_id: str | None = None,
limit: int = 200) -> list[tuple[AlertEvent, AlertRule]]:
"""Alert transitions since `since`, newest first, with their rule; one
rule's when `rule_id` is given. Returns up to limit+1 rows so a caller can
tell the list was cut."""
stmt = (select(AlertEvent, AlertRule)
.join(AlertRule, AlertEvent.rule_id == AlertRule.id)
.where(AlertEvent.transitioned_at >= since)
.order_by(AlertEvent.transitioned_at.desc())
.limit(limit + 1))
if rule_id:
stmt = stmt.where(AlertEvent.rule_id == rule_id)
return [tuple(row) for row in (await db.execute(stmt)).all()]
async def list_maintenance_windows(db) -> list[MaintenanceWindow]:
"""Every window, latest start first."""
return list((await db.execute(