refactor(host_agent): extract metric-query helpers to a model-free module
The previous commit's integration test failed at collection-time import:
importing plugins.host_agent.routes (which imports the host_agent ORM models at
top level) double-registers host_agent_registrations against the app-loaded
plugin's metadata ("Table already defined").
Move the two pure read helpers (_latest_metrics_for_host, _history_for_host)
plus SOURCE_MODULE / HISTORY_METRICS into plugins/host_agent/metrics_query.py,
which imports only the core PluginMetric model — no plugin models. routes.py
imports them back. The integration test now imports from metrics_query and no
longer trips the loader's registration guard. No behavior change.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Jg27rgypiW2efULXJDtMC
This commit is contained in:
@@ -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
|
||||||
@@ -15,15 +15,20 @@ from sqlalchemy import select, func, or_, and_
|
|||||||
from datetime import timedelta
|
from datetime import timedelta
|
||||||
|
|
||||||
from steward.core.settings import public_base_url
|
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.hosts import Host
|
||||||
from steward.models.metrics import PluginMetric
|
from steward.models.metrics import PluginMetric
|
||||||
from .models import HostAgentRegistration
|
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")
|
host_agent_bp = Blueprint("host_agent", __name__, template_folder="templates")
|
||||||
|
|
||||||
SOURCE_MODULE = "host_agent"
|
|
||||||
|
|
||||||
|
|
||||||
def _hash_token(raw: str) -> str:
|
def _hash_token(raw: str) -> str:
|
||||||
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
|
||||||
@@ -540,77 +545,6 @@ def _downsample(series: list[list], target: int = 120) -> list[list]:
|
|||||||
return out
|
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("/<host_id>/")
|
@host_agent_bp.get("/<host_id>/")
|
||||||
@require_role(UserRole.viewer)
|
@require_role(UserRole.viewer)
|
||||||
async def host_detail(host_id: str):
|
async def host_detail(host_id: str):
|
||||||
|
|||||||
@@ -34,7 +34,7 @@ def app():
|
|||||||
def test_latest_distinct_on_and_sql_bucketed_history(app):
|
def test_latest_distinct_on_and_sql_bucketed_history(app):
|
||||||
from sqlalchemy import text
|
from sqlalchemy import text
|
||||||
from steward.models.metrics import PluginMetric
|
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,
|
SOURCE_MODULE, _history_for_host, _latest_metrics_for_host,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user