FabledSteward/steward/timeline/record.py

"""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, "value", 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))