diff --git a/plugins/host_agent/agent.py b/plugins/host_agent/agent.py index 2051c60..f78cef0 100644 --- a/plugins/host_agent/agent.py +++ b/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], diff --git a/plugins/host_agent/plugin.yaml b/plugins/host_agent/plugin.yaml index d6716e8..ab23525 100644 --- a/plugins/host_agent/plugin.yaml +++ b/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" diff --git a/plugins/host_agent/routes.py b/plugins/host_agent/routes.py index 584db94..8b7cf96 100644 --- a/plugins/host_agent/routes.py +++ b/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 diff --git a/tests/plugins/host_agent/test_collectors.py b/tests/plugins/host_agent/test_collectors.py new file mode 100644 index 0000000..33370d3 --- /dev/null +++ b/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)} diff --git a/tests/plugins/host_agent/test_expand_metrics.py b/tests/plugins/host_agent/test_expand_metrics.py new file mode 100644 index 0000000..112dd97 --- /dev/null +++ b/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 + )