bvandeusenandClaude Opus 5.5
0945d78290
get_anomaly gives each occurrence against what is usual at that hour, the metric around the newest one with its normal, what else happened on the resource's host around each occurrence (alert transitions, Ansible runs, other anomalies there and a count elsewhere, plus plugins' own events through a new mcp_events hook: the host agent's maintenance jobs and downtime, Docker's container starts, deaths and OOM kills), and how earlier anomalies on the same metric were judged. get_metric_normal reads the learned slots and the level that would be flagged. review_anomaly takes one id or several, all or nothing, with a note. get_anomaly_settings and (admin) set_anomaly_settings read and change the review window and each normal's floor, as Settings → Thresholds does. metric_series.points takes an optional `until`; core.alerts gains host_of_resource; floor_setting names its floor. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
855 lines
35 KiB
Python
855 lines
35 KiB
Python
from __future__ import annotations
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import TYPE_CHECKING, Awaitable, Callable
|
|
|
|
from sqlalchemy import event as sa_event, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from steward.models.alerts import (
|
|
AlertEvent,
|
|
AlertOperator,
|
|
AlertRule,
|
|
AlertState,
|
|
AlertStateEnum,
|
|
MaintenanceWindow,
|
|
)
|
|
from steward.core.formatting import human_duration
|
|
from steward.core.watch import watch
|
|
from steward.models.metrics import PluginMetric, PluginMetricLatest
|
|
|
|
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
|
|
|
|
|
|
# ── One silent agent is one alert (#5721) ───────────────────────────────────
|
|
# A host's agent carries more than its own figures (its containers' too), so
|
|
# when it goes quiet every rule it feeds goes silent with it. That is one
|
|
# cause: "agent offline" says it, and the other rules' silence is recorded but
|
|
# not emailed, nor is their quiet return. A return past a threshold still fires.
|
|
|
|
AGENT_OFFLINE = "while its agent is offline"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Heartbeat:
|
|
source_module: str
|
|
metric_name: str
|
|
offline_after: Callable[[], float] # seconds; read at each use
|
|
|
|
|
|
_HEARTBEAT: Heartbeat | None = None
|
|
_AGENT_FED: dict[str, Callable[[str], str]] = {}
|
|
_AWAY: Callable[[AsyncSession], Awaitable[set[str]]] | None = None
|
|
|
|
|
|
def agent_heartbeat(source_module: str, metric_name: str,
|
|
offline_after: Callable[[], float]) -> None:
|
|
"""The metric that says how long each host's agent has been quiet; past
|
|
`offline_after()` the agent is offline. The host agent plugin registers it."""
|
|
global _HEARTBEAT
|
|
_HEARTBEAT = Heartbeat(source_module, metric_name, offline_after)
|
|
|
|
|
|
def agent_fed(source_module: str, host_of: Callable[[str], str]) -> None:
|
|
"""A source whose samples a host's agent carries; `host_of(resource)`
|
|
names the host ("nas:/" → "nas", "nas/web" → "nas")."""
|
|
_AGENT_FED[source_module] = host_of
|
|
|
|
|
|
def agents_away(hosts_away: Callable[[AsyncSession], Awaitable[set[str]]]) -> None:
|
|
"""`await hosts_away(session)` names the hosts whose agent said it was
|
|
stopping and is still inside its grace (#5756): their silence has a
|
|
reason, so their other rules go quiet with it as when it is offline."""
|
|
global _AWAY
|
|
_AWAY = hosts_away
|
|
|
|
|
|
def is_heartbeat(rule) -> bool:
|
|
"""Whether `rule` watches the agent heartbeat ("agent offline")."""
|
|
return _HEARTBEAT is not None and (rule.source_module, rule.metric_name) == (
|
|
_HEARTBEAT.source_module, _HEARTBEAT.metric_name)
|
|
|
|
|
|
def host_of_resource(source_module: str, resource: str) -> str | None:
|
|
"""The host whose agent carries this series, or None for a source no
|
|
agent feeds (a monitor, a Traefik router)."""
|
|
host_of = _AGENT_FED.get(source_module)
|
|
return host_of(resource) if host_of is not None else None
|
|
|
|
|
|
def host_behind(rule) -> str | None:
|
|
"""The host whose agent feeds `rule`, or None: a rule on another source,
|
|
or the heartbeat itself (agent offline is the one that speaks)."""
|
|
if is_heartbeat(rule):
|
|
return None
|
|
return host_of_resource(rule.source_module, rule.resource_name)
|
|
|
|
|
|
async def offline_agents(session) -> set[str]:
|
|
"""Hosts whose agent's last heartbeat was past its offline window, and
|
|
those whose agent said it was stopping (agents_away)."""
|
|
if _HEARTBEAT is None:
|
|
return set()
|
|
offline = set((await session.execute(
|
|
select(PluginMetricLatest.resource_name).where(
|
|
PluginMetricLatest.source_module == _HEARTBEAT.source_module,
|
|
PluginMetricLatest.metric_name == _HEARTBEAT.metric_name,
|
|
PluginMetricLatest.value > _HEARTBEAT.offline_after()))).scalars())
|
|
return offline | (await _AWAY(session) if _AWAY is not None else set())
|
|
|
|
|
|
async def _went_quiet_with_its_agent(session, rule_id: str) -> bool:
|
|
"""Whether a rule's latest silence was recorded while its agent was offline."""
|
|
note = (await session.execute(
|
|
select(AlertEvent.note).where(AlertEvent.rule_id == rule_id,
|
|
AlertEvent.to_state == "no_data")
|
|
.order_by(AlertEvent.transitioned_at.desc()).limit(1))).scalar_one_or_none()
|
|
return bool(note and AGENT_OFFLINE in note)
|
|
|
|
|
|
def alert_extra_vars(rule: AlertRule, value: float) -> dict:
|
|
"""Ansible extra-vars describing the alert that fired.
|
|
|
|
Injected into every alert-triggered playbook run so the playbook can target
|
|
the exact resource/metric that fired without templating. They reach
|
|
ansible-playbook in its JSON extra-vars file, not as `-e key=value`, which
|
|
Ansible splits on whitespace: a rule named "High CPU on web" stays one
|
|
value (#5157). Pure function.
|
|
"""
|
|
return {
|
|
"steward_alert_rule": rule.name,
|
|
"steward_alert_source": rule.source_module,
|
|
"steward_alert_resource": rule.resource_name,
|
|
"steward_alert_metric": rule.metric_name,
|
|
"steward_alert_value": value,
|
|
"steward_alert_threshold": rule.threshold,
|
|
"steward_alert_operator": rule.operator.value,
|
|
}
|
|
|
|
|
|
def action_vars(lines) -> dict:
|
|
"""An alert action's stored `key=value` lines as a map, split at the first
|
|
`=` so a value keeps its spaces, quotes and further `=`."""
|
|
out: dict = {}
|
|
for line in lines or ():
|
|
key, sep, val = str(line).partition("=")
|
|
if sep and key.strip():
|
|
out[key.strip()] = val
|
|
return out
|
|
|
|
|
|
async def record_metric(
|
|
session: AsyncSession,
|
|
source_module: str,
|
|
resource_name: str,
|
|
metric_name: str,
|
|
value: float,
|
|
) -> None:
|
|
"""One metric sample: record_metrics with a single row."""
|
|
await record_metrics(session, [PluginMetric(
|
|
source_module=source_module, resource_name=resource_name,
|
|
metric_name=metric_name, value=value)])
|
|
|
|
|
|
class MetricBatch:
|
|
"""Gathers one transaction's samples so they are recorded, and their alert
|
|
rules evaluated, in a single record_metrics call:
|
|
|
|
batch = MetricBatch()
|
|
batch.add(source_module="unifi", resource_name="wan", metric_name="is_up", value=1.0)
|
|
await batch.record(session)
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self.rows: list[PluginMetric] = []
|
|
|
|
def add(self, source_module: str, resource_name: str, metric_name: str, value: float,
|
|
recorded_at: datetime | None = None) -> None:
|
|
self.rows.append(PluginMetric(source_module=source_module, resource_name=resource_name,
|
|
metric_name=metric_name, value=float(value),
|
|
recorded_at=recorded_at))
|
|
|
|
async def record(self, session: AsyncSession) -> None:
|
|
rows, self.rows = self.rows, []
|
|
await record_metrics(session, rows)
|
|
|
|
|
|
def _in_window(w: MaintenanceWindow, source_module: str, resource_name: str) -> bool:
|
|
if w.scope_source is None:
|
|
return True
|
|
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 _within(txn, boundary) -> bool:
|
|
"""Whether `txn` (a savepoint, or None for the outer transaction) is
|
|
`boundary` or opened inside it."""
|
|
while txn is not None:
|
|
if txn is boundary:
|
|
return True
|
|
txn = txn.parent
|
|
return False
|
|
|
|
|
|
def _after_commit(session: AsyncSession, fn, *args) -> None:
|
|
"""Run fn(app, *args) once the session's outer transaction commits, and
|
|
never if what added it rolls back: a notification or a playbook run must
|
|
describe what was actually stored.
|
|
|
|
SQLAlchemy fires after_commit and after_rollback for a savepoint too
|
|
(#5807). Releasing one, as the host-agent ingest does after its metrics,
|
|
is not the commit: a job started then raced the real commit, and an email
|
|
queued for an event not yet visible broke its foreign key. A savepoint
|
|
rolling back drops only the jobs added inside it."""
|
|
if _app is None:
|
|
return
|
|
info = session.info
|
|
sync = session.sync_session
|
|
if _AFTER_COMMIT not in info:
|
|
info[_AFTER_COMMIT] = []
|
|
app = _app
|
|
|
|
def _committed(sync_session) -> None:
|
|
if sync_session.get_nested_transaction() is not None:
|
|
return # a savepoint released: its jobs wait for the outer commit
|
|
jobs, info[_AFTER_COMMIT] = info[_AFTER_COMMIT], []
|
|
for _owner, job_fn, job_args in jobs:
|
|
_spawn(job_fn(app, *job_args))
|
|
|
|
def _rolled_back(_sync_session, previous_transaction) -> None:
|
|
# What was actually rolled back: the nearest savepoint, or the
|
|
# outer transaction (a failed flush rolls back to one of them).
|
|
boundary = previous_transaction
|
|
while boundary.parent is not None and not boundary.nested:
|
|
boundary = boundary.parent
|
|
if not boundary.nested:
|
|
info[_AFTER_COMMIT] = []
|
|
else:
|
|
info[_AFTER_COMMIT] = [job for job in info[_AFTER_COMMIT]
|
|
if not _within(job[0], boundary)]
|
|
|
|
sa_event.listen(sync, "after_commit", _committed)
|
|
sa_event.listen(sync, "after_soft_rollback", _rolled_back)
|
|
info[_AFTER_COMMIT].append((sync.get_nested_transaction(), fn, args))
|
|
|
|
|
|
LATEST_CHUNK = 1000
|
|
|
|
|
|
async def _keep_latest(session: AsyncSession, rows: list[PluginMetric]) -> None:
|
|
"""Move each key's row in plugin_metric_latest to the newest sample in
|
|
`rows`. A row only moves forward, so a backlog arriving after newer samples
|
|
leaves it alone. Keys go in sorted order, so two batches sharing keys lock
|
|
them in the same order and cannot deadlock."""
|
|
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
|
|
|
newest: dict[tuple[str, str, str], PluginMetric] = {}
|
|
for row in rows:
|
|
key = (row.source_module, row.resource_name, row.metric_name)
|
|
if key not in newest or row.recorded_at > newest[key].recorded_at:
|
|
newest[key] = row
|
|
values = [{"source_module": k[0], "resource_name": k[1], "metric_name": k[2],
|
|
"value": newest[k].value, "recorded_at": newest[k].recorded_at}
|
|
for k in sorted(newest)]
|
|
# Five parameters a row; chunks keep a statement well under asyncpg's 32767.
|
|
for i in range(0, len(values), LATEST_CHUNK):
|
|
stmt = pg_insert(PluginMetricLatest).values(values[i:i + LATEST_CHUNK])
|
|
await session.execute(stmt.on_conflict_do_update(
|
|
index_elements=["source_module", "resource_name", "metric_name"],
|
|
set_={"value": stmt.excluded.value, "recorded_at": stmt.excluded.recorded_at},
|
|
where=PluginMetricLatest.recorded_at <= stmt.excluded.recorded_at,
|
|
))
|
|
|
|
|
|
async def record_metrics(session: AsyncSession, rows: list[PluginMetric], phases=None) -> None:
|
|
"""Write metric rows and evaluate the alert rules that watch them.
|
|
|
|
Every ingest path comes through here, so a rule on any metric Steward
|
|
records can fire. Must be called inside an active transaction:
|
|
async with session.begin():
|
|
await record_metrics(session, rows)
|
|
|
|
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. 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.
|
|
|
|
`phases`, when given, is marked after each step (flush, rules, evaluate)
|
|
so a slow caller can say which one its time went to.
|
|
"""
|
|
mark = phases.mark if phases is not None else (lambda _name: None)
|
|
if not rows:
|
|
return
|
|
now = datetime.now(timezone.utc)
|
|
for row in rows:
|
|
if row.recorded_at is None:
|
|
row.recorded_at = now
|
|
session.add_all(rows)
|
|
await session.flush()
|
|
await _keep_latest(session, rows)
|
|
mark("flush")
|
|
|
|
by_key: dict[tuple[str, str, str], list[PluginMetric]] = {}
|
|
for row in rows:
|
|
by_key.setdefault((row.source_module, row.resource_name, row.metric_name), []).append(row)
|
|
sources = {k[0] for k in by_key}
|
|
metric_names = {k[2] for k in by_key}
|
|
|
|
rules = (await session.execute(
|
|
select(AlertRule).where(
|
|
AlertRule.source_module.in_(sources),
|
|
AlertRule.metric_name.in_(metric_names),
|
|
AlertRule.enabled.is_(True),
|
|
)
|
|
)).scalars().all()
|
|
rules = [r for r in rules if (r.source_module, r.resource_name, r.metric_name) in by_key]
|
|
if not rules:
|
|
mark("rules")
|
|
return
|
|
|
|
windows = (await session.execute(
|
|
select(MaintenanceWindow).where(
|
|
MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now)
|
|
)).scalars().all()
|
|
states = {st.rule_id: st for st in (await session.execute(
|
|
select(AlertState).where(AlertState.rule_id.in_([r.id for r in rules]))
|
|
)).scalars()}
|
|
mark("rules")
|
|
|
|
for rule in rules:
|
|
alert_state = states.get(rule.id)
|
|
if alert_state is None:
|
|
# A restored rule (backups leave state out); alert_reconcile adds it.
|
|
logger.warning("No alert_state row for rule %s yet, skipping", rule.id)
|
|
continue
|
|
samples = sorted(by_key[(rule.source_module, rule.resource_name, rule.metric_name)],
|
|
key=lambda r: r.recorded_at)
|
|
muted = _muted_by(windows, rule)
|
|
resumed = _track_samples(alert_state, samples, now)
|
|
# A silence ends when Steward hears of it: an agent's flushed backlog
|
|
# brings samples hours old, and stamping their events at their own
|
|
# times put the resume inside the silence it ends (#5720).
|
|
back = resumed is not None
|
|
for sample in samples:
|
|
at = max(sample.recorded_at, now) if back else sample.recorded_at
|
|
notifications = await _evaluate_rule(session, rule, alert_state, sample.value,
|
|
at, note=muted, resumed=resumed)
|
|
resumed = None # said once, with the first sample back
|
|
if muted:
|
|
continue
|
|
for notif_type, event_id, note in notifications:
|
|
if notif_type == "data_resumed" and not rule.notify_no_data:
|
|
continue
|
|
# Back with its agent: the agent's return is the news (#5721).
|
|
if notif_type == "data_resumed" and host_behind(rule) is not None \
|
|
and await _went_quiet_with_its_agent(session, rule.id):
|
|
continue
|
|
_after_commit(session, _dispatch_notification, notif_type, rule, sample.value,
|
|
event_id, note)
|
|
# 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)
|
|
mark("evaluate")
|
|
|
|
|
|
# A series is silent once it has gone this long without a sample, or three of
|
|
# its usual gaps if that is longer. Until its cadence is known (fewer than two
|
|
# samples seen), it gets an hour: a monitor may check only every ten minutes.
|
|
NO_DATA_FLOOR_S = 300.0
|
|
NO_DATA_GAPS = 3
|
|
NO_DATA_UNKNOWN_S = 3600.0
|
|
|
|
|
|
def next_interval(interval: float | None, gap: float) -> float:
|
|
"""A series' usual gap between samples, updated with one more gap.
|
|
|
|
A smoothed average, so one late sample barely moves it. A gap longer than
|
|
the silence threshold counts only as that threshold: an outage is not the
|
|
series' cadence. A real change of cadence still gets learned, a step at a
|
|
time (each step at most 1.5x).
|
|
"""
|
|
gap = max(0.0, gap)
|
|
if interval is None or interval <= 0:
|
|
return gap
|
|
return 0.75 * interval + 0.25 * min(gap, NO_DATA_GAPS * interval)
|
|
|
|
|
|
def no_data_after(interval: float | None) -> float:
|
|
"""Seconds without a sample before a series counts as silent."""
|
|
if not interval:
|
|
return NO_DATA_UNKNOWN_S
|
|
return max(NO_DATA_FLOOR_S, NO_DATA_GAPS * interval)
|
|
|
|
|
|
def silent_since(last_sample_at: datetime | None, interval: float | None,
|
|
watching_since: datetime, now: datetime,
|
|
awake_since: datetime | None = None) -> datetime | None:
|
|
"""When a series went silent, or None while it is still reporting.
|
|
|
|
A series that has never had a sample counts from when the rule started
|
|
watching it (created, changed or re-enabled). Nothing counts from before
|
|
`awake_since`, when Steward last started watching without a break: no
|
|
sample can arrive while Steward is down, so its own outage is not the
|
|
series' silence (#5441).
|
|
"""
|
|
last = last_sample_at or watching_since
|
|
if awake_since is not None:
|
|
last = max(last, awake_since)
|
|
limit = timedelta(seconds=no_data_after(interval))
|
|
return last + limit if now - last > limit else None
|
|
|
|
|
|
def _track_samples(state: AlertState, samples: list[PluginMetric], now: datetime) -> str | None:
|
|
"""Note a rule's new samples (oldest first) for silence tracking. When the
|
|
series had been marked no_data, clear the mark and return the note saying
|
|
so ("data resumed after 15m silent"); the first sample's evaluation
|
|
records it, in one event with whatever that sample does to the rule
|
|
(#5442).
|
|
|
|
A resumed series counts as last heard `now`, when Steward heard it, not at
|
|
its newest sample's time: a backlog flushed a few samples a post is hours
|
|
old, and judging silence by it marked the series silent again between
|
|
posts, so one silence was resumed twice (#5720). The rest of the backlog
|
|
then reads as late samples, which teach the cadence nothing."""
|
|
for sample in samples:
|
|
if state.last_sample_at is not None:
|
|
if sample.recorded_at <= state.last_sample_at:
|
|
continue # late or duplicate: not a gap
|
|
gap = (sample.recorded_at - state.last_sample_at).total_seconds()
|
|
state.sample_interval_s = next_interval(state.sample_interval_s, gap)
|
|
state.last_sample_at = sample.recorded_at
|
|
if state.no_data_since is None or not samples:
|
|
return None
|
|
note = f"data resumed after {human_duration((now - state.no_data_since).total_seconds())} silent"
|
|
state.no_data_since = None
|
|
state.last_sample_at = max(state.last_sample_at or now, now)
|
|
return note
|
|
|
|
|
|
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
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Step:
|
|
"""One sample's effect on an alert: the state after it, the two run
|
|
counters, and the event to record ("firing" / "resolved") if any."""
|
|
state: AlertStateEnum
|
|
fail_count: int
|
|
ok_count: int
|
|
event: str | None = None
|
|
|
|
|
|
def next_step(state: AlertStateEnum, fail_count: int, ok_count: int, breached: bool,
|
|
fail_required: int, ok_required: int) -> Step:
|
|
"""The alert state machine, as a pure function of one sample.
|
|
|
|
inactive → pending → firing when `fail_required` breaching checks run in a
|
|
row; a non-breaching check while pending goes back to inactive.
|
|
firing and acknowledged resolve (to inactive, with a resolved event) only
|
|
after `ok_required` non-breaching checks in a row; a breaching check
|
|
resets that run. An acknowledgement holds while the alert is still
|
|
breached: only a resolve clears it.
|
|
"""
|
|
fail_required, ok_required = max(1, fail_required), max(1, ok_required)
|
|
S = AlertStateEnum
|
|
if state in (S.firing, S.acknowledged):
|
|
if breached:
|
|
return Step(state, fail_count + 1, 0)
|
|
if ok_count + 1 >= ok_required:
|
|
return Step(S.inactive, 0, 0, "resolved")
|
|
return Step(state, fail_count, ok_count + 1)
|
|
# inactive, pending (and the transient resolved, read as inactive)
|
|
if not breached:
|
|
return Step(S.inactive, 0, 0)
|
|
fails = fail_count + 1 if state == S.pending else 1
|
|
if fails >= fail_required:
|
|
return Step(S.firing, fails, 0, "firing")
|
|
return Step(S.pending, fails, 0)
|
|
|
|
|
|
async def _evaluate_rule(
|
|
session: AsyncSession,
|
|
rule: AlertRule,
|
|
state: AlertState,
|
|
value: float,
|
|
now: datetime,
|
|
note: str | None = None,
|
|
resumed: str | None = None,
|
|
) -> list[tuple[str, str, str | None]]:
|
|
"""Apply one sample to a rule's state. Returns [(notif_type, event_id,
|
|
note)]; an event it records carries `note`.
|
|
|
|
`resumed` says this sample ends a silence. The rule then gets one event
|
|
from no_data: to the state the sample moved it to (a "firing"
|
|
notification), or back to the state it was in ("data_resumed"). Two
|
|
events at one instant used to sort either way round, so the history
|
|
could read "back to inactive" right after "firing" (#5442).
|
|
"""
|
|
from_state = state.state
|
|
step = next_step(from_state, state.consecutive_failure_count, state.consecutive_ok_count,
|
|
_is_breached(value, rule.operator, rule.threshold),
|
|
rule.consecutive_failures_required, rule.consecutive_ok_required)
|
|
state.consecutive_failure_count = step.fail_count
|
|
state.consecutive_ok_count = step.ok_count
|
|
notifications: list[tuple[str, str, str | None]] = []
|
|
if step.state != from_state:
|
|
state.state = step.state
|
|
state.last_transition_at = now
|
|
if step.state == AlertStateEnum.inactive:
|
|
state.acknowledged_by = None
|
|
state.acknowledged_at = None
|
|
said = "; ".join(n for n in (resumed, note) if n) or None
|
|
if step.event:
|
|
event_id = _add_event(session, rule.id, "no_data" if resumed else from_state.value,
|
|
step.event, value, now, note=said)
|
|
notifications.append((step.event, event_id, None))
|
|
elif resumed:
|
|
event_id = _add_event(session, rule.id, "no_data", state.state.value, value, now, note=said)
|
|
notifications.append(("data_resumed", event_id, resumed))
|
|
await session.flush()
|
|
return notifications
|
|
|
|
|
|
async def reset_rule_state(session: AsyncSession, rule: AlertRule, note: str,
|
|
actor_id: str | None = None) -> None:
|
|
"""Put a rule's alert back to inactive because the rule itself changed
|
|
(what it watches, its threshold, or it was disabled or re-enabled), and
|
|
start its silence watch over. An alert that was
|
|
firing or acknowledged gets a resolved event carrying `note` and the
|
|
person who changed the rule (`actor_id`); nobody is notified, since
|
|
nothing about the watched resource changed."""
|
|
state = (await session.execute(
|
|
select(AlertState).where(AlertState.rule_id == rule.id))).scalar_one_or_none()
|
|
if state is None:
|
|
return
|
|
now = datetime.now(timezone.utc)
|
|
if state.state in (AlertStateEnum.firing, AlertStateEnum.acknowledged):
|
|
_add_event(session, rule.id, state.state.value, "resolved", None, now, note=note,
|
|
actor_id=actor_id)
|
|
# The watch starts over now: silence is measured from here, against the
|
|
# cadence of whatever the rule watches from now on.
|
|
state.last_transition_at = now
|
|
state.last_sample_at = None
|
|
state.sample_interval_s = None
|
|
state.no_data_since = None
|
|
state.state = AlertStateEnum.inactive
|
|
state.consecutive_failure_count = 0
|
|
state.consecutive_ok_count = 0
|
|
state.acknowledged_by = None
|
|
state.acknowledged_at = None
|
|
|
|
|
|
def _add_event(
|
|
session: AsyncSession,
|
|
rule_id: str,
|
|
from_state: str,
|
|
to_state: str,
|
|
value: float | None,
|
|
now: datetime,
|
|
note: str | None = None,
|
|
actor_id: str | None = None,
|
|
) -> 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,
|
|
note=note,
|
|
actor_id=actor_id,
|
|
)
|
|
session.add(event)
|
|
return event_id
|
|
|
|
|
|
async def _dispatch_notification(
|
|
app: "Quart",
|
|
notif_type: str,
|
|
rule: AlertRule,
|
|
value: float | None,
|
|
event_id: str,
|
|
note: str | None = None,
|
|
) -> None:
|
|
"""Run after the transaction commits. Queues the alert's email in the
|
|
outbox, gathered for 2 minutes with any others (core.outbox.hold_alert),
|
|
posts its webhook, and records the webhook's outcome on the event; the
|
|
outbox records the email's once it settles."""
|
|
from steward.core.notifications import dispatch_notifications
|
|
|
|
alert_data = {
|
|
"rule_name": rule.name,
|
|
"state": notif_type.upper(),
|
|
"metric": rule.metric_name,
|
|
"value": value,
|
|
"operator": rule.operator.value,
|
|
"threshold": rule.threshold,
|
|
"resource": rule.resource_name,
|
|
"source_module": rule.source_module,
|
|
"note": note,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
}
|
|
|
|
# Email waits so alerts that arrive together go out as one message per
|
|
# recipient (#5465, #5507); the webhook posts this alert now.
|
|
from steward.core import outbox
|
|
if app.config.get("SMTP", {}).get("host"):
|
|
try:
|
|
await outbox.hold_alert(app, alert_data, event_id)
|
|
except Exception:
|
|
logger.exception("Alert %s: its email could not be queued", event_id)
|
|
results = await dispatch_notifications(app.config.get("WEBHOOK", {}), alert_data)
|
|
|
|
webhook_r = results.get("webhook")
|
|
if not webhook_r:
|
|
return
|
|
async with app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
event = (await db.execute(
|
|
select(AlertEvent).where(AlertEvent.id == event_id))).scalar_one_or_none()
|
|
if event:
|
|
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).
|
|
|
|
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.runner import IN_FLIGHT, trigger_run
|
|
from steward.models.ansible import AnsibleRun
|
|
|
|
if rule.id in _actions_starting:
|
|
await _note_event(app, event_id, "Ansible action skipped: an earlier "
|
|
"firing's action was still starting")
|
|
return
|
|
async with app.db_sessionmaker() as db:
|
|
busy = (await db.execute(
|
|
select(AnsibleRun.id)
|
|
.join(AlertEvent, AlertEvent.ansible_run_id == AnsibleRun.id)
|
|
.where(AlertEvent.rule_id == rule.id, AnsibleRun.status.in_(IN_FLIGHT))
|
|
.limit(1))).scalar_one_or_none()
|
|
if busy:
|
|
# One action run at a time per rule (#5158): a re-firing while the
|
|
# last one is still going would run the same fix over it.
|
|
await _note_event(app, event_id, f"Ansible action skipped: run {busy} "
|
|
"from an earlier firing was still going")
|
|
return
|
|
_actions_starting.add(rule.id)
|
|
try:
|
|
await _start_ansible_action(app, rule, value, event_id, trigger_run)
|
|
finally:
|
|
_actions_starting.discard(rule.id)
|
|
|
|
|
|
# Rules whose action is between the in-flight check and its run being linked
|
|
# to the event, so two firings that close together can't both start one.
|
|
_actions_starting: set[str] = set()
|
|
|
|
|
|
async def _note_event(app: "Quart", event_id: str, why: str) -> None:
|
|
async with app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
ev = (await db.execute(
|
|
select(AlertEvent).where(AlertEvent.id == event_id))).scalar_one_or_none()
|
|
if ev is not None:
|
|
ev.note = f"{ev.note}; {why}" if ev.note else why
|
|
|
|
|
|
async def _start_ansible_action(app, rule, value, event_id, trigger_run) -> None:
|
|
action = rule.ansible_action or {}
|
|
playbook = action.get("playbook") or ""
|
|
inventory = action.get("inventory") or ""
|
|
source_name = action.get("source") or ""
|
|
# One JSON extra-vars file; the firing's own vars win over the action's.
|
|
params: dict = {
|
|
"extra_vars_map": {**action_vars(action.get("extra_vars")),
|
|
**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
|
|
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,
|
|
origin="alert", origin_id=rule.id)
|
|
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)
|
|
|
|
if run is None:
|
|
await _note_event(app, event_id, f"Ansible action not started: {err}")
|
|
return
|
|
async with app.db_sessionmaker() as db:
|
|
async with db.begin():
|
|
ev = (await db.execute(
|
|
select(AlertEvent).where(AlertEvent.id == event_id))).scalar_one_or_none()
|
|
if ev is not None:
|
|
ev.ansible_run_id = run.id
|
|
|
|
|
|
async def check_silence(app: "Quart") -> None:
|
|
"""Mark the enabled rules whose series has gone silent (no_data).
|
|
|
|
Run every minute by the core scheduler. A rule is silent once its series
|
|
has had no sample for no_data_after(its usual gap); one that never had a
|
|
sample counts from when the rule started watching. The mark sits beside
|
|
the state, so a firing alert that goes silent stays firing. Each silence
|
|
is recorded once, as an event, and notified once (rule.notify_no_data,
|
|
outside maintenance windows), after the transaction commits.
|
|
"""
|
|
now = datetime.now(timezone.utc)
|
|
notify: list[tuple[AlertRule, str, str]] = []
|
|
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.no_data_since.is_(None))
|
|
)).all()
|
|
awake_since = watch(app, "alert_silence").tick(now)
|
|
if not rows:
|
|
return
|
|
windows = (await session.execute(
|
|
select(MaintenanceWindow).where(
|
|
MaintenanceWindow.start_at <= now, MaintenanceWindow.end_at >= now)
|
|
)).scalars().all()
|
|
offline = await offline_agents(session) if _AGENT_FED else set()
|
|
for state, rule in rows:
|
|
watching_since = max(rule.created_at, state.last_transition_at)
|
|
since = silent_since(state.last_sample_at, state.sample_interval_s,
|
|
watching_since, now, awake_since)
|
|
if since is None:
|
|
continue
|
|
state.no_data_since = since
|
|
if state.last_sample_at is None:
|
|
note = (f"no sample yet, {human_duration((now - watching_since).total_seconds())}"
|
|
" after the rule started watching")
|
|
else:
|
|
note = f"no sample for {human_duration((now - state.last_sample_at).total_seconds())}"
|
|
if state.sample_interval_s:
|
|
note += f" (usually every {human_duration(state.sample_interval_s)})"
|
|
muted = _muted_by(windows, rule)
|
|
# Silent with its agent: recorded, and "agent offline" is the email (#5721).
|
|
hushed = AGENT_OFFLINE if host_behind(rule) in offline else None
|
|
event_id = _add_event(session, rule.id, state.state.value, "no_data", None, now,
|
|
note="; ".join(n for n in (note, hushed, muted) if n))
|
|
if rule.notify_no_data and not muted and not hushed:
|
|
notify.append((rule, event_id, note))
|
|
for rule, event_id, note in notify:
|
|
_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))
|