feat(host_agent): collect network, disk I/O, per-core CPU, temps, memory PSI

Extends the push agent (v1.2.0) with the Netdata-style signals the operator
wants: per-core CPU, per-interface network throughput and per-disk I/O
(rates derived in-agent from monotonic /proc counter deltas, with
counter-reset clamping), hardware temperatures (/sys/class/hwmon),
memory-pressure PSI (/proc/pressure), and cached/buffers memory breakdown.
All stdlib-only.

Server side needs no migration — these land as additional rows in the shared
PluginMetric table via _expand_sample_to_metrics. Per-resource series use a
':' in resource_name (host:net:eth0, host:core0, host:temp:Package) so the
existing fleet widget's ':' filter ignores them; host-level totals/max are
emitted at the bare host resource. New sample keys are optional, so older
agents keep ingesting unchanged.

Unit tests for the new parsers, rate/reset logic, and the expander contract.
Data foundation for the Netdata-style host view (task #867).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

plugins/host_agent/agent.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 
 import json
 import os
+import re
 import shutil
 import signal
 import socket
@@ -18,7 +19,7 @@ import urllib.request
 from collections import deque
 from datetime import datetime, timezone
 
-AGENT_VERSION = "1.1.0"
+AGENT_VERSION = "1.2.0"
 
 
 class ConfigError(Exception):
@@ -71,6 +72,15 @@ MEMINFO_PATH = "/proc/meminfo"
 LOADAVG_PATH = "/proc/loadavg"
 UPTIME_PATH = "/proc/uptime"
 OS_RELEASE_PATH = "/etc/os-release"
+NETDEV_PATH = "/proc/net/dev"
+DISKSTATS_PATH = "/proc/diskstats"
+HWMON_DIR = "/sys/class/hwmon"
+PSI_DIR = "/proc/pressure"
+
+# Partition names to drop from disk I/O (we report whole-disk throughput only).
+_PARTITION_RE = re.compile(
+    r"^(?:sd[a-z]+\d+|vd[a-z]+\d+|hd[a-z]+\d+|nvme\d+n\d+p\d+|mmcblk\d+p\d+)$"
+)
 
 
 def _read_file(path: str) -> str:
@@ -78,28 +88,49 @@ def _read_file(path: str) -> str:
         return f.read()
 
 
-def _parse_cpu_line(stat_text: str) -> tuple[int, int]:
-    """Return (total_jiffies, idle_jiffies) for the aggregate cpu line."""
+def _parse_cpu_lines(stat_text: str) -> dict:
+    """Return {name: (total_jiffies, idle_jiffies)} for every cpu line.
+
+    'cpu' is the aggregate; 'cpu0', 'cpu1', … are per-core.
+    """
+    out: dict = {}
     for line in stat_text.splitlines():
-        if line.startswith("cpu "):
-            parts = line.split()
+        if not line.startswith("cpu"):
+            continue
+        parts = line.split()
+        try:
             fields = [int(x) for x in parts[1:]]
-            idle = fields[3] + (fields[4] if len(fields) > 4 else 0)  # idle + iowait
-            total = sum(fields)
-            return total, idle
-    raise RuntimeError("no aggregate cpu line in /proc/stat")
+        except ValueError:
+            continue
+        if len(fields) < 4:
+            continue
+        idle = fields[3] + (fields[4] if len(fields) > 4 else 0)  # idle + iowait
+        out[parts[0]] = (sum(fields), idle)
+    return out
 
 
-def collect_cpu(sample_window: float = 0.2) -> float:
-    """Return CPU utilization % over sample_window seconds."""
-    t1, i1 = _parse_cpu_line(_read_file(STAT_PATH))
+def collect_cpu(sample_window: float = 0.2) -> tuple[float, list]:
+    """Return (aggregate_pct, [per_core_pct, …]) over sample_window seconds."""
+    a = _parse_cpu_lines(_read_file(STAT_PATH))
     time.sleep(sample_window)
-    t2, i2 = _parse_cpu_line(_read_file(STAT_PATH))
-    dt = t2 - t1
-    di = i2 - i1
-    if dt <= 0:
-        return 0.0
-    return round(100.0 * (dt - di) / dt, 2)
+    b = _parse_cpu_lines(_read_file(STAT_PATH))
+
+    def _pct(name: str):
+        if name not in a or name not in b:
+            return None
+        dt = b[name][0] - a[name][0]
+        di = b[name][1] - a[name][1]
+        if dt <= 0:
+            return 0.0
+        return round(100.0 * (dt - di) / dt, 2)
+
+    agg = _pct("cpu") or 0.0
+    cores: list = []
+    i = 0
+    while f"cpu{i}" in b:
+        cores.append(_pct(f"cpu{i}"))
+        i += 1
+    return agg, cores
 
 
 def collect_memory() -> dict:
@@ -125,6 +156,8 @@ def collect_memory() -> dict:
         "used_bytes": max(total - available, 0),
         "available_bytes": available,
         "swap_used_bytes": max(swap_total - swap_free, 0),
+        "cached_bytes": info.get("Cached", 0),
+        "buffers_bytes": info.get("Buffers", 0),
     }
 
 
