diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 1b20286..7adef6a 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -15,7 +15,7 @@ from sqlalchemy import select, func, or_, and_ from datetime import timedelta from steward.core.settings import public_base_url -from steward.core.time_range import parse_range, RANGE_OPTIONS +from steward.core.time_range import parse_range, RANGE_OPTIONS, bucket_seconds from steward.models.hosts import Host from steward.models.metrics import PluginMetric from .models import HostAgentRegistration @@ -550,27 +550,28 @@ HISTORY_METRICS = ( async def _latest_metrics_for_host(session, host_name: str) -> dict[str, dict[str, float]]: - """{resource_name: {metric: value}} of the latest sample for a host + sub-resources.""" - subq = ( - select( + """{resource_name: {metric: value}} — latest sample per (resource, metric) for + a host + its sub-resources. + + DISTINCT ON picks the newest row per group in one index-ordered pass over + ix_plugin_metrics_module_resource_metric_recorded, instead of a GROUP-BY-max + subquery self-joined back to the table (two passes over the whole history). + """ + rows = (await session.execute( + select(PluginMetric) + .where( + PluginMetric.source_module == SOURCE_MODULE, + or_( + PluginMetric.resource_name == host_name, + PluginMetric.resource_name.like(host_name + ":%"), + ), + ) + .distinct(PluginMetric.resource_name, PluginMetric.metric_name) + .order_by( PluginMetric.resource_name, PluginMetric.metric_name, - func.max(PluginMetric.recorded_at).label("max_ts"), + PluginMetric.recorded_at.desc(), ) - .where(PluginMetric.source_module == SOURCE_MODULE) - .where(or_( - PluginMetric.resource_name == host_name, - PluginMetric.resource_name.like(host_name + ":%"), - )) - .group_by(PluginMetric.resource_name, PluginMetric.metric_name) - ).subquery() - rows = (await session.execute( - select(PluginMetric).join( - subq, - (PluginMetric.resource_name == subq.c.resource_name) & - (PluginMetric.metric_name == subq.c.metric_name) & - (PluginMetric.recorded_at == subq.c.max_ts), - ).where(PluginMetric.source_module == SOURCE_MODULE) )).scalars().all() out: dict[str, dict[str, float]] = {} for r in rows: @@ -579,24 +580,35 @@ async def _latest_metrics_for_host(session, host_name: str) -> dict[str, dict[st async def _history_for_host(session, host_name: str, since) -> dict[str, list]: - """{metric: [[epoch_ms, value], …]} host-level series since `since`. + """{metric: [[epoch_ms, avg_value], …]} host-level series since `since`. - Epoch-ms x values let the charts use a linear axis (no Chart.js date adapter). + Buckets + averages in SQL (date_bin to ~120 buckets) so we return a readable + point count instead of shipping every raw sample to Python and downsampling + there — a 30d range was reading hundreds of thousands of rows per load. The + bucket width is epoch-aligned so the x axis is stable across refreshes. + Epoch-ms x values feed a linear chart axis (no Chart.js date adapter). """ + width_s = bucket_seconds(since, 120) + bucket = func.date_bin( + func.make_interval(0, 0, 0, 0, 0, 0, width_s), # width_s seconds + PluginMetric.recorded_at, + func.to_timestamp(0), # epoch origin + ).label("bucket") rows = (await session.execute( - select(PluginMetric).where( + select(PluginMetric.metric_name, bucket, func.avg(PluginMetric.value)) + .where( PluginMetric.source_module == SOURCE_MODULE, PluginMetric.resource_name == host_name, PluginMetric.metric_name.in_(HISTORY_METRICS), PluginMetric.recorded_at >= since, - ).order_by(PluginMetric.recorded_at) - )).scalars().all() + ) + .group_by(PluginMetric.metric_name, bucket) + .order_by(bucket) + )).all() series: dict[str, list] = {m: [] for m in HISTORY_METRICS} - for p in rows: - series[p.metric_name].append([int(p.recorded_at.timestamp() * 1000), round(p.value, 2)]) - # Downsample to a readable point count (see _downsample) — raw agent cadence - # is too dense to read over a multi-hour window. - return {m: _downsample(v) for m, v in series.items()} + for metric_name, b, avg in rows: + series[metric_name].append([int(b.timestamp() * 1000), round(float(avg), 2)]) + return series @host_agent_bp.get("//") diff --git a/tests/integration/test_host_metrics.py b/tests/integration/test_host_metrics.py new file mode 100644 index 0000000..626a493 --- /dev/null +++ b/tests/integration/test_host_metrics.py @@ -0,0 +1,80 @@ +"""Integration: host-metrics read paths (DISTINCT ON latest + SQL date_bin history). + +Validates the two host_agent query helpers that back the slow host views, against +a live Postgres: the latest-per-(resource,metric) lookup and the bucket-averaged +history (which now aggregates in SQL instead of shipping raw rows to Python). +Requires STEWARD_DATABASE_URL. +""" +from __future__ import annotations + +import asyncio +import os +import uuid +from datetime import datetime, timedelta, timezone + +import pytest + +pytestmark = pytest.mark.integration + +_NEEDS_DB = pytest.mark.skipif( + not os.environ.get("STEWARD_DATABASE_URL"), + reason="integration test needs a live Postgres (STEWARD_DATABASE_URL)", +) + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +@_NEEDS_DB +def test_latest_distinct_on_and_sql_bucketed_history(app): + from sqlalchemy import text + from steward.models.metrics import PluginMetric + from plugins.host_agent.routes import ( + SOURCE_MODULE, _history_for_host, _latest_metrics_for_host, + ) + + now = datetime.now(timezone.utc) + hostname = "metrics-host-" + uuid.uuid4().hex[:8] + + async def _go(): + async with app.db_sessionmaker() as s: + async with s.begin(): + await s.execute( + text("DELETE FROM plugin_metrics WHERE resource_name LIKE :p"), + {"p": hostname + "%"}, + ) + rows = [] + # Host-level CPU samples, oldest → newest (10, 20, 30). + for i, val in enumerate([10.0, 20.0, 30.0]): + rows.append(PluginMetric( + source_module=SOURCE_MODULE, resource_name=hostname, + metric_name="cpu_pct", value=val, + recorded_at=now - timedelta(minutes=30 - i * 10), + )) + # A sub-resource (root mount) to exercise the host:% match. + rows.append(PluginMetric( + source_module=SOURCE_MODULE, resource_name=hostname + ":/", + metric_name="disk_used_pct", value=80.0, + recorded_at=now - timedelta(minutes=5), + )) + s.add_all(rows) + + latest = await _latest_metrics_for_host(s, hostname) + hist = await _history_for_host(s, hostname, now - timedelta(hours=1)) + return latest, hist + + latest, hist = asyncio.run(_go()) + + # DISTINCT ON returns the newest sample per (resource, metric). + assert latest[hostname]["cpu_pct"] == 30.0 + assert latest[hostname + ":/"]["disk_used_pct"] == 80.0 + + # SQL date_bin aggregation returns cpu_pct buckets; averages stay in range. + cpu = hist["cpu_pct"] + assert cpu, "expected bucketed cpu_pct history" + assert all(10.0 <= v <= 30.0 for _, v in cpu)