FabledSteward/steward/core/formatting.py

"""steward/core/formatting.py

How Steward writes a size, a rate, a span of time and a moment, and how it
grades a reading against Settings → Thresholds — once, for every page and
plugin (#5093). Registered as Jinja filters (|bytes, |rate, |duration, |ago)
and the status_class global in app.py; Python callers import them from here.
"""
from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

from markupsafe import Markup, escape

NONE = "—"
_UNITS = ("B", "KiB", "MiB", "GiB", "TiB", "PiB")
_SPANS = (("d", 86400), ("h", 3600), ("m", 60), ("s", 1))


def human_bytes(n: float | None) -> str:
    """Binary units: '0 B', '512 B', '1.5 KiB', '1.4 GiB'."""
    if n is None:
        return NONE
    size = float(n)
    for unit in _UNITS:
        if abs(size) < 1024 or unit == _UNITS[-1]:
            return f"{int(size)} B" if unit == "B" else f"{size:.1f} {unit}"
        size /= 1024
    raise AssertionError("unreachable")


def human_rate(bytes_per_s: float | None) -> str:
    """Bytes per second in binary units: '12.3 MiB/s'."""
    return NONE if bytes_per_s is None else f"{human_bytes(bytes_per_s)}/s"


def human_duration(seconds: float | None) -> str:
    """The two largest units of a span: '45s', '8m', '1m 30s', '2h 5m', '3d 4h'.
    The second unit is dropped when it is zero. Negative spans read as None."""
    if seconds is None or seconds < 0:
        return NONE
    left = int(seconds)
    if left == 0:
        return "0s"
    parts = []
    for i, (unit, size) in enumerate(_SPANS):
        count, left = divmod(left, size)
        if count:
            parts.append(f"{count}{unit}")
            nxt = _SPANS[i + 1] if i + 1 < len(_SPANS) else None
            if nxt and left // nxt[1]:
                parts.append(f"{left // nxt[1]}{nxt[0]}")
            break
    return " ".join(parts)


def _utc(when: datetime) -> datetime:
    return when.replace(tzinfo=timezone.utc) if when.tzinfo is None else when.astimezone(timezone.utc)


def seconds_since(when: datetime | None, now: datetime | None = None) -> float | None:
    """Seconds from `when` to now; a naive datetime is taken as UTC."""
    if when is None:
        return None
    return ((now or datetime.now(timezone.utc)) - _utc(when)).total_seconds()


def ago(when: datetime | None, now: datetime | None = None) -> Markup:
    """'3m ago' (or 'in 3m' for a future moment) in a <time> element whose
    title is the absolute UTC time, so a hover gives the exact moment."""
    if when is None:
        return Markup(NONE)
    when = _utc(when)
    secs = seconds_since(when, now)
    if abs(secs) < 1:
        text = "just now"
    elif secs > 0:
        text = f"{human_duration(secs)} ago"
    else:
        text = f"in {human_duration(-secs)}"
    return Markup('<time datetime="{}" title="{}">{}</time>').format(
        when.strftime("%Y-%m-%dT%H:%M:%SZ"), when.strftime("%Y-%m-%d %H:%M:%S UTC"), escape(text))


def status_class(value: Any, kind: str, thresholds: dict) -> str:
    """Grade a reading against Settings → Thresholds (to_thresholds_cfg):
    'ok', 'warn' or 'crit', or 'unknown' when there is no reading.

    A kind the thresholds don't name is a mistake in the caller, not a
    reading to colour as healthy, so it raises."""
    if value is None:
        return "unknown"
    th = thresholds[kind]
    warn, crit = th["warn"], th["crit"]
    if th.get("dir", "high") == "low":
        return "crit" if value < crit else "warn" if value < warn else "ok"
    return "crit" if value >= crit else "warn" if value >= warn else "ok"


# Every state a page shows, and the tone it is drawn in: ok (green), warn
# (amber), crit (red), info (blue) or dim. The status_badge and status_dot
# macros read this, so one state looks the same on every page.
STATES: dict[str, tuple[str, str]] = {
    "up": ("ok", "Up"),
    "ok": ("ok", "OK"),
    "resolved": ("ok", "Resolved"),
    "down": ("crit", "Down"),
    "firing": ("crit", "Firing"),
    "crit": ("crit", "Critical"),
    "warn": ("warn", "Warning"),
    "pending": ("warn", "Pending"),
    "degraded": ("warn", "Degraded"),
    "stale": ("warn", "Stale"),
    "no_data": ("warn", "No data"),
    "acknowledged": ("info", "Acknowledged"),
    "inactive": ("dim", "Inactive"),
    "paused": ("dim", "Paused"),
    "unknown": ("dim", "Unknown"),
    # An Ansible run (AnsibleRunStatus): working is blue, not amber, so a run
    # in progress doesn't read as a problem.
    "queued": ("dim", "Queued"),
    "running": ("info", "Running"),
    "success": ("ok", "Succeeded"),
    "failed": ("crit", "Failed"),
    "interrupted": ("warn", "Interrupted"),
    "cancelled": ("dim", "Cancelled"),
}


def status_info(state: Any) -> tuple[str, str]:
    """(tone, label) for a state name or enum. A state the map doesn't know
    is drawn dim under its own name rather than breaking the page."""
    key = str(getattr(state, "value", state) or "unknown").lower()
    return STATES.get(key, ("dim", key.replace("_", " ").capitalize()))


def required(value: Any, what: str) -> Any:
    """For a macro argument that must be given: raise if it is empty."""
    if value is None or not str(value).strip():
        raise ValueError(f"{what} is required")
    return value


_SEVERITY = {"crit": 2, "warn": 1}