@@ -210,6 +243,149 @@ def default_mounts() -> list[str]:
     return mounts or ["/"]
 
 
+def collect_net_raw() -> dict:
+    """Return {iface: (rx_bytes, tx_bytes)} cumulative counters from /proc/net/dev.
+
+    Skips loopback and veth* (container) interfaces to cut noise. Counters are
+    monotonic; per-second rates are derived from deltas in build_sample().
+    """
+    out: dict = {}
+    try:
+        text = _read_file(NETDEV_PATH)
+    except OSError:
+        return out
+    for line in text.splitlines():
+        if ":" not in line:
+            continue
+        name, _, rest = line.partition(":")
+        name = name.strip()
+        if name == "lo" or name.startswith("veth"):
+            continue
+        f = rest.split()
+        if len(f) < 16:
+            continue
+        try:
+            out[name] = (int(f[0]), int(f[8]))  # rx_bytes, tx_bytes
+        except ValueError:
+            continue
+    return out
+
+
+def collect_diskio_raw() -> dict:
+    """Return {device: (read_bytes, write_bytes)} cumulative from /proc/diskstats.
+
+    Whole disks only — loop/ram/sr/fd and partitions are skipped. Sectors are
+    512 bytes. Counters are monotonic; rates derived from deltas.
+    """
+    out: dict = {}
+    try:
+        text = _read_file(DISKSTATS_PATH)
+    except OSError:
+        return out
+    for line in text.splitlines():
+        f = line.split()
+        if len(f) < 10:
+            continue
+        name = f[2]
+        if name.startswith(("loop", "ram", "sr", "fd")) or _PARTITION_RE.match(name):
+            continue
+        try:
+            out[name] = (int(f[5]) * 512, int(f[9]) * 512)  # sectors_read, sectors_written
+        except ValueError:
+            continue
+    return out
+
+
+def collect_temps() -> list:
+    """Return [{label, celsius}] from /sys/class/hwmon (best-effort, may be empty)."""
+    out: list = []
+    try:
+        chips = os.listdir(HWMON_DIR)
+    except OSError:
+        return out
+    for hw in chips:
+        d = os.path.join(HWMON_DIR, hw)
+        chip = ""
+        try:
+            chip = _read_file(os.path.join(d, "name")).strip()
+        except OSError:
+            pass
+        try:
+            files = os.listdir(d)
+        except OSError:
+            continue
+        for fn in files:
+            if not (fn.startswith("temp") and fn.endswith("_input")):
+                continue
+            idx = fn[: -len("_input")]  # e.g. "temp1"
+            try:
+                milli = int(_read_file(os.path.join(d, fn)).strip())
+            except (OSError, ValueError):
+                continue
+            label = ""
+            try:
+                label = _read_file(os.path.join(d, idx + "_label")).strip()
+            except OSError:
+                pass
+            name = label or (f"{chip}_{idx}" if chip else idx)
+            out.append({"label": name, "celsius": round(milli / 1000.0, 1)})
+    return out
+
+
+def _parse_psi(text: str) -> dict:
+    """Parse a /proc/pressure/* file into {some_avg10, some_avg60, full_avg10, …}."""
+    res: dict = {}
+    for line in text.splitlines():
+        parts = line.split()
+        if not parts:
+            continue
+        kind = parts[0]  # "some" | "full"
+        for p in parts[1:]:
+            for prefix, suffix in (("avg10=", "avg10"), ("avg60=", "avg60")):
+                if p.startswith(prefix):
+                    try:
+                        res[f"{kind}_{suffix}"] = float(p[len(prefix):])
+                    except ValueError:
+                        pass
+    return res
+
+
+def collect_psi() -> dict:
+    """Return pressure-stall (PSI) gauges, prefixed by resource (mem/cpu/io).
+
+    Absent on kernels without CONFIG_PSI — returns only what exists.
+    """
+    out: dict = {}
+    for res_name, fname in (("mem", "memory"), ("cpu", "cpu"), ("io", "io")):
+        try:
+            text = _read_file(os.path.join(PSI_DIR, fname))
+        except OSError:
+            continue
+        for k, v in _parse_psi(text).items():
+            out[f"{res_name}_{k}"] = v
+    return out
+
+
+def _rates(cur: dict, prev: dict, dt: float) -> dict:
+    """Per-second rates for counters present in both snapshots over dt seconds.
+
+    Skips names missing from prev and negative deltas (counter reset / iface or
+    disk hot-plug), so a reboot never emits a bogus spike.
+    """
+    out: dict = {}
+    if dt <= 0:
+        return out
+    for name, cur_vals in cur.items():
+        if name not in prev:
+            continue
+        prev_vals = prev[name]
+        deltas = [c - p for c, p in zip(cur_vals, prev_vals)]
+        if any(d < 0 for d in deltas):
+            continue
+        out[name] = tuple(d / dt for d in deltas)
+    return out
+
+
 # ─── ring buffer ─────────────────────────────────────────────────────────────
 
 
