diff --git a/plugins/host_agent/metrics_query.py b/plugins/host_agent/metrics_query.py new file mode 100644 index 0000000..9b18843 --- /dev/null +++ b/plugins/host_agent/metrics_query.py @@ -0,0 +1,85 @@ +"""Host-metric read helpers (latest snapshot + bucketed history). + +Kept separate from routes.py so it imports only the core PluginMetric model — not +the host_agent ORM models — which lets integration tests import these helpers +without tripping the plugin-loader's double-registration ("Table already +defined") guard. +""" +from __future__ import annotations + +from sqlalchemy import func, or_, select + +from steward.core.time_range import bucket_seconds +from steward.models.metrics import PluginMetric + +SOURCE_MODULE = "host_agent" + +# Host-level metrics charted on the detail page (sub-resources are shown as +# current-value lists, not time series, to keep the page readable). +HISTORY_METRICS = ( + "cpu_pct", "mem_used_pct", "disk_used_pct_worst", "load_1m", + "net_rx_bps", "net_tx_bps", "disk_read_bps", "disk_write_bps", + "temp_c_max", "psi_mem_some_avg10", +) + + +async def _latest_metrics_for_host(session, host_name: str) -> dict[str, dict[str, float]]: + """{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, + PluginMetric.recorded_at.desc(), + ) + )).scalars().all() + out: dict[str, dict[str, float]] = {} + for r in rows: + out.setdefault(r.resource_name, {})[r.metric_name] = r.value + return out + + +async def _history_for_host(session, host_name: str, since) -> dict[str, list]: + """{metric: [[epoch_ms, avg_value], …]} host-level series since `since`. + + 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.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, + ) + .group_by(PluginMetric.metric_name, bucket) + .order_by(bucket) + )).all() + series: dict[str, list] = {m: [] for m in HISTORY_METRICS} + for metric_name, b, avg in rows: + series[metric_name].append([int(b.timestamp() * 1000), round(float(avg), 2)]) + return series diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 7adef6a..effb412 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -15,15 +15,20 @@ 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, bucket_seconds +from steward.core.time_range import parse_range, RANGE_OPTIONS from steward.models.hosts import Host from steward.models.metrics import PluginMetric from .models import HostAgentRegistration +# Query helpers live in a model-free module so integration tests can import them +# without the plugin-loader double-registration guard tripping (see metrics_query). +from .metrics_query import ( + SOURCE_MODULE, + _history_for_host, + _latest_metrics_for_host, +) host_agent_bp = Blueprint("host_agent", __name__, template_folder="templates") -SOURCE_MODULE = "host_agent" - def _hash_token(raw: str) -> str: return hashlib.sha256(raw.encode("utf-8")).hexdigest() @@ -540,77 +545,6 @@ def _downsample(series: list[list], target: int = 120) -> list[list]: return out -# Host-level metrics charted on the detail page (sub-resources are shown as -# current-value lists, not time series, to keep the page readable). -HISTORY_METRICS = ( - "cpu_pct", "mem_used_pct", "disk_used_pct_worst", "load_1m", - "net_rx_bps", "net_tx_bps", "disk_read_bps", "disk_write_bps", - "temp_c_max", "psi_mem_some_avg10", -) - - -async def _latest_metrics_for_host(session, host_name: str) -> dict[str, dict[str, float]]: - """{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, - PluginMetric.recorded_at.desc(), - ) - )).scalars().all() - out: dict[str, dict[str, float]] = {} - for r in rows: - out.setdefault(r.resource_name, {})[r.metric_name] = r.value - return out - - -async def _history_for_host(session, host_name: str, since) -> dict[str, list]: - """{metric: [[epoch_ms, avg_value], …]} host-level series since `since`. - - 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.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, - ) - .group_by(PluginMetric.metric_name, bucket) - .order_by(bucket) - )).all() - series: dict[str, list] = {m: [] for m in HISTORY_METRICS} - 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("//") @require_role(UserRole.viewer) async def host_detail(host_id: str): diff --git a/tests/integration/test_host_metrics.py b/tests/integration/test_host_metrics.py index 626a493..2c59e46 100644 --- a/tests/integration/test_host_metrics.py +++ b/tests/integration/test_host_metrics.py @@ -34,7 +34,7 @@ def app(): 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 ( + from plugins.host_agent.metrics_query import ( SOURCE_MODULE, _history_for_host, _latest_metrics_for_host, )