diff --git a/steward/core/reports.py b/steward/core/reports.py index fddd63b..a066ca6 100644 --- a/steward/core/reports.py +++ b/steward/core/reports.py @@ -102,28 +102,13 @@ async def build_report_data(app) -> dict: ] # ── Monitors currently down (latest result is_up = false) ───────────── - latest_subq = ( - select(MonitorResult.monitor_id, func.max(MonitorResult.checked_at).label("max_at")) - .group_by(MonitorResult.monitor_id) - .subquery() - ) - down_result = await db.execute( - select(Monitor.name) - .join(MonitorResult, MonitorResult.monitor_id == Monitor.id) - .join( - latest_subq, - and_( - MonitorResult.monitor_id == latest_subq.c.monitor_id, - MonitorResult.checked_at == latest_subq.c.max_at, - ), - ) - .where( - Monitor.enabled.is_(True), - MonitorResult.is_up.is_(False), - ) - .order_by(Monitor.name) - ) - hosts_down = [row.name for row in down_result] + from steward.core.status import _last_n_results + enabled = (await db.execute( + select(Monitor.id, Monitor.name).where(Monitor.enabled.is_(True)) + )).all() + latest = await _last_n_results(db, [m.id for m in enabled], n=1) + hosts_down = sorted(name for mid, name in enabled + if latest.get(mid) and not latest[mid][-1].is_up) return { "generated_at": now, diff --git a/steward/core/status.py b/steward/core/status.py index 5f22c59..6bc88e2 100644 --- a/steward/core/status.py +++ b/steward/core/status.py @@ -105,25 +105,33 @@ def sparkline_svg(values: list[float], width: int = 80, height: int = 20, async def _last_n_results(db, monitor_ids: list[str], n: int = HEARTBEAT_COUNT) -> dict[str, list]: - """Return {monitor_id: [MonitorResult]} of the last n results, oldest-first.""" - from steward.models.monitors import MonitorResult + """Return {monitor_id: [MonitorResult]} of the last n results, oldest-first. + + One LATERAL probe per monitor walks ix_monitor_results_monitor_checked + backwards and stops after n rows. A row_number() over each monitor's + results read its whole history first, which held Status and the monitor + widgets at ~16 s (#5208). + """ + from sqlalchemy import true + from sqlalchemy.orm import aliased + from steward.models.monitors import Monitor, MonitorResult if not monitor_ids: return {} - rn = func.row_number().over( - partition_by=MonitorResult.monitor_id, order_by=MonitorResult.checked_at.desc() - ).label("rn") - subq = select(MonitorResult.id, rn).where( - MonitorResult.monitor_id.in_(monitor_ids) - ).subquery() - res = await db.execute( + ids = select(Monitor.id.label("mid")).where(Monitor.id.in_(monitor_ids)).subquery("wanted") + recent = ( select(MonitorResult) - .join(subq, MonitorResult.id == subq.c.id) - .where(subq.c.rn <= n) - .order_by(MonitorResult.monitor_id, MonitorResult.checked_at.asc()) + .where(MonitorResult.monitor_id == ids.c.mid) + .order_by(MonitorResult.checked_at.desc()) + .limit(n) + .lateral("recent") ) + row = aliased(MonitorResult, recent) + res = await db.execute(select(row).select_from(ids).join(recent, true())) out: dict[str, list] = {mid: [] for mid in monitor_ids} - for row in res.scalars(): - out[row.monitor_id].append(row) + for r in res.scalars(): + out[r.monitor_id].append(r) + for rows in out.values(): + rows.sort(key=lambda r: r.checked_at) return out diff --git a/steward/hosts/service.py b/steward/hosts/service.py index f8a0344..f625864 100644 --- a/steward/hosts/service.py +++ b/steward/hosts/service.py @@ -93,21 +93,12 @@ async def host_status(db, host_ids: list[str]) -> dict[str, dict]: """ if not host_ids: return {} - rn = func.row_number().over( - partition_by=MonitorResult.monitor_id, order_by=MonitorResult.checked_at.desc() - ).label("rn") - subq = ( - select(MonitorResult.id, rn) - .join(Monitor, Monitor.id == MonitorResult.monitor_id) + monitors = (await db.execute( + select(Monitor.id, Monitor.host_id) .where(Monitor.host_id.in_(host_ids), Monitor.enabled.is_(True)) - .subquery() - ) - rows = (await db.execute( - select(MonitorResult, Monitor.host_id) - .join(subq, MonitorResult.id == subq.c.id) - .join(Monitor, Monitor.id == MonitorResult.monitor_id) - .where(subq.c.rn == 1) )).all() + latest = await _last_n_results(db, [m.id for m in monitors], n=1) + rows = [(latest[mid][-1], host_id) for mid, host_id in monitors if latest.get(mid)] out: dict[str, dict] = {} for res, host_id in rows: d = out.setdefault(host_id, {"state": "pending", "latency_ms": None, "count": 0}) diff --git a/tests/integration/test_metric_latest.py b/tests/integration/test_metric_latest.py index 5b3a2c5..add7a59 100644 --- a/tests/integration/test_metric_latest.py +++ b/tests/integration/test_metric_latest.py @@ -54,7 +54,8 @@ def test_the_newest_sample_wins_and_a_late_backlog_does_not_move_it_back(app): _sample(host, "cpu_pct", 20.0, now - timedelta(minutes=1)), _sample(host, "mem_used_pct", 40.0, now - timedelta(minutes=1)), ]) - after_first = await _latest(s, host) + async with s.begin(): + after_first = await _latest(s, host) async with s.begin(): # An agent catching up posts older samples after newer ones. await record_metrics(s, [ diff --git a/tests/integration/test_monitors.py b/tests/integration/test_monitors.py index 0e0ebd7..d55afa3 100644 --- a/tests/integration/test_monitors.py +++ b/tests/integration/test_monitors.py @@ -83,3 +83,45 @@ def test_status_source_rolls_up_results(app): assert e.status == "up" # latest result is up assert len(e.heartbeat) == 4 assert e.uptime["24h"] == 75.0 # 3 of 4 up + + +@_NEEDS_DB +def test_last_results_are_the_newest_n_oldest_first_and_host_status_reads_the_last(app): + """_last_n_results takes each monitor's newest n by a LATERAL probe (#5208); + host_status reads the newest one per enabled monitor.""" + from steward.core.status import _last_n_results + from steward.hosts.service import host_status + from steward.models.hosts import Host + from steward.models.monitors import Monitor, MonitorResult + + now = datetime.now(timezone.utc) + hid, a, b, quiet = (str(uuid.uuid4()) for _ in range(4)) + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(Host(id=hid, name="lateral-" + hid[:8], address="10.8.8.8")) + async with s.begin(): + s.add(Monitor(id=a, name="a-" + a[:8], type="icmp", target="10.8.8.8", + host_id=hid, config_json="{}", enabled=True)) + s.add(Monitor(id=b, name="b-" + b[:8], type="icmp", target="10.8.8.9", + host_id=hid, config_json="{}", enabled=True)) + s.add(Monitor(id=quiet, name="q-" + quiet[:8], type="icmp", target="10.8.8.7", + config_json="{}", enabled=True)) + async with s.begin(): + for i in range(5): # a: up, up, up, up, down (newest) + s.add(MonitorResult(monitor_id=a, checked_at=now - timedelta(minutes=5 - i), + is_up=i < 4)) + s.add(MonitorResult(monitor_id=b, checked_at=now - timedelta(minutes=1), + is_up=True, response_ms=12.0)) + async with s.begin(): + last = await _last_n_results(s, [a, b, quiet], n=3) + status = await host_status(s, [hid]) + return last, status + + last, status = asyncio.run(_go()) + assert [r.is_up for r in last[a]] == [True, True, False] + assert last[a][0].checked_at < last[a][-1].checked_at # oldest first + assert len(last[b]) == 1 and last[quiet] == [] + assert status[hid]["state"] == "down" and status[hid]["count"] == 2 + assert status[hid]["latency_ms"] == 12.0