perf(host_agent): the sensor judgment reads the week once, then only new hours (#5786)

On the live instance the hourly refresh took 3 min 8 s, averaging a week of
raw temp_c samples each time. The hourly averages are now kept in memory.
The first refresh after startup reads the whole window. Each later one reads
raw samples from the start of the hour the last read ended in, so the hour
that was still filling is read again whole. merge_hours folds the new rows
in and drops hours older than the window.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

plugins/host_agent/sensors.py

@@ -40,6 +40,12 @@ _REFERENCES = (re.compile(r"^Package id \d+$", re.I), re.compile(r"^(Tctl|Tdie)$
 
 # {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]:
@@ -109,6 +115,21 @@ def judge_hosts(rows) -> dict[str, dict[str, str]]:
     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.
@@ -118,8 +139,12 @@ async def refresh(session, now: datetime) -> dict[str, dict[str, str]]:
     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."""
-    global _unwired
+    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
@@ -130,15 +155,19 @@ async def refresh(session, now: datetime) -> dict[str, dict[str, str]]:
     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)
-    raw = (await session.execute(
+    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 >= since)
-        .group_by(M.resource_name, hour))).all()
-    rolled = (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()
-    _unwired = judge_hosts([*rolled, *raw])
+               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

tests/integration/test_host_metrics.py

@@ -208,6 +208,16 @@ def test_sensors_are_judged_from_raw_hours_and_rollups_together(app):
                                              metric_name="temp_c", value=f(99), recorded_at=now))
             return await sensors.refresh(s, now)
 
+    sensors._hours.clear()
+    sensors._read_from = None
     got = asyncio.run(_go())
     assert list(got[host]) == ["SYSTIN"]
     assert got[host]["SYSTIN"].startswith("falls as the CPU warms (r -1.00 over 100 h")
+
+    async def _again():                        # the next hour reads only the hour it ended in
+        async with app.db_sessionmaker() as s:
+            return await sensors.refresh(s, now + timedelta(hours=1))
+
+    assert sensors._read_from == now.replace(minute=0)
+    again = asyncio.run(_again())
+    assert again[host]["SYSTIN"].startswith("falls as the CPU warms (r -1.00 over 100 h")

tests/plugins/host_agent/test_sensors.py

@@ -72,3 +72,13 @@ def test_a_judged_sensor_is_left_out_of_the_host_temperature_and_listed_with_why
     real, unwired = split_temps("util", {"temp3": 62.0, "Package id 0": 55.0})
     assert list(real) == ["Package id 0"]
     assert unwired == {"temp3": (62.0, "falls as the CPU warms (r -0.81 over 168 h)")}
+
+
+def test_a_later_read_replaces_the_hour_that_was_filling_and_old_hours_drop():
+    hours = {"h:temp:a": {T0: 50.0, T0 + timedelta(hours=1): 51.0}}
+    s.merge_hours(hours, [("h:temp:a", T0 + timedelta(hours=1), 55.0),     # the hour, read whole
+                          ("h:temp:a", T0 + timedelta(hours=2), 56.0),
+                          ("h:temp:b", T0 + timedelta(hours=2), None)], since=T0 + timedelta(hours=1))
+    assert hours == {"h:temp:a": {T0 + timedelta(hours=1): 55.0, T0 + timedelta(hours=2): 56.0}}
+    s.merge_hours(hours, [], since=T0 + timedelta(hours=9))                # a sensor gone quiet
+    assert hours == {}