@@ -232,11 +408,17 @@ class RingBuffer:
 # ─── payload ─────────────────────────────────────────────────────────────────
 
 
-def build_sample(mounts: list[str]) -> dict:
-    """Collect one full sample. Partial samples allowed if a collector fails."""
+def build_sample(mounts: list[str], state: dict) -> dict:
+    """Collect one full sample. Partial samples allowed if a collector fails.
+
+    `state` carries the previous network/disk counters + monotonic timestamp so
+    throughput rates can be derived from deltas; it is mutated in place.
+    """
     sample: dict = {"ts": datetime.now(timezone.utc).isoformat()}
     try:
-        sample["cpu_pct"] = collect_cpu()
+        agg, cores = collect_cpu()
+        sample["cpu_pct"] = agg
+        sample["cpu_cores"] = cores
     except Exception:
         sample["cpu_pct"] = None
     try:
@@ -255,6 +437,35 @@ def build_sample(mounts: list[str]) -> dict:
         sample["storage"] = collect_storage(mounts)
     except Exception:
         sample["storage"] = []
+    try:
+        sample["temps"] = collect_temps()
+    except Exception:
+        sample["temps"] = []
+    try:
+        sample["psi"] = collect_psi()
+    except Exception:
+        sample["psi"] = {}
+
+    # Throughput rates: diff cumulative counters against the previous sample.
+    now_mono = time.monotonic()
+    try:
+        net_raw = collect_net_raw()
+    except Exception:
+        net_raw = {}
+    try:
+        disk_raw = collect_diskio_raw()
+    except Exception:
+        disk_raw = {}
+    dt = now_mono - state["mono"] if state.get("mono") else 0.0
+    sample["net"] = [
+        {"iface": n, "rx_bps": round(rx, 1), "tx_bps": round(tx, 1)}
+        for n, (rx, tx) in _rates(net_raw, state.get("net", {}), dt).items()
+    ]
+    sample["diskio"] = [
+        {"device": n, "read_bps": round(rd, 1), "write_bps": round(wr, 1)}
+        for n, (rd, wr) in _rates(disk_raw, state.get("disk", {}), dt).items()
+    ]
+    state["net"], state["disk"], state["mono"] = net_raw, disk_raw, now_mono
     return sample
 
 
@@ -338,6 +549,8 @@ def main_loop(conf_path: str) -> int:
     mounts = cfg.get("mounts") or default_mounts()
     buffer = RingBuffer(maxlen=20)
     backoff = 0
