bvandeusenandClaude Opus 5.5
e51d969c43
A rule compared a reading with a fixed number, so Synology Disk 1's bad sectors going from 3,160 to 3,392 could say nothing: a count that only grows is soon past any line. "rose by" compares each reading with the one before it, kept on the alert's state and only for such a rule. A fall (a replaced disk) doesn't fire. On the rule form, through MCP, and worded in emails, the alert sentence, history and the timeline. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
292 lines
14 KiB
Python
292 lines
14 KiB
Python
"""Writing the timeline. Every source writes through record(), in the same
|
|
transaction as its own record, as the thing happens: the entry stands on its
|
|
own, so it is still there after the source's own row ages out.
|
|
|
|
from_alert, from_run and from_audit build the entry for one alert event,
|
|
finished Ansible run and audit row. The live writers and the backfill use the
|
|
same three, so an entry reads the same however it got there.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy import select
|
|
|
|
from . import catalog
|
|
from steward.models.timeline import TimelineEntry
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _entry(*, type: str, at: datetime, summary: str, area: str | None = None,
|
|
hosts=(), subject: str | None = None, detail: dict | None = None,
|
|
actor: str = "system", until: datetime | None = None,
|
|
ref: tuple[str, str] | None = None) -> TimelineEntry:
|
|
known = catalog.TYPES.get(type)
|
|
return TimelineEntry(
|
|
at=at, until=until, type=type,
|
|
area=area or (known.area if known else catalog.OTHER[0]),
|
|
hosts=sorted({h for h in hosts if h}), subject=subject[:255] if subject else None, summary=summary,
|
|
detail=detail or None, actor=(actor or "system")[:64],
|
|
ref_type=ref[0] if ref else None, ref_id=str(ref[1]) if ref else None)
|
|
|
|
|
|
def record(session, **fields) -> TimelineEntry:
|
|
"""Add one entry to `session`: type, at and summary, and optionally
|
|
area (defaults to its type's), hosts, subject, detail, actor, until and
|
|
ref, the (kind, id) of the full record it points to, which is at most
|
|
one entry."""
|
|
entry = _entry(**fields)
|
|
session.add(entry)
|
|
return entry
|
|
|
|
|
|
async def record_or_update(session, ref: tuple[str, str], **fields) -> TimelineEntry:
|
|
"""record() the entry for the record `ref` points to, or, when it has
|
|
one already, bring that one up to date: a source whose record changes
|
|
after it is written (a downtime that ends, an anomaly reviewed, a job
|
|
whose host came back late) keeps one entry for it, not one per change."""
|
|
entry = (await session.execute(select(TimelineEntry).where(
|
|
TimelineEntry.ref_type == ref[0], TimelineEntry.ref_id == str(ref[1])))).scalar_one_or_none()
|
|
if entry is None:
|
|
return record(session, ref=ref, **fields)
|
|
fresh = _entry(ref=ref, **fields)
|
|
for column in ("at", "until", "type", "area", "hosts", "subject", "summary", "detail", "actor"):
|
|
setattr(entry, column, getattr(fresh, column))
|
|
return entry
|
|
|
|
|
|
async def username(session, user_id: str | None) -> str | None:
|
|
if not user_id:
|
|
return None
|
|
from steward.models.users import User
|
|
user = await session.get(User, user_id)
|
|
return user.username if user else None
|
|
|
|
|
|
# ── Alerts ──────────────────────────────────────────────────────────────────
|
|
|
|
_MONITOR_SOURCES = ("icmp", "tcp", "dns", "http")
|
|
|
|
|
|
async def hosts_of_rule(session, source: str, resource: str) -> list[str]:
|
|
"""The host an alert rule's resource is on, by name, when it has one: an
|
|
agent series ("<host>" or "<host>:<part>"), a container ("<host>/<name>")
|
|
or a monitor tied to a host."""
|
|
from steward.core.alerts import host_of_resource
|
|
if source == "host_agent":
|
|
return [host_of_resource(source, resource) or resource.split(":", 1)[0]]
|
|
if source == "docker":
|
|
return [resource.split("/", 1)[0]]
|
|
if source in _MONITOR_SOURCES:
|
|
from steward.models.hosts import Host
|
|
from steward.models.monitors import Monitor
|
|
name = (await session.execute(
|
|
select(Host.name).join(Monitor, Monitor.host_id == Host.id)
|
|
.where(Monitor.type == source, Monitor.name == resource).limit(1))).scalar()
|
|
return [name] if name else []
|
|
return []
|
|
|
|
|
|
def _alert_type(from_state: str, to_state: str) -> str | None:
|
|
if to_state == "firing":
|
|
return "alert.fired"
|
|
if to_state == "resolved":
|
|
return "alert.resolved"
|
|
if to_state == "acknowledged":
|
|
return "alert.acknowledged"
|
|
if to_state == "no_data":
|
|
return "alert.silent"
|
|
if from_state == "no_data":
|
|
return "alert.resumed"
|
|
return None
|
|
|
|
|
|
def alert_summary(type_: str, rule, value: float | None) -> str:
|
|
from steward.alerts.service import _BINARY
|
|
from steward.core.formatting import metric_label, metric_value
|
|
metric, res = rule.metric_name, rule.resource_name
|
|
said = metric_label(metric)
|
|
# The reading, and on a firing the condition it broke.
|
|
if metric in _BINARY:
|
|
reading = breach = f"{res} is {metric_value(metric, value)}"
|
|
else:
|
|
op = getattr(rule.operator, "said", rule.operator)
|
|
reading = f"{said} on {res} is {metric_value(metric, value)}"
|
|
breach = f"{reading} ({op} {metric_value(metric, rule.threshold)})"
|
|
return {
|
|
"alert.fired": f"{rule.name} fired: {breach}" if value is not None else f"{rule.name} fired",
|
|
"alert.resolved": f"{rule.name} resolved" + (f": {reading}" if value is not None else ""),
|
|
"alert.acknowledged": f"{rule.name} acknowledged",
|
|
"alert.silent": f"{rule.name}: no {said} reading from {res}",
|
|
"alert.resumed": f"{rule.name}: {said} readings from {res} again",
|
|
}[type_]
|
|
|
|
|
|
async def from_alert(session, rule, event, actor: str | None = None,
|
|
resumed: bool = False) -> TimelineEntry | None:
|
|
"""The entry for one alert event (an AlertEvent row), or None for a
|
|
transition the timeline doesn't show (pending and back). `resumed` is a
|
|
series back from silence with the alert's state unchanged, which can be
|
|
an already-firing alert still firing."""
|
|
type_ = "alert.resumed" if resumed else _alert_type(event.from_state, event.to_state)
|
|
if type_ is None:
|
|
return None
|
|
return record(
|
|
session, type=type_, at=event.transitioned_at,
|
|
summary=alert_summary(type_, rule, event.metric_value),
|
|
area=catalog.area_of_metric(rule.source_module, rule.metric_name),
|
|
hosts=await hosts_of_rule(session, rule.source_module, rule.resource_name),
|
|
subject=rule.resource_name,
|
|
detail={"rule_id": rule.id, "rule": rule.name, "source": rule.source_module,
|
|
"metric": rule.metric_name, "value": event.metric_value, "note": event.note,
|
|
"url": f"/alerts/history?rule={rule.id}"},
|
|
actor=actor or await username(session, getattr(event, "actor_id", None)) or "system",
|
|
ref=("alert_event", event.id))
|
|
|
|
|
|
# ── Anomalies ───────────────────────────────────────────────────────────────
|
|
|
|
_REVIEWED = {"to_review": "", "real": ": a real problem", "normal": ": accepted as normal",
|
|
"lapsed": ": not reviewed"}
|
|
|
|
|
|
async def from_anomaly(session, anomaly) -> TimelineEntry:
|
|
"""The entry for one anomaly, written when it opens and brought up to
|
|
date as it recurs and is reviewed: its type follows its review."""
|
|
from steward.core.formatting import metric_label, metric_value
|
|
metric, res = anomaly.metric_name, anomaly.resource_name
|
|
times = f", {anomaly.occurrences} times" if anomaly.occurrences > 1 else ""
|
|
return await record_or_update(
|
|
session, ("metric_anomaly", anomaly.id),
|
|
type=f"anomaly.{anomaly.review}", at=anomaly.started_at, until=anomaly.ended_at,
|
|
summary=(f"{metric_label(metric)} on {res} out of its normal (peak "
|
|
f"{metric_value(metric, anomaly.peak_level)}, usual "
|
|
f"{metric_value(metric, anomaly.usual_hold)}{times}){_REVIEWED.get(anomaly.review, '')}"),
|
|
area=catalog.area_of_metric(anomaly.source_module, metric),
|
|
hosts=await hosts_of_rule(session, anomaly.source_module, res), subject=res,
|
|
detail={"source": anomaly.source_module, "metric": metric, "review": anomaly.review,
|
|
"note": anomaly.note, "url": "/alerts/anomalies"},
|
|
actor=anomaly.reviewed_by if anomaly.review in ("real", "normal") and anomaly.reviewed_by
|
|
else "Steward")
|
|
|
|
|
|
async def anomaly_changed(session, anomaly) -> None:
|
|
"""Write an anomaly's entry in a savepoint. Never raises: the anomaly's
|
|
own record matters more than its entry."""
|
|
await session.flush() # the anomaly's id, and its row before the savepoint
|
|
try:
|
|
async with session.begin_nested():
|
|
await from_anomaly(session, anomaly)
|
|
except Exception:
|
|
logger.exception("Anomaly %s: its timeline entry could not be written", anomaly.id)
|
|
|
|
|
|
# ── Ansible runs ────────────────────────────────────────────────────────────
|
|
|
|
def _run_type(run) -> str:
|
|
status = getattr(run.status, "value", run.status)
|
|
if status != "success":
|
|
return "ansible.failed"
|
|
if (run.params or {}).get("check"):
|
|
return "ansible.check"
|
|
hosts = ((run.results or {}).get("hosts") or {}).values()
|
|
return "ansible.changed" if any((h or {}).get("changed") for h in hosts) else "ansible.unchanged"
|
|
|
|
|
|
async def hosts_of_run(session, names) -> list[str]:
|
|
"""The Steward hosts behind a run's inventory names: a target linked to a
|
|
host is that host; a name that is a host's own is that host."""
|
|
from steward.models.ansible_inventory import AnsibleTarget
|
|
from steward.models.hosts import Host
|
|
names = set(names)
|
|
if not names:
|
|
return []
|
|
linked = dict((await session.execute(
|
|
select(AnsibleTarget.name, Host.name).join(Host, AnsibleTarget.host_id == Host.id)
|
|
.where(AnsibleTarget.name.in_(names)))).all())
|
|
direct = set((await session.execute(
|
|
select(Host.name).where(Host.name.in_(names - set(linked))))).scalars())
|
|
return sorted(set(linked.values()) | direct)
|
|
|
|
|
|
async def from_run(session, run) -> TimelineEntry:
|
|
"""The entry for one finished Ansible run."""
|
|
from steward.ansible.inventory_gen import run_title, scope_labels
|
|
from steward.ansible.origins import run_origins
|
|
type_ = _run_type(run)
|
|
per_host = (run.results or {}).get("hosts") or {}
|
|
scope = (await scope_labels(session, [run.inventory_scope])).get(run.inventory_scope)
|
|
title = run_title(run.playbook_path, scope)
|
|
status = getattr(run.status, "value", run.status)
|
|
changed = sum(1 for h in per_host.values() if (h or {}).get("changed"))
|
|
failed = sum(1 for h in per_host.values() if (h or {}).get("failed") or (h or {}).get("unreachable"))
|
|
if type_ == "ansible.failed":
|
|
said = {"cancelled": "cancelled", "interrupted": "stopped by a restart"}.get(status, "failed")
|
|
tail = f"{said} on {failed} host{'s' if failed != 1 else ''}" if failed else said
|
|
elif type_ == "ansible.check":
|
|
tail = f"dry run, would change {changed} host{'s' if changed != 1 else ''}" if changed \
|
|
else "dry run, nothing to change"
|
|
elif type_ == "ansible.changed":
|
|
tail = f"changed {changed} host{'s' if changed != 1 else ''}"
|
|
else:
|
|
tail = "nothing to change"
|
|
origin = (await run_origins(session, [run])).get(run.id) or {}
|
|
return record(
|
|
session, type=type_, at=run.started_at, until=run.finished_at,
|
|
summary=f"{title}: {tail}", hosts=await hosts_of_run(session, per_host),
|
|
subject=run.playbook_path,
|
|
detail={"status": status, "scope": scope, "origin": origin.get("text"),
|
|
"failures": ((run.results or {}).get("failures") or [])[:3],
|
|
"url": f"/ansible/runs/{run.id}"},
|
|
actor=origin.get("text") or "Steward", ref=("ansible_run", run.id))
|
|
|
|
|
|
async def run_finished(session, run_id: str) -> None:
|
|
"""Write the entry for a run that has just recorded its final status, in
|
|
a savepoint of the caller's transaction. Never raises: the run's own
|
|
record matters more than its entry."""
|
|
from steward.models.ansible import AnsibleRun
|
|
try:
|
|
async with session.begin_nested():
|
|
run = await session.get(AnsibleRun, run_id)
|
|
if run is not None:
|
|
await from_run(session, run)
|
|
except Exception:
|
|
logger.exception("Ansible run %s: its timeline entry could not be written", run_id)
|
|
|
|
|
|
# ── The audit log ───────────────────────────────────────────────────────────
|
|
|
|
async def from_audit(session, event, detail: dict | None) -> TimelineEntry | None:
|
|
"""The entry for one audit row (an AuditEvent), or None for an action the
|
|
timeline leaves to another writer (catalog.audit_type)."""
|
|
from steward.audit.service import action_text, entity_labels
|
|
type_ = catalog.audit_type(event.action)
|
|
if type_ is None:
|
|
return None
|
|
detail = detail or {}
|
|
entity = None
|
|
if event.entity_type and event.entity_id:
|
|
labels = await entity_labels(session, [{"event": event, "detail": detail}])
|
|
entity = labels.get((event.entity_type, str(event.entity_id)))
|
|
what = action_text(event.action)
|
|
if entity:
|
|
what = f"{what} {entity['label']}"
|
|
hosts = []
|
|
if event.entity_type == "host" and entity:
|
|
hosts = [entity["label"]]
|
|
elif event.entity_type == "monitor":
|
|
from steward.models.hosts import Host
|
|
from steward.models.monitors import Monitor
|
|
name = (await session.execute(
|
|
select(Host.name).join(Monitor, Monitor.host_id == Host.id)
|
|
.where(Monitor.id == str(event.entity_id)))).scalar()
|
|
hosts = [name] if name else []
|
|
return record(
|
|
session, type=type_, at=event.timestamp, summary=what[:1].upper() + what[1:],
|
|
hosts=hosts, subject=entity["label"] if entity else None,
|
|
detail={"action": event.action, "url": entity.get("url") if entity else None},
|
|
actor=event.username, ref=("audit_event", event.id))
|