From 613ed4e78b4063ad6983772fb2b1202894b8790c Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 19 Oct 2026 14:35:24 -0400 Subject: [PATCH] feat(traffic): each container's network rate, and each NFS mount's bytes, so a host's traffic has a sender (#6115) Docker02's outbound traffic tops out the internet connection and Steward couldn't say who sends it: a container's counters were kept only as a running total, and NFS traffic belongs to no container. - Docker ingest turns each running container's cumulative network counters into net_rx_bps / net_tx_bps on "/" (net_rates), counting from the stored row so a restart of Steward loses nothing; a container new, recreated or restarted has no rate until its next snapshot. - Agent 1.25.0 reads the bytes sent and received per NFS mount from mountstats (already parsed for the delays) as tx_bps / rx_bps; Steward stores them as nfs_tx_bps / nfs_rx_bps on ":". Co-Authored-By: Claude Opus 5.5 --- docs/plugins/host-agent-design.md | 2 + plugins/docker/ingest.py | 53 ++++++++++++++++--- plugins/host_agent/__init__.py | 6 ++- plugins/host_agent/agent.py | 20 ++++--- plugins/host_agent/routes.py | 6 ++- steward/core/formatting.py | 2 + tests/plugins/docker/test_net_rates.py | 37 +++++++++++++ .../host_agent/test_agent_collectors.py | 29 +++++----- 8 files changed, 123 insertions(+), 32 deletions(-) create mode 100644 tests/plugins/docker/test_net_rates.py diff --git a/docs/plugins/host-agent-design.md b/docs/plugins/host-agent-design.md index 72e6ef4..9039657 100644 --- a/docs/plugins/host-agent-design.md +++ b/docs/plugins/host-agent-design.md @@ -128,6 +128,8 @@ Each mount is read in a daemon thread under one 5 s deadline. A hard-mounted NFS `mounts_stalled` says a mount is stuck; it can't say a share was slow before it stuck, or how slow it was when it came back. Each sample also reads `/proc/self/mountstats`, the kernel's own cumulative count of every NFS request each mount has finished. Reading it never touches the share, so it answers while one is stuck. The agent sends each NFS mount's requests since the last sample as `nfs: [{mount, ops_per_sec, retrans, timeouts, rtt_ms, exe_ms, read_ms, write_ms}]`: `rtt_ms` is the network and server's round trip, `exe_ms` that plus the time queued on the host, which is what a program waits. The delays are averages over the requests that finished, and are left out when none did. A mount new since the last sample, or whose counters went back (remounted), waits for the next one. +**Bytes per mount (1.25.0, #6115).** Each op's line also counts the bytes sent and received on the wire, so each mount's row carries `tx_bps` and `rx_bps` too: the share's part of the host's network traffic, which belongs to no container. A large copy to the NAS shows here rather than as traffic nobody sent. + Steward stores them per mount (`nfs_*` on `:`), and the slowest mount's `exe_ms` on the host as `nfs_exe_max_ms`, for a rule ("NFS slow": over 1000 ms for 3 checks) and for the host page's NFS chart beside `mounts_stalled`. Each mount's bar on the host page says what a request took and how many were sent again or timed out. A request is counted when it finishes, so a stuck share shows nothing until it comes back, then its long requests all at once. ### A container's memory is its processes' own (1.23.0, #6030) diff --git a/plugins/docker/ingest.py b/plugins/docker/ingest.py index 903606f..6b37d89 100644 --- a/plugins/docker/ingest.py +++ b/plugins/docker/ingest.py @@ -30,6 +30,32 @@ def _parse_started_at(value) -> datetime | None: return None +def net_rates(before: dict, snapshots: list) -> list: + """(recorded_at, name, rx_bps, tx_bps) for each running container in each + snapshot, from its cumulative network counters against the last ones seen + for it: `before` is {name: (at, container_id, rx, tx)} from the stored rows, + and each snapshot moves it on. A container new since then, recreated under + the same name, or whose counters went back (restarted) has no rate until + its next snapshot (#6115).""" + last = dict(before) + out = [] + for at, containers in snapshots: + for c in containers: + name, rx, tx = c.get("name"), c.get("net_rx_bytes"), c.get("net_tx_bytes") + if not name or c.get("status") != "running" or rx is None or tx is None: + continue + cid = c.get("container_id") or "" + prev = last.get(name) + last[name] = (at, cid, rx, tx) + if prev is None or prev[1] != cid or prev[2] is None or prev[3] is None: + continue + dt = (at - prev[0]).total_seconds() + if dt <= 0 or rx < prev[2] or tx < prev[3]: + continue + out.append((at, name, (rx - prev[2]) / dt, (tx - prev[3]) / dt)) + return out + + def _derive_events(old_state: dict, new_containers: list) -> list: """Diff a fresh snapshot against stored per-container state → lifecycle events. @@ -346,7 +372,8 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, one entry per ingested sample carrying docker data (usually one; more when the agent flushes a backlog). Every snapshot's running containers record cpu_pct, mem_used_pct, mem_used_bytes and mem_cache_bytes into the one - metric store, at the snapshot's time; the newest snapshot drives current container state and + metric store, at the snapshot's time, and net_rx_bps / net_tx_bps from + their counters (net_rates); the newest snapshot drives current container state and lifecycle-event derivation. `swarm` is the newest sample's swarm object (or None off managers) — persisted when present. `disk` is the newest sample's /system/df summary (or None on Docker-less @@ -394,6 +421,14 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, # the one metric store (#5752): resource "/", so # containers of the same name on different hosts don't collide. Recorded # with the alert batch below, at each snapshot's own time. + # Snapshot of stored per-container state BEFORE the upsert overwrites it — + # the network counters each rate counts from, and the lifecycle-event diff. + # Skip event derivation on the host's first snapshot (empty state) so we + # don't emit a start per pre-existing container. + old_rows = (await session.execute( + select(DockerContainer).where(DockerContainer.host_id == host.id) + )).scalars().all() + batch = MetricBatch() for recorded_at, containers in ordered: for c in containers: @@ -404,14 +439,16 @@ async def persist_host_docker(session, host, snapshots, swarm=None, disk=None, if c.get(metric) is not None: batch.add(source_module="docker", resource_name=resource, metric_name=metric, value=c[metric], recorded_at=recorded_at) + # Bytes a second each container receives and sends, so a host's traffic + # can be put down to the container behind it (#6115). + before = {r.name: (r.scraped_at, r.container_id, r.net_rx_bytes, r.net_tx_bytes) for r in old_rows} + for recorded_at, name, rx_bps, tx_bps in net_rates(before, ordered): + resource = f"{host.name}/{name}" + batch.add(source_module="docker", resource_name=resource, + metric_name="net_rx_bps", value=round(rx_bps, 1), recorded_at=recorded_at) + batch.add(source_module="docker", resource_name=resource, + metric_name="net_tx_bps", value=round(tx_bps, 1), recorded_at=recorded_at) await mark("metrics") - - # Snapshot of stored per-container state BEFORE the upsert overwrites it — - # used to diff lifecycle events. Skip event derivation on the host's first - # snapshot (empty state) so we don't emit a start per pre-existing container. - old_rows = (await session.execute( - select(DockerContainer).where(DockerContainer.host_id == host.id) - )).scalars().all() old_state = { r.name: {"status": r.status, "health": r.health, "oom_killed": r.oom_killed, "exit_code": r.exit_code, diff --git a/plugins/host_agent/__init__.py b/plugins/host_agent/__init__.py index 340127a..de4c478 100644 --- a/plugins/host_agent/__init__.py +++ b/plugins/host_agent/__init__.py @@ -96,7 +96,9 @@ def get_alert_metrics() -> dict[str, list[str]]: average time a program waited on one (queueing included), nfs_rtt_ms the server's round trip, nfs_read_ms / nfs_write_ms the same for reads and writes alone, nfs_ops_per_sec how many, and nfs_retrans / nfs_timeouts the - retransmissions and major timeouts among them. nfs_exe_max_ms is the + retransmissions and major timeouts among them; nfs_tx_bps / nfs_rx_bps + (agent ≥ 1.25.0, #6115) are the bytes a second it sent and received, its + share of the host's network traffic. nfs_exe_max_ms is the host's slowest mount, for a covering rule: "NFS slow" is > 1000 for 3 checks. A healthy share answers in a few ms. @@ -125,7 +127,7 @@ def get_alert_metrics() -> dict[str, list[str]]: "updates_pending", "updates_security", "reboot_required", "maintenance_failed", "temp_c_max", "temp_c", "net_rx_bps", "net_tx_bps", "disk_read_bps", "disk_write_bps", "nfs_exe_max_ms", "nfs_exe_ms", "nfs_rtt_ms", "nfs_read_ms", "nfs_write_ms", - "nfs_ops_per_sec", "nfs_retrans", "nfs_timeouts", + "nfs_ops_per_sec", "nfs_retrans", "nfs_timeouts", "nfs_tx_bps", "nfs_rx_bps", *DRIVE_METRICS, ]} diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index 755c003..71b19d8 100644 --- a/plugins/host_agent/agent.py +++ b/plugins/host_agent/agent.py @@ -22,7 +22,7 @@ from collections import deque from concurrent.futures import ThreadPoolExecutor, wait as wait_futures from datetime import datetime, timezone -AGENT_VERSION = "1.24.0" +AGENT_VERSION = "1.25.0" # Default path to the local Docker Engine socket. Overridable via the # `docker_socket` config key; collection is silently skipped if it's absent or @@ -408,7 +408,8 @@ def collect_diskio_raw() -> dict: def read_mountstats(text: str) -> dict: """{mount point: {op: (ops, transmissions, timeouts, queue_ms, rtt_ms, - execute_ms)}} for each NFS mount in a mountstats text.""" + execute_ms, bytes_sent, bytes_received)}} for each NFS mount in a + mountstats text.""" out: dict = {} mount = None in_ops = False @@ -432,8 +433,8 @@ def read_mountstats(text: str) -> dict: name, sep, rest = line.strip().partition(":") nums = rest.split() if sep and len(nums) >= 8 and all(n.isdigit() for n in nums[:8]): - ops, trans, timeouts, _sent, _recv, queue, rtt, exe = (int(n) for n in nums[:8]) - out[mount][name] = (ops, trans, timeouts, queue, rtt, exe) + ops, trans, timeouts, sent, recv, queue, rtt, exe = (int(n) for n in nums[:8]) + out[mount][name] = (ops, trans, timeouts, queue, rtt, exe, sent, recv) return out @@ -448,8 +449,9 @@ def nfs_delays(cur: dict, prev: dict, dt: float) -> list: """Each NFS mount's requests since the last sample: how many a second, their average round trip and execute time (queueing included, what a program waits), READ and WRITE on their own, and the retransmissions and - major timeouts among them. A mount new since the last sample, or whose - counters went back (remounted), is left for the next one.""" + major timeouts among them, and the bytes a second it sent and received + on the wire (#6115). A mount new since the last sample, or whose counters + went back (remounted), is left for the next one.""" out: list = [] if dt <= 0: return out @@ -457,13 +459,15 @@ def nfs_delays(cur: dict, prev: dict, dt: float) -> list: before = prev.get(mount) if before is None: continue - d = {op: tuple(a - b for a, b in zip(v, before.get(op, (0,) * 6))) for op, v in ops.items()} + d = {op: tuple(a - b for a, b in zip(v, before.get(op, (0,) * len(v)))) for op, v in ops.items()} if any(x < 0 for v in d.values() for x in v): continue n = sum(v[0] for v in d.values()) row = {"mount": mount, "ops_per_sec": round(n / dt, 2), "retrans": max(0, sum(v[1] - v[0] for v in d.values())), - "timeouts": sum(v[2] for v in d.values())} + "timeouts": sum(v[2] for v in d.values()), + "tx_bps": round(sum(v[6] for v in d.values()) / dt, 1), + "rx_bps": round(sum(v[7] for v in d.values()) / dt, 1)} if n: row["rtt_ms"] = round(sum(v[4] for v in d.values()) / n, 2) row["exe_ms"] = round(sum(v[5] for v in d.values()) / n, 2) diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index dfc59fa..1ab07aa 100644 --- a/plugins/host_agent/routes.py +++ b/plugins/host_agent/routes.py @@ -292,7 +292,8 @@ def _expand_sample_to_metrics( row("mounts_stalled", host_name, stalled) # NFS delay per mount (agent ≥ 1.22.0, #5996): what its requests took - # since the last sample. The host's slowest mount is its own metric, for + # since the last sample, and (agent ≥ 1.25.0, #6115) the bytes a second + # it sent and received. The host's slowest mount is its own metric, for # a rule; recorded only when some mount finished a request, since a mount # with none, idle or stuck, has no delay to give (mounts_stalled covers # stuck). @@ -302,7 +303,8 @@ def _expand_sample_to_metrics( for key, metric in (("ops_per_sec", "nfs_ops_per_sec"), ("rtt_ms", "nfs_rtt_ms"), ("exe_ms", "nfs_exe_ms"), ("read_ms", "nfs_read_ms"), ("write_ms", "nfs_write_ms"), ("retrans", "nfs_retrans"), - ("timeouts", "nfs_timeouts")): + ("timeouts", "nfs_timeouts"), ("tx_bps", "nfs_tx_bps"), + ("rx_bps", "nfs_rx_bps")): if isinstance(nfs.get(key), (int, float)): row(metric, mount_res, nfs[key]) if isinstance(nfs.get("exe_ms"), (int, float)): diff --git a/steward/core/formatting.py b/steward/core/formatting.py index a9a9023..8c0468a 100644 --- a/steward/core/formatting.py +++ b/steward/core/formatting.py @@ -349,6 +349,8 @@ METRICS: dict[str, tuple[str, str]] = { "nfs_read_ms": ("NFS read delay", "one NFS mount: how long a read took on average"), "nfs_write_ms": ("NFS write delay", "one NFS mount: how long a write took on average"), "nfs_ops_per_sec": ("NFS requests", "one NFS mount: requests finished a second"), + "nfs_tx_bps": ("NFS sent", "one NFS mount: bytes a second sent to its server, writes included"), + "nfs_rx_bps": ("NFS received", "one NFS mount: bytes a second received from its server, reads included"), "nfs_retrans": ("NFS retransmissions", "one NFS mount: requests sent again since the last sample"), "nfs_timeouts": ("NFS timeouts", "one NFS mount: requests that timed out since the last sample"), "agent_rss_bytes": ("Agent memory", "the host agent's own resident memory; tens of MB is normal"), diff --git a/tests/plugins/docker/test_net_rates.py b/tests/plugins/docker/test_net_rates.py new file mode 100644 index 0000000..3029152 --- /dev/null +++ b/tests/plugins/docker/test_net_rates.py @@ -0,0 +1,37 @@ +"""net_rates: a container's bytes a second, from its cumulative network +counters against the last ones seen for it (#6115). Pure; no DB.""" +from datetime import datetime, timedelta, timezone + +from plugins.docker.ingest import net_rates + +T0 = datetime(2026, 10, 19, 18, 0, tzinfo=timezone.utc) + + +def _c(name, rx, tx, cid="abc", status="running"): + return {"name": name, "container_id": cid, "status": status, + "net_rx_bytes": rx, "net_tx_bytes": tx} + + +def test_rates_count_from_the_stored_row_and_then_from_each_snapshot(): + before = {"web": (T0, "abc", 1_000, 2_000)} + snaps = [(T0 + timedelta(seconds=10), [_c("web", 11_000, 52_000)]), + (T0 + timedelta(seconds=20), [_c("web", 11_000, 62_000)])] + assert net_rates(before, snaps) == [ + (T0 + timedelta(seconds=10), "web", 1_000.0, 5_000.0), + (T0 + timedelta(seconds=20), "web", 0.0, 1_000.0), + ] + + +def test_no_rate_for_a_new_recreated_or_restarted_container_until_its_next_snapshot(): + before = {"db": (T0, "old", 100, 100), "app": (T0, "abc", 5_000, 5_000)} + t1, t2 = T0 + timedelta(seconds=10), T0 + timedelta(seconds=20) + snaps = [(t1, [_c("new", 10, 10), _c("db", 200, 200, cid="fresh"), _c("app", 10, 10)]), + (t2, [_c("new", 110, 210), _c("db", 300, 300, cid="fresh"), _c("app", 20, 30)])] + assert net_rates(before, snaps) == [ + (t2, "new", 10.0, 20.0), (t2, "db", 10.0, 10.0), (t2, "app", 1.0, 2.0)] + + +def test_stopped_containers_and_missing_counters_are_skipped(): + before = {"web": (T0, "abc", 0, 0), "old": (T0, "abc", 0, 0)} + snaps = [(T0 + timedelta(seconds=10), [_c("web", None, None), _c("old", 9, 9, status="exited")])] + assert net_rates(before, snaps) == [] diff --git a/tests/plugins/host_agent/test_agent_collectors.py b/tests/plugins/host_agent/test_agent_collectors.py index ac31ea6..39b789a 100644 --- a/tests/plugins/host_agent/test_agent_collectors.py +++ b/tests/plugins/host_agent/test_agent_collectors.py @@ -174,9 +174,9 @@ MOUNTSTATS = ( def test_mountstats_reads_each_nfs_mount_s_ops_and_skips_the_rest(): got = a.read_mountstats(MOUNTSTATS) assert set(got) == {"/nfs/data", "/nfs/my media"} - assert got["/nfs/data"]["READ"] == (10, 10, 0, 5, 30, 40) - assert got["/nfs/data"]["NULL"] == (0, 0, 0, 0, 0, 0) - assert got["/nfs/my media"] == {"GETATTR": (3, 3, 0, 0, 3, 3)} + assert got["/nfs/data"]["READ"] == (10, 10, 0, 5, 30, 40, 1200, 40960) + assert got["/nfs/data"]["NULL"] == (0, 0, 0, 0, 0, 0, 0, 0) + assert got["/nfs/my media"] == {"GETATTR": (3, 3, 0, 0, 3, 3, 300, 300)} def test_mountstats_that_cannot_be_read_is_no_mounts(tmp_path): @@ -185,18 +185,23 @@ def test_mountstats_that_cannot_be_read_is_no_mounts(tmp_path): def test_nfs_delay_is_what_the_requests_since_the_last_sample_took(): - prev = {"/nfs/data": {"READ": (10, 10, 0, 5, 30, 40), "WRITE": (4, 4, 0, 2, 20, 24)}, - "/nfs/idle": {"GETATTR": (3, 3, 0, 0, 3, 3)}, - "/nfs/remounted": {"READ": (50, 50, 0, 0, 50, 50)}} - cur = {"/nfs/data": {"READ": (20, 21, 1, 7, 130, 240), "WRITE": (4, 4, 0, 2, 20, 24)}, - "/nfs/idle": {"GETATTR": (3, 3, 0, 0, 3, 3)}, - "/nfs/remounted": {"READ": (2, 2, 0, 0, 2, 2)}, - "/nfs/new": {"READ": (5, 5, 0, 0, 5, 5)}} + prev = {"/nfs/data": {"READ": (10, 10, 0, 5, 30, 40, 1200, 40960), + "WRITE": (4, 4, 0, 2, 20, 24, 8192, 512)}, + "/nfs/idle": {"GETATTR": (3, 3, 0, 0, 3, 3, 300, 300)}, + "/nfs/remounted": {"READ": (50, 50, 0, 0, 50, 50, 5000, 5000)}} + cur = {"/nfs/data": {"READ": (20, 21, 1, 7, 130, 240, 21680, 450560), + "WRITE": (4, 4, 0, 2, 20, 24, 8192, 512)}, + "/nfs/idle": {"GETATTR": (3, 3, 0, 0, 3, 3, 300, 300)}, + "/nfs/remounted": {"READ": (2, 2, 0, 0, 2, 2, 200, 200)}, + "/nfs/new": {"READ": (5, 5, 0, 0, 5, 5, 500, 500)}} got = {r["mount"]: r for r in a.nfs_delays(cur, prev, 10.0)} # ten READs in 10 s, 100 ms of round trip and 200 ms of waiting between - # them, one sent twice and one that hit a major timeout; no WRITE finished + # them, one sent twice and one that hit a major timeout; no WRITE finished. + # 20,480 bytes sent and 409,600 received on the wire. assert got["/nfs/data"] == {"mount": "/nfs/data", "ops_per_sec": 1.0, "retrans": 1, "timeouts": 1, + "tx_bps": 2048.0, "rx_bps": 40960.0, "rtt_ms": 10.0, "exe_ms": 20.0, "read_ms": 20.0} - assert got["/nfs/idle"] == {"mount": "/nfs/idle", "ops_per_sec": 0.0, "retrans": 0, "timeouts": 0} + assert got["/nfs/idle"] == {"mount": "/nfs/idle", "ops_per_sec": 0.0, "retrans": 0, "timeouts": 0, + "tx_bps": 0.0, "rx_bps": 0.0} assert set(got) == {"/nfs/data", "/nfs/idle"} # counters gone back, or no baseline yet assert a.nfs_delays(cur, prev, 0) == []