+    # Carries previous net/disk counters + monotonic ts for rate computation.
+    rate_state: dict = {}
 
     _log("INFO", f"steward-host-agent {AGENT_VERSION} starting "
                  f"(url={cfg['url']}, interval={cfg['interval_seconds']}s)")
@@ -353,7 +566,7 @@ def main_loop(conf_path: str) -> int:
                 _log("ERROR", f"reload failed: {e}")
             _reload_requested = False
 
-        sample = build_sample(mounts)
+        sample = build_sample(mounts, rate_state)
         buffered = buffer.drain()
         payload = build_payload(
             samples=buffered + [sample],

plugins/host_agent/plugin.yaml

@@ -1,7 +1,7 @@
 # plugins/host_agent/plugin.yaml
 name: host_agent
-version: "1.1.0"
-description: "Remote Linux host resource monitoring via a lightweight Python push agent (CPU, memory, storage, load, uptime)"
+version: "1.2.0"
+description: "Remote Linux host resource monitoring via a lightweight Python push agent (CPU incl. per-core, memory + PSI, storage, disk I/O, network throughput, load, temperatures, uptime)"
 author: "Steward"
 license: "MIT"
 min_app_version: "0.1.0"

plugins/host_agent/routes.py

@@ -95,6 +95,60 @@ def _expand_sample_to_metrics(
     if sample.get("storage"):
         row("disk_used_pct_worst", host_name, worst_pct)
 
+    # Per-core CPU. Sub-resources carry a ':' so the fleet widget filters them out.
+    for i, core_pct in enumerate(sample.get("cpu_cores") or []):
+        if core_pct is not None:
+            row("cpu_pct", f"{host_name}:core{i}", core_pct)
+
+    # Richer memory breakdown.
+    if mem.get("cached_bytes") is not None:
+        row("mem_cached_bytes", host_name, mem["cached_bytes"])
+    if mem.get("buffers_bytes") is not None:
+        row("mem_buffers_bytes", host_name, mem["buffers_bytes"])
+
+    # Network throughput — per interface plus a host-level total for the fleet view.
+    net_rx_total = net_tx_total = 0.0
+    for iface in sample.get("net") or []:
+        rx, tx = iface.get("rx_bps", 0.0), iface.get("tx_bps", 0.0)
+        res = f"{host_name}:net:{iface['iface']}"
+        row("net_rx_bps", res, rx)
+        row("net_tx_bps", res, tx)
+        net_rx_total += rx
+        net_tx_total += tx
+    if sample.get("net"):
+        row("net_rx_bps", host_name, net_rx_total)
+        row("net_tx_bps", host_name, net_tx_total)
+
+    # Disk I/O — per device plus host-level total.
+    dr_total = dw_total = 0.0
+    for dev in sample.get("diskio") or []:
+        rd, wr = dev.get("read_bps", 0.0), dev.get("write_bps", 0.0)
+        res = f"{host_name}:diskio:{dev['device']}"
+        row("disk_read_bps", res, rd)
+        row("disk_write_bps", res, wr)
+        dr_total += rd
+        dw_total += wr
+    if sample.get("diskio"):
+        row("disk_read_bps", host_name, dr_total)
+        row("disk_write_bps", host_name, dw_total)
+
+    # Temperatures — per sensor plus the host max for at-a-glance.
+    max_temp: float | None = None
+    for t in sample.get("temps") or []:
+        c = t.get("celsius")
+        if c is None:
+            continue
+        row("temp_c", f"{host_name}:temp:{t['label']}", c)
+        if max_temp is None or c > max_temp:
+            max_temp = c
+    if max_temp is not None:
+        row("temp_c_max", host_name, max_temp)
+
+    # Pressure stall information (PSI) — host-level gauges (mem/cpu/io some|full).
+    for k, v in (sample.get("psi") or {}).items():
+        if isinstance(v, (int, float)):
+            row(f"psi_{k}", host_name, v)
+
     return rows
 
 

tests/plugins/host_agent/test_collectors.py

@@ -0,0 +1,96 @@
+"""Unit tests for the host agent's new metric collectors / parsers.
+
+Pure parsers and the rate helper are tested directly; the file-reading
+collectors are tested by monkeypatching agent._read_file with fixtures.
+"""
+import plugins.host_agent.agent as agent
+
+
+def test_parse_cpu_lines_aggregate_and_per_core():
+    text = (
+        "cpu  100 0 100 800 0 0 0\n"
+        "cpu0 50 0 50 400 0\n"
+        "cpu1 50 0 50 400 0\n"
+        "intr 12345\n"
+    )
+    parsed = agent._parse_cpu_lines(text)
+    assert set(parsed) == {"cpu", "cpu0", "cpu1"}
+    assert parsed["cpu"] == (1000, 800)  # total, idle(=idle+iowait)
+
+
+def test_collect_cpu_returns_aggregate_and_cores(monkeypatch):
+    snapshots = iter([
+        "cpu 100 0 100 800\ncpu0 50 0 50 400\ncpu1 50 0 50 400\n",
+        "cpu 200 0 200 1200\ncpu0 100 0 100 600\ncpu1 100 0 100 600\n",
+    ])
+    monkeypatch.setattr(agent, "_read_file", lambda p: next(snapshots))
+    monkeypatch.setattr(agent.time, "sleep", lambda s: None)
+    agg, cores = agent.collect_cpu()
+    assert isinstance(agg, float)
+    assert len(cores) == 2
+
+
+def test_parse_psi():
+    text = (
+        "some avg10=1.50 avg60=2.00 avg300=3.00 total=12345\n"
+        "full avg10=0.50 avg60=0.10 avg300=0.00 total=678\n"
+    )
+    r = agent._parse_psi(text)
+    assert r["some_avg10"] == 1.5
+    assert r["some_avg60"] == 2.0
+    assert r["full_avg10"] == 0.5
+
+
+def test_rates_basic_and_counter_reset():
+    cur = {"eth0": (2000, 1000), "sda": (5000, 0)}
+    prev = {"eth0": (1000, 500), "sda": (10000, 0)}  # sda counter went backwards
+    out = agent._rates(cur, prev, dt=10.0)
+    assert out["eth0"] == (100.0, 50.0)
+    assert "sda" not in out                    # negative delta dropped
+    assert agent._rates(cur, prev, dt=0.0) == {}   # no elapsed time → no rates
+
+
+def test_rates_skips_names_missing_from_prev():
+    assert agent._rates({"new": (5, 5)}, {}, dt=1.0) == {}
+
+
+def test_collect_net_raw_skips_lo_and_veth(monkeypatch):
+    fixture = (
+        "Inter-|   Receive\n"
+        " face |bytes\n"
+        "    lo: 100 1 0 0 0 0 0 0 200 2 0 0 0 0 0 0\n"
+        "  eth0: 1000 5 0 0 0 0 0 0 500 4 0 0 0 0 0 0\n"
+        " veth9: 9 0 0 0 0 0 0 0 9 0 0 0 0 0 0 0\n"
+    )
+    monkeypatch.setattr(agent, "_read_file", lambda p: fixture)
+    assert agent.collect_net_raw() == {"eth0": (1000, 500)}
+
+
+def test_collect_diskio_raw_whole_disks_only(monkeypatch):
+    # fields: major minor name reads rd_merged sectors_read ms ... sectors_written ...
+    fixture = (
+        "   8       0 sda 1 0 10 0 1 0 20 0 0 0 0\n"
+        "   8       1 sda1 1 0 5 0 1 0 5 0 0 0 0\n"
+        " 259       0 nvme0n1 1 0 100 0 1 0 200 0 0 0 0\n"
+        "   7       0 loop0 1 0 1 0 1 0 1 0 0 0 0\n"
+    )
+    monkeypatch.setattr(agent, "_read_file", lambda p: fixture)
+    out = agent.collect_diskio_raw()
+    assert out == {"sda": (10 * 512, 20 * 512), "nvme0n1": (100 * 512, 200 * 512)}
+
+
+def test_build_sample_first_call_has_no_rates_but_keeps_state(monkeypatch):
+    monkeypatch.setattr(agent, "collect_cpu", lambda: (5.0, [5.0]))
+    monkeypatch.setattr(agent, "collect_memory", lambda: {"total_bytes": 10, "available_bytes": 5})
+    monkeypatch.setattr(agent, "collect_load", lambda: {"1m": 0.0, "5m": 0.0, "15m": 0.0})
+    monkeypatch.setattr(agent, "collect_uptime", lambda: 1)
+    monkeypatch.setattr(agent, "collect_storage", lambda m: [])
+    monkeypatch.setattr(agent, "collect_temps", lambda: [])
+    monkeypatch.setattr(agent, "collect_psi", lambda: {})
+    monkeypatch.setattr(agent, "collect_net_raw", lambda: {"eth0": (1000, 500)})
+    monkeypatch.setattr(agent, "collect_diskio_raw", lambda: {"sda": (10, 20)})
+    state: dict = {}
+    s1 = agent.build_sample(["/"], state)
+    assert s1["net"] == [] and s1["diskio"] == []   # no prior counters yet
+    assert state["mono"] is not None
+    assert state["net"] == {"eth0": (1000, 500)}

tests/plugins/host_agent/test_expand_metrics.py

@@ -0,0 +1,60 @@
+"""Unit tests for _expand_sample_to_metrics — sample dict → PluginMetric rows.
+
+No DB: PluginMetric objects are constructed in-memory, so we assert the
+(metric_name, resource_name) → value contract directly.
+"""
+from datetime import datetime, timezone
+
+from plugins.host_agent.routes import _expand_sample_to_metrics
+
+
+def _index(rows):
+    return {(r.metric_name, r.resource_name): r.value for r in rows}
+
+
+def test_expand_emits_all_new_metric_families():
+    sample = {
+        "ts": "2026-06-16T00:00:00+00:00",
+        "cpu_pct": 12.5,
+        "cpu_cores": [10.0, 15.0],
+        "mem": {"total_bytes": 100, "available_bytes": 40, "swap_used_bytes": 5,
+                "cached_bytes": 20, "buffers_bytes": 3},
+        "load": {"1m": 0.1, "5m": 0.2, "15m": 0.3},
+        "uptime_secs": 1000,
+        "storage": [{"mount": "/", "total_bytes": 100, "used_bytes": 50}],
+        "net": [{"iface": "eth0", "rx_bps": 1000.0, "tx_bps": 500.0}],
+        "diskio": [{"device": "sda", "read_bps": 2000.0, "write_bps": 1000.0}],
+        "temps": [{"label": "Package", "celsius": 45.0}, {"label": "Core0", "celsius": 50.0}],
+        "psi": {"mem_some_avg10": 1.5, "cpu_some_avg10": 0.0},
+    }
+    idx = _index(_expand_sample_to_metrics(sample, "alpha", datetime.now(timezone.utc)))
+
+    # per-core CPU as sub-resources
+    assert idx[("cpu_pct", "alpha:core0")] == 10.0
+    assert idx[("cpu_pct", "alpha:core1")] == 15.0
+    # richer memory
+    assert idx[("mem_cached_bytes", "alpha")] == 20
+    assert idx[("mem_buffers_bytes", "alpha")] == 3
+    # network: per-iface + host total
+    assert idx[("net_rx_bps", "alpha:net:eth0")] == 1000.0
+    assert idx[("net_rx_bps", "alpha")] == 1000.0
+    # disk I/O: per-device + host total
+    assert idx[("disk_read_bps", "alpha:diskio:sda")] == 2000.0
+    assert idx[("disk_write_bps", "alpha")] == 1000.0
+    # temps: per-sensor + host max
+    assert idx[("temp_c", "alpha:temp:Package")] == 45.0
+    assert idx[("temp_c_max", "alpha")] == 50.0
+    # PSI
+    assert idx[("psi_mem_some_avg10", "alpha")] == 1.5
+
+
+def test_expand_is_backcompat_with_old_agent_sample():
+    """An old agent (no new keys) must still expand cleanly to legacy metrics."""
+    sample = {"ts": "x", "cpu_pct": 5.0, "mem": {"total_bytes": 10, "available_bytes": 5}}
+    rows = _expand_sample_to_metrics(sample, "h", datetime.now(timezone.utc))
+    names = {r.metric_name for r in rows}
+    assert "cpu_pct" in names
+    assert not any(
+        n.startswith(("net_", "disk_read", "disk_write", "psi_", "temp_", "mem_cached"))
+        for n in names
+    )