diff --git a/plugins/host_agent/sensors.py b/plugins/host_agent/sensors.py index 8579d4c..4b3297b 100644 --- a/plugins/host_agent/sensors.py +++ b/plugins/host_agent/sensors.py @@ -111,13 +111,34 @@ def judge_hosts(rows) -> dict[str, dict[str, str]]: 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.""" + 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.""" global _unwired - from sqlalchemy import select + from sqlalchemy import func, select + from steward.models.metrics import PluginMetric as M from steward.models.metrics import PluginMetricHourly as H - rows = (await session.execute( + 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 + hour = func.date_trunc("hour", M.recorded_at) + raw = (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.like("%:temp:%"), H.bucket >= now - WINDOW))).all() - _unwired = judge_hosts(rows) + H.resource_name.in_(names), H.bucket >= since))).all() + _unwired = judge_hosts([*rolled, *raw]) return _unwired diff --git a/tests/integration/test_host_metrics.py b/tests/integration/test_host_metrics.py index ee09572..fa29f1f 100644 --- a/tests/integration/test_host_metrics.py +++ b/tests/integration/test_host_metrics.py @@ -164,3 +164,50 @@ def test_history_merges_hourly_and_raw(app): assert 42.0 in vals, "expected the rolled-up hourly point" assert 77.0 in vals, "expected the recent raw point" assert cpu == sorted(cpu, key=lambda p: p[0]), "series must be time-ordered" + + +@_NEEDS_DB +def test_sensors_are_judged_from_raw_hours_and_rollups_together(app): + """The live instance keeps raw samples for 7 days and rolls up only what + is older, so a week of hourly rollups alone is empty: the first deploy of + #5783 judged nothing. refresh reads both, by sensor name.""" + import math + + from sqlalchemy import text + from steward.models.metrics import PluginMetric, PluginMetricHourly, PluginMetricLatest + from plugins.host_agent import sensors + + now = datetime.now(timezone.utc).replace(minute=30, second=0, microsecond=0) + host = "sensor-host-" + uuid.uuid4().hex[:8] + + def cpu(i): + return 65 + 15 * math.sin(i / 24 * 2 * math.pi) + + series = {"Package id 0": cpu, "SYSTIN": lambda i: 114 - 0.1 * (cpu(i) - 65), + "CPUTIN": lambda i: 40 + 0.3 * (cpu(i) - 65)} + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + for tbl in ("plugin_metrics", "plugin_metrics_hourly", "plugin_metric_latest"): + await s.execute(text(f"DELETE FROM {tbl} WHERE resource_name LIKE :p"), {"p": host + "%"}) + for label, f in series.items(): + res = f"{host}:temp:{label}" + for i in range(100): # 100 h back; the oldest 30 rolled up + at = now - timedelta(hours=100 - i) + if i < 30: + s.add(PluginMetricHourly(source_module="host_agent", resource_name=res, + metric_name="temp_c", bucket=at.replace(minute=0), + value_avg=f(i), value_max=f(i), sample_count=1)) + else: # two samples an hour, averaged by refresh + for m, d in ((5, -0.2), (35, 0.2)): + s.add(PluginMetric(source_module="host_agent", resource_name=res, + metric_name="temp_c", value=f(i) + d, + recorded_at=at.replace(minute=m))) + s.add(PluginMetricLatest(source_module="host_agent", resource_name=res, + metric_name="temp_c", value=f(99), recorded_at=now)) + return await sensors.refresh(s, now) + + 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")