feat(alerts): Send test on a rule checks that its alerts reach someone (#5466)
#5466 "Send test: check a rule's delivery from the rule". A rule's page gets a Send test button (operator role), and MCP gets send_test_alert. Each sends a TEST alert for that rule now, by email to everyone who gets alerts and by webhook. The alert carries the rule's name and condition, at the latest reading of what it watches; a rule covering hosts uses its first host. The result for each channel shows under the button, or "Nothing is set up to send to". Nothing fires, no event is written, and the test is audited as alert_rule.tested. An alert with no reading now says "has no reading yet" instead of "is None". Docs: alerting.md also gains the burst email from #5465 "One email per burst of alerts". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -6,7 +6,7 @@ from datetime import datetime, timedelta, timezone
|
||||
from sqlalchemy import select
|
||||
|
||||
from steward.alerts import service as alert_svc
|
||||
from steward.mcp._context import current_caller
|
||||
from steward.mcp._context import current_caller, mcp_app
|
||||
from steward.models.alerts import AlertOperator, AlertRule, AlertState, MaintenanceWindow
|
||||
from steward.models.users import UserRole
|
||||
|
||||
@@ -391,6 +391,25 @@ async def acknowledge_alert(rule: str) -> dict:
|
||||
"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"))
|
||||
@@ -467,7 +486,7 @@ 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,
|
||||
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)
|
||||
|
||||
Reference in New Issue
Block a user