From 403025e20162e65c57a4f731219d575b43e2395e Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 19 Oct 2026 14:59:50 -0400 Subject: [PATCH] fix(traffic): find a host's senders in the latest-value table, not by a prefix scan of every metric (#6115) get_host_senders on VDNT-Docker02 ran past 120 s: a prefix LIKE on plugin_metrics.resource_name can't use its index (the column's collation isn't C), so each read went through the whole table, and the host page's Who's sending panel repeats it on every refresh. The containers and NFS mounts under a host are now named from plugin_metric_latest, and their series read by resource IN (...) on the series index. Co-Authored-By: Claude Opus 5.5 --- plugins/host_agent/senders.py | 32 +++++++++++++++++--------- tests/integration/test_host_senders.py | 9 +++++++- 2 files changed, 29 insertions(+), 12 deletions(-) diff --git a/plugins/host_agent/senders.py b/plugins/host_agent/senders.py index 9efd758..38b9dfc 100644 --- a/plugins/host_agent/senders.py +++ b/plugins/host_agent/senders.py @@ -19,7 +19,7 @@ from datetime import datetime, timedelta, timezone from sqlalchemy import func, select -from steward.models.metrics import PluginMetric +from steward.models.metrics import PluginMetric, PluginMetricLatest SOURCE_MODULE = "host_agent" DIRECTIONS = ("tx", "rx") @@ -80,24 +80,34 @@ async def senders(session, host_name: str, since: datetime, direction: str = "tx func.to_timestamp(0)).label("bucket") metric = f"net_{direction}_bps" - async def series(source, metric_name, resource): - rows = (await session.execute( + async def named(source, metric_name, prefix): + # The resources under the host, from the small latest-value table: a + # prefix LIKE on plugin_metrics itself can't use its index (the + # column's collation isn't C) and reads the whole table. + return list((await session.execute( + select(PluginMetricLatest.resource_name).where( + PluginMetricLatest.source_module == source, PluginMetricLatest.metric_name == metric_name, + PluginMetricLatest.resource_name.startswith(prefix, autoescape=True), + PluginMetricLatest.recorded_at >= since))).scalars()) + + async def series(source, metric_name, resources): + if not resources: + return [] + return (await session.execute( select(PluginMetric.resource_name, bucket, func.avg(PluginMetric.value)) - .where(PluginMetric.source_module == source, PluginMetric.metric_name == metric_name, - resource, PluginMetric.recorded_at >= since) + .where(PluginMetric.source_module == source, PluginMetric.resource_name.in_(resources), + PluginMetric.metric_name == metric_name, PluginMetric.recorded_at >= since) .group_by(PluginMetric.resource_name, bucket))).all() - return rows host: dict = {} - for _, b, v in await series(SOURCE_MODULE, metric, PluginMetric.resource_name == host_name): + for _, b, v in await series(SOURCE_MODULE, metric, [host_name]): host[b] = float(v) out: dict = {} - for res, b, v in await series("docker", metric, - PluginMetric.resource_name.startswith(host_name + "/", autoescape=True)): + for res, b, v in await series("docker", metric, await named("docker", metric, host_name + "/")): key = ("container", sender_name(res[len(host_name) + 1:])) out.setdefault(key, {}) out[key][b] = out[key].get(b, 0.0) + float(v) # a service's tasks add up - for res, b, v in await series(SOURCE_MODULE, f"nfs_{direction}_bps", - PluginMetric.resource_name.startswith(host_name + ":/", autoescape=True)): + nfs = f"nfs_{direction}_bps" + for res, b, v in await series(SOURCE_MODULE, nfs, await named(SOURCE_MODULE, nfs, host_name + ":/")): out.setdefault(("nfs", res[len(host_name) + 1:]), {})[b] = float(v) return {"since": since, "bucket_s": width, "direction": direction, **rank(host, out, limit)} diff --git a/tests/integration/test_host_senders.py b/tests/integration/test_host_senders.py index 35212e7..128b52b 100644 --- a/tests/integration/test_host_senders.py +++ b/tests/integration/test_host_senders.py @@ -21,7 +21,7 @@ def app(): def test_the_host_s_peak_is_put_down_to_its_senders(app): - from steward.models.metrics import PluginMetric + from steward.models.metrics import PluginMetric, PluginMetricLatest from plugins.host_agent.senders import senders now = datetime.now(timezone.utc) @@ -47,6 +47,13 @@ def test_the_host_s_peak_is_put_down_to_its_senders(app): m("host_agent", f"{host}:/nfs/data", "nfs_tx_bps", 100.0, peak), m("host_agent", f"{host}:net:eno1", "net_tx_bps", 1000.0, peak), ]) + # the latest-value table names the resources under the host + for src, res, metric in (("docker", f"{host}/web.1.{task}", "net_tx_bps"), + ("docker", f"{host}/web.2.{task}", "net_tx_bps"), + ("docker", f"{host}x/web.1.{task}", "net_tx_bps"), + ("host_agent", f"{host}:/nfs/data", "nfs_tx_bps")): + await s.merge(PluginMetricLatest(source_module=src, resource_name=res, + metric_name=metric, value=1.0, recorded_at=peak)) async with app.db_sessionmaker() as s: return await senders(s, host, now - timedelta(hours=1))