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 <noreply@anthropic.com>

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)}

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))