diff --git a/plugins/host_agent/sensors.py b/plugins/host_agent/sensors.py index 4b3297b..031f176 100644 --- a/plugins/host_agent/sensors.py +++ b/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 diff --git a/tests/integration/test_host_metrics.py b/tests/integration/test_host_metrics.py index fa29f1f..bf1330e 100644 --- a/tests/integration/test_host_metrics.py +++ b/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") diff --git a/tests/plugins/host_agent/test_sensors.py b/tests/plugins/host_agent/test_sensors.py index bdb6d1c..b918383 100644 --- a/tests/plugins/host_agent/test_sensors.py +++ b/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 == {}