FabledSteward/plugins/host_agent/sensors.py

"""Which of a host's temperature sensors are thermometers (#5783).

A board's spare thermistor inputs often have nothing wired to them, and the
kernel reports them anyway. No flag says so; their behaviour does. A
thermometer in a host warms as its CPU warms. An open input reads the chip's
own reference, which drifts the other way as the chip warms, or holds still:
VDNT-Util01's SYSTIN ran at r = −0.81 against its CPU package over a week,
while CPUTIN beside it ran at +0.83 (#5782).

So each sensor is judged against the host's CPU over its hourly averages, by
a scheduled task, and the verdicts are kept here for ingest and the pages to
read. Until a host has enough history, or when its CPU never moved enough to
tell, nothing is judged and routes.counts_as_hottest's value rule decides.
"""
from __future__ import annotations

import logging
import re
from datetime import datetime, timedelta
from math import sqrt

logger = logging.getLogger(__name__)

# Hours of history judged, the shared hours a verdict needs, and how far the
# CPU must have moved in them for "it didn't follow" to mean anything.
WINDOW = timedelta(days=7)
MIN_HOURS = 72
REFERENCE_SWING_C = 10.0
# Falling as the CPU warms, this strongly, is no thermometer's behaviour.
# A wired board or drive sensor may barely follow the CPU (r near 0), so
# only a clear negative counts, not a weak one.
FALLS_R = -0.5
# Held within this over days while the CPU moved: a constant, not a reading.
FLAT_C = 0.5

# The host's CPU, in order of preference: Intel coretemp's package, AMD
# k10temp's control/die/CCD readings, then the PECI agent.
_REFERENCES = (re.compile(r"^Package id \d+$", re.I), re.compile(r"^(Tctl|Tdie)$", re.I),
               re.compile(r"^Tccd\d+$", re.I), re.compile(r"^PECI Agent \d+$", re.I))

# {host name: {sensor label: why it isn't a thermometer}}, from the last refresh.
_unwired: dict[str, dict[str, str]] = {}
# {resource: {hour: average}} over the last WINDOW, and the hour the next
# refresh reads raw samples from. Reading the whole week took over three
# minutes on the live instance, so it is read once at startup, and each later
# refresh reads only the hours since the last one.
_hours: dict[str, dict[datetime, float]] = {}
_read_from: datetime | None = None


def unwired(host_name: str) -> dict[str, str]:
    """The sensors judged not to be thermometers on a host, with why."""
    return _unwired.get(host_name, {})


def reference_label(labels) -> str | None:
    """The sensor that reads the host's CPU, or None when it has none."""
    labels = sorted(labels)
    for pattern in _REFERENCES:
        found = [label for label in labels if pattern.match(label)]
        if found:
            return found[0]
    return None


def pearson(xs: list[float], ys: list[float]) -> float | None:
    """Pearson's r, or None when either series doesn't move."""
    n = len(xs)
    mx, my = sum(xs) / n, sum(ys) / n
    sxx = sum((x - mx) ** 2 for x in xs)
    syy = sum((y - my) ** 2 for y in ys)
    if sxx == 0 or syy == 0:
        return None
    return sum((x - mx) * (y - my) for x, y in zip(xs, ys)) / sqrt(sxx * syy)


def judge(reference: dict[datetime, float], sensor: dict[datetime, float]) -> str | None:
    """Why `sensor` isn't a thermometer, judged against the host's CPU over
    the hours both have; None when it behaves like one, or when there isn't
    enough to tell (too few hours, or a CPU that never moved)."""
    hours = sorted(set(reference) & set(sensor))
    if len(hours) < MIN_HOURS:
        return None
    cpu = [reference[h] for h in hours]
    got = [sensor[h] for h in hours]
    if max(cpu) - min(cpu) < REFERENCE_SWING_C:
        return None
    if max(got) - min(got) < FLAT_C:
        return f"holds at {sum(got) / len(got):.0f} °C while the CPU moved {min(cpu):.0f}–{max(cpu):.0f} °C"
    r = pearson(cpu, got)
    if r is not None and r <= FALLS_R:
        return f"falls as the CPU warms (r {r:.2f} over {len(hours)} h)"
    return None


def judge_hosts(rows) -> dict[str, dict[str, str]]:
    """{host: {label: why}} from hourly rows of (resource, bucket, value_avg),
    where resource is "<host>:temp:<label>"."""
    series: dict[str, dict[str, dict[datetime, float]]] = {}
    for resource, bucket, value in rows:
        host, sep, label = resource.partition(":temp:")
        if sep and value is not None:
            series.setdefault(host, {}).setdefault(label, {})[bucket] = value
    out: dict[str, dict[str, str]] = {}
    for host, sensors in series.items():
        ref = reference_label(sensors)
        if ref is None:
            continue
        for label, points in sensors.items():
            if label == ref:
                continue
            why = judge(sensors[ref], points)
            if why:
                out.setdefault(host, {})[label] = why
    return out


def merge_hours(hours: dict[str, dict[datetime, float]], rows, since: datetime) -> None:
    """Fold (resource, hour, average) rows into `hours`, a newer read of an
    hour replacing an older one (the hour that was still filling last time),
    and drop the hours older than `since`."""
    for resource, bucket, value in rows:
        if value is not None:
            hours.setdefault(resource, {})[bucket] = float(value)
    for resource in list(hours):
        kept = {h: v for h, v in hours[resource].items() if h >= since}
        if kept:
            hours[resource] = kept
        else:
            del hours[resource]


async def refresh(session, now: datetime) -> dict[str, dict[str, str]]:
    """Re-judge every host's sensors from the last WINDOW of hourly averages,
    and keep the verdicts for ingest and the pages.

    Hourly rollups exist only for hours older than the raw window
    (metrics.retention.raw_days, 7 days by default), so the recent hours are
    averaged from the raw samples here. Retention deletes the raw rows it
    rolls up, so the two never cover the same hour. The sensors are named
    from plugin_metric_latest, so each series is read by its index rather
    than by a pattern over every host_agent row.

    The first refresh reads the whole WINDOW; later ones read raw samples
    from the start of the hour the last one ended in, so that hour is
    re-read whole."""
    global _unwired, _read_from
    from sqlalchemy import func, select
    from steward.models.metrics import PluginMetric as M
    from steward.models.metrics import PluginMetricHourly as H
    from steward.models.metrics import PluginMetricLatest as L
    since = now - WINDOW
    names = (await session.execute(
        select(L.resource_name).where(L.source_module == "host_agent", L.metric_name == "temp_c"))).scalars().all()
    if not names:
        _unwired = {}
        return _unwired
    read_from = since if _read_from is None or _read_from < since else _read_from
    hour = func.date_trunc("hour", M.recorded_at)
    rows = list((await session.execute(
        select(M.resource_name, hour, func.avg(M.value))
        .where(M.source_module == "host_agent", M.metric_name == "temp_c",
               M.resource_name.in_(names), M.recorded_at >= read_from)
        .group_by(M.resource_name, hour))).all())
    if _read_from is None:
        rows += (await session.execute(
            select(H.resource_name, H.bucket, H.value_avg)
            .where(H.source_module == "host_agent", H.metric_name == "temp_c",
                   H.resource_name.in_(names), H.bucket >= since))).all()
    merge_hours(_hours, rows, since)
    _read_from = now.replace(minute=0, second=0, microsecond=0)
    _unwired = judge_hosts((r, h, v) for r, pts in _hours.items() for h, v in pts.items())
    return _unwired