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