def worst_reading(readings, thresholds: dict):
    """The reading furthest past its cutoff, as (label, value, grade), or None
    when every reading is within its cutoffs or missing.

    `readings` is (label, value, kind) triples, graded like status_class; the
    first of equally bad readings wins, so list them in the order a reader
    should hear about them."""
    worst, rank = None, 0
    for label, value, kind in readings:
        grade = status_class(value, kind, thresholds)
        if _SEVERITY.get(grade, 0) > rank:
            worst, rank = (label, value, grade), _SEVERITY[grade]
    return worst


def metric_value(name: str, value: Any) -> str:
    """A metric reading in its unit, from the name's suffix: _pct → 94%,
    _ms → 120 ms, _bytes → 1.2 GiB, _bps → its rate, _secs / seconds_ → a
    span, _c → 71°C, _days → 12 days; is_up / is_healthy read as yes or no.
    None (no reading) is "—"."""
    if value is None:
        return "—"
    n = name or ""
    if n in ("is_up", "is_healthy"):
        return {"is_up": ("up", "down"), "is_healthy": ("healthy", "unhealthy")}[n][0 if value >= 0.5 else 1]
    if n.endswith("_pct") or "_pct_" in n:
        return f"{value:.1f}%" if abs(value) < 10 else f"{value:.0f}%"
    if n.endswith("_ms"):
        return f"{value:.0f} ms"
    if n.endswith("_bytes"):
        return human_bytes(value)
    if n.endswith(("_bps", "bytes_rate")):
        return human_rate(value)
    if n.endswith("_secs") or n.startswith("seconds_"):
        return human_duration(value)
    if n.endswith("_c"):
        return f"{value:.0f}°C"
    if n.endswith("_days"):
        return f"{value:.0f} days"
    return f"{value:g}"


# What each alertable metric is called on a page, and what it measures. The
# key is the name a source records; a name not listed reads as itself with
# its underscores spaced (SNMP's OID labels are the operator's own words).
METRICS: dict[str, tuple[str, str]] = {
    "is_up": ("Up or down", "whether the check passed"),
    "is_healthy": ("Health check", "the container's own HEALTHCHECK result"),
    "response_ms": ("Response time", "how long the check took"),
    "open_fds": ("Open files", "files Steward itself has open"),
    "open_fds_pct": ("Open files (% of limit)", "files Steward has open, against its limit"),
    "request_rate": ("Requests per second", "requests through this router"),
    "error_rate_4xx_pct": ("4xx responses", "share of responses that were client errors"),
    "error_rate_5xx_pct": ("5xx responses", "share of responses that were server errors"),
    "latency_p50_ms": ("Response time (median)", "half of requests were faster than this"),
    "latency_p95_ms": ("Response time (95th percentile)", "95% of requests were faster than this"),
    "latency_p99_ms": ("Response time (99th percentile)", "99% of requests were faster than this"),
    "response_bytes_rate": ("Response throughput", "bytes per second sent back"),
    "cert_expiry_days": ("Certificate expires in", "days until the TLS certificate expires"),
    "latency_ms": ("WAN latency", "round trip from the gateway"),
    "total_clients": ("Connected clients", "devices on the network"),
    "cpu_pct": ("CPU", "CPU in use; for a container, 100% is one core"),
    "mem_pct": ("Memory (% of limit)", "memory in use, against the container's limit or the host's RAM"),
    "restart_count": ("Restarts", "times the container has restarted, in total"),
    "mem_used_pct": ("Memory used", "memory in use"),
    "mem_available_bytes": ("Memory available", "memory free for new work"),
    "swap_used_bytes": ("Swap used", "memory paged out to disk"),
    "disk_used_pct_worst": ("Fullest disk", "the most-full mount on the host"),
    "disk_used_pct": ("Disk used", "one mount (resource host:mount)"),
    "load_1m": ("Load (1 min)", "runnable processes, averaged over a minute"),
    "load_5m": ("Load (5 min)", "runnable processes, averaged over 5 minutes"),
    "load_15m": ("Load (15 min)", "runnable processes, averaged over 15 minutes"),
    "uptime_secs": ("Uptime", "time since the host booted"),
    "seconds_since_seen": ("Time since the agent reported", "over 3 minutes means the agent is offline"),
}

# Where a rule's metric comes from, as the picker and the rules list say it.
SOURCES: dict[str, str] = {
    "icmp": "Ping monitors", "tcp": "TCP monitors", "dns": "DNS monitors",
    "http": "HTTP monitors", "steward": "Steward itself", "host_agent": "Host agent",
    "docker": "Docker containers", "traefik": "Traefik routers", "unifi": "UniFi",
    "snmp": "SNMP devices",
}


def metric_label(name: str) -> str:
    entry = METRICS.get(name or "")
    return entry[0] if entry else (name or "").replace("_", " ")


def metric_about(name: str) -> str:
    entry = METRICS.get(name or "")
    return entry[1] if entry else ""


def source_label(source: str) -> str:
    return SOURCES.get(source or "", (source or "").replace("_", " "))


def metric_unit(name: str) -> str:
    """The unit a threshold for this metric is written in, from the same
    suffixes metric_value reads; "" for a count or a plain number."""
    n = name or ""
    if n in ("is_up", "is_healthy"):
        return ""
    if n.endswith("_pct") or "_pct_" in n:
        return "%"
    if n.endswith("_ms"):
        return "ms"
    if n.endswith("_bytes"):
        return "bytes"
    if n.endswith(("_bps", "bytes_rate")):
        return "bytes/s"
    if n.endswith("_secs") or n.startswith("seconds_"):
        return "seconds"
    if n.endswith("_c"):
        return "°C"
    if n.endswith("_days"):
        return "days"
    if n == "request_rate":
        return "req/s"
    return ""