bvandeusen
4771d17f6d
Rebuilds the deleted NUT/UPS automation as a general alert action: any metric can drive a playbook run on transition-to-firing. - models/alerts.py + migration 0014: AlertRule.ansible_action (JSON, admin-only, reuses the #546 param shape); AlertEvent.ansible_run_id links a firing event to the run it triggered - core/alerts.py: pure alert_extra_vars() injects steward_alert_* context; on ('firing', event) with an action set, schedule _run_ansible_action (deferred after commit, same pattern as notifications) — fires once per transition, consecutive_failures_required is the debounce; system-triggered AnsibleRun - alerts/routes.py: admin-only parse/validate of the action (source must be configured, playbook must exist); operators keep editing rules, action preserved - rules_form.html: admin-only 'On firing -> run a playbook' section - tests: unit for alert_extra_vars; integration drives record_metric to firing and asserts a system AnsibleRun ran the playbook with the injected var and the event linked to it Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
348 lines
12 KiB
Python
348 lines
12 KiB
Python
from __future__ import annotations
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
from typing import TYPE_CHECKING
|
|
|
|
from sqlalchemy import and_, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from steward.models.alerts import (
|
|
AlertEvent,
|
|
AlertOperator,
|
|
AlertRule,
|
|
AlertState,
|
|
AlertStateEnum,
|
|
MaintenanceWindow,
|
|
)
|
|
from steward.models.metrics import PluginMetric
|
|
|
|
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) -> list[str]:
|
|
"""Ansible extra-vars describing the alert that fired, as `key=value` strings.
|
|
|
|
Injected into every alert-triggered playbook run so the playbook can target
|
|
the exact resource/metric that fired without templating. Pure function.
|
|
"""
|
|
return [
|
|
f"steward_alert_rule={rule.name}",
|
|
f"steward_alert_source={rule.source_module}",
|
|
f"steward_alert_resource={rule.resource_name}",
|
|
f"steward_alert_metric={rule.metric_name}",
|
|
f"steward_alert_value={value}",
|
|
f"steward_alert_threshold={rule.threshold}",
|
|
f"steward_alert_operator={rule.operator.value}",
|
|
]
|
|
|
|
|
|
async def record_metric(
|
|
session: AsyncSession,
|
|
source_module: str,
|
|
resource_name: str,
|
|
metric_name: str,
|
|
value: float,
|
|
) -> None:
|
|
"""Write a metric row and evaluate alert rules inline.
|
|
|
|
Must be called inside an active transaction:
|
|
async with session.begin():
|
|
await record_metric(session, ...)
|
|
|
|
Notification I/O is deferred via asyncio.create_task() so no network
|
|
calls occur while the transaction is open.
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
|
|
# Write metric row
|
|
metric = PluginMetric(
|
|
source_module=source_module,
|
|
resource_name=resource_name,
|
|
metric_name=metric_name,
|
|
value=value,
|
|
recorded_at=now,
|
|
)
|
|
session.add(metric)
|
|
await session.flush()
|
|
|
|
# Check active maintenance windows — skip rule evaluation if suppressed
|
|
mw_result = await session.execute(
|
|
select(MaintenanceWindow).where(
|
|
MaintenanceWindow.start_at <= now,
|
|
MaintenanceWindow.end_at >= now,
|
|
or_(
|
|
MaintenanceWindow.scope_source.is_(None),
|
|
and_(
|
|
MaintenanceWindow.scope_source == source_module,
|
|
or_(
|
|
MaintenanceWindow.scope_resource.is_(None),
|
|
MaintenanceWindow.scope_resource == resource_name,
|
|
),
|
|
),
|
|
),
|
|
).limit(1)
|
|
)
|
|
if mw_result.scalar_one_or_none() is not None:
|
|
return # suppressed by maintenance window
|
|
|
|
# Load matching enabled rules
|
|
result = await session.execute(
|
|
select(AlertRule).where(
|
|
AlertRule.source_module == source_module,
|
|
AlertRule.resource_name == resource_name,
|
|
AlertRule.metric_name == metric_name,
|
|
AlertRule.enabled.is_(True),
|
|
)
|
|
)
|
|
rules = result.scalars().all()
|
|
|
|
for rule in rules:
|
|
state_result = await session.execute(
|
|
select(AlertState).where(AlertState.rule_id == rule.id)
|
|
)
|
|
alert_state = state_result.scalar_one_or_none()
|
|
if alert_state is None:
|
|
logger.warning(f"No alert_state row for rule {rule.id}, skipping")
|
|
continue
|
|
|
|
notifications = await _evaluate_rule(session, rule, alert_state, value, now)
|
|
|
|
for notif_type, event_id in notifications:
|
|
if _app is not None:
|
|
asyncio.create_task(
|
|
_dispatch_notification(
|
|
_app, notif_type, rule, 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, value, event_id)
|
|
)
|
|
|
|
|
|
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
|
|
|
|
|
|
async def _evaluate_rule(
|
|
session: AsyncSession,
|
|
rule: AlertRule,
|
|
state: AlertState,
|
|
value: float,
|
|
now: datetime,
|
|
) -> list[tuple[str, str]]:
|
|
"""Evaluate a single rule against a metric value. Returns list of (notif_type, event_id)."""
|
|
breached = _is_breached(value, rule.operator, rule.threshold)
|
|
from_state = state.state
|
|
notifications: list[tuple[str, str]] = []
|
|
|
|
if state.state == AlertStateEnum.inactive:
|
|
if breached:
|
|
state.consecutive_failure_count = 1
|
|
if state.consecutive_failure_count >= rule.consecutive_failures_required:
|
|
state.state = AlertStateEnum.firing
|
|
state.last_transition_at = now
|
|
event_id = _add_event(session, rule.id, from_state.value, "firing", value, now)
|
|
notifications.append(("firing", event_id))
|
|
else:
|
|
state.state = AlertStateEnum.pending
|
|
state.last_transition_at = now
|
|
|
|
elif state.state == AlertStateEnum.pending:
|
|
if breached:
|
|
state.consecutive_failure_count += 1
|
|
if state.consecutive_failure_count >= rule.consecutive_failures_required:
|
|
state.state = AlertStateEnum.firing
|
|
state.last_transition_at = now
|
|
event_id = _add_event(session, rule.id, from_state.value, "firing", value, now)
|
|
notifications.append(("firing", event_id))
|
|
else:
|
|
state.consecutive_failure_count = 0
|
|
state.state = AlertStateEnum.inactive
|
|
state.last_transition_at = now
|
|
|
|
elif state.state == AlertStateEnum.firing:
|
|
if not breached:
|
|
# RESOLVED is transient: record event, notify, then immediately go inactive
|
|
event_id = _add_event(session, rule.id, from_state.value, "resolved", value, now)
|
|
notifications.append(("resolved", event_id))
|
|
state.consecutive_failure_count = 0
|
|
state.state = AlertStateEnum.inactive
|
|
state.last_transition_at = now
|
|
|
|
elif state.state == AlertStateEnum.acknowledged:
|
|
if not breached:
|
|
event_id = _add_event(session, rule.id, from_state.value, "resolved", value, now)
|
|
notifications.append(("resolved", event_id))
|
|
state.consecutive_failure_count = 0
|
|
state.state = AlertStateEnum.inactive
|
|
state.last_transition_at = now
|
|
state.acknowledged_by = None
|
|
state.acknowledged_at = None
|
|
else:
|
|
state.consecutive_failure_count += 1
|
|
if state.consecutive_failure_count >= rule.consecutive_failures_required:
|
|
state.state = AlertStateEnum.firing
|
|
state.last_transition_at = now
|
|
event_id = _add_event(session, rule.id, from_state.value, "firing", value, now)
|
|
notifications.append(("firing", event_id))
|
|
|
|
await session.flush()
|
|
return notifications
|
|
|
|
|
|
def _add_event(
|
|
session: AsyncSession,
|
|
rule_id: str,
|
|
from_state: str,
|
|
to_state: str,
|
|
value: float,
|
|
now: datetime,
|
|
) -> 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,
|
|
)
|
|
session.add(event)
|
|
return event_id
|
|
|
|
|
|
async def _dispatch_notification(
|
|
app: "Quart",
|
|
notif_type: str,
|
|
rule: AlertRule,
|
|
value: float,
|
|
event_id: str,
|
|
) -> 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,
|
|
"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).
|
|
|
|
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.
|
|
"""
|
|
from steward.ansible import executor, sources as src_module
|
|
from steward.models.ansible import AnsibleRun, AnsibleRunStatus
|
|
|
|
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).
|
|
params: dict = {
|
|
"extra_vars": list(action.get("extra_vars") or []) + 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
|
|
|
|
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(
|
|
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)
|