ducoterra
c4bb374ccc
Three dashboards served by one FastAPI app with a SQLite history store
(raw / 1m / 1h) and hand-written SVG charts: no Prometheus, no Grafana,
no CDN, no build step. Nothing is fetched from the internet at runtime.
* system: a stdlib-only agent (agent/host_agent.py) reports /proc and
/sys counters; the dashboard turns them into rates over a labelled
window, so every number says how it was measured.
* llm: gufo/llama.cpp and vLLM /metrics are parsed by an in-house
exposition parser. Measured delta/elapsed throughput beats a native
per-request gauge, and each value carries its provenance.
* ups: a NUT wire-protocol client plus health checks that name what
they looked at. Commands are gated by NUT_ALLOW_ACTIONS,
DASH_READ_ONLY and a DANGER confirmation, re-checked server-side.
Missing values render as n/a rather than 0, and an unconfigured source
answers 404 {"configured": false} without dialling anything, so "not
deployed here" never looks like an outage.
253 tests at 94% coverage against captured real fixtures; ruff and
pyright clean.
LLM-Generated: true
243 lines
11 KiB
Python
243 lines
11 KiB
Python
"""Agent snapshot -> dashboard payload: rates, null-safety, and honest fleet totals.
|
|
|
|
The rule this file defends hardest is *null in, null out*. A missing sensor rendered as 0
|
|
is a dashboard that says "the GPU is idle" when it means "the driver said nothing", and on
|
|
a machine you are debugging at 2am that difference is the whole point.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import math
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from dashboards.rates import RateEstimator
|
|
from dashboards.sources.host import HostCollector, fleet_summary, parse_host_snapshot
|
|
|
|
TICKS_PER_SECOND = 100.0
|
|
|
|
|
|
def make_collector() -> HostCollector:
|
|
return HostCollector(RateEstimator(span=60.0))
|
|
|
|
|
|
def advance(snapshot: dict[str, Any], dt: float = 10.0, *, total_ticks: float = 1000.0,
|
|
idle_ticks: float = 900.0, read_bytes: float = 0.0, write_bytes: float = 0.0,
|
|
rx_bytes: float = 0.0, tx_bytes: float = 0.0) -> dict[str, Any]:
|
|
"""A second scrape `dt` later with counters moved by known amounts."""
|
|
snap = copy.deepcopy(snapshot)
|
|
snap["ts"] = float(snapshot["ts"]) + dt
|
|
modes = snap["cpu"]["modes"]
|
|
idle_parts = {"idle": idle_ticks * 0.9, "iowait": idle_ticks * 0.1}
|
|
for key, value in modes.items():
|
|
if key == "total":
|
|
modes["total"] = value + total_ticks
|
|
elif key in idle_parts:
|
|
modes[key] = value + idle_parts[key]
|
|
elif key != "guest" and key != "guest_nice":
|
|
modes[key] = value + (total_ticks - idle_ticks) / 5
|
|
for device in snap["disk_io"]["devices"]:
|
|
device["read_bytes_total"] += read_bytes
|
|
device["write_bytes_total"] += write_bytes
|
|
device["reads_total"] += read_bytes / 4096
|
|
device["writes_total"] += write_bytes / 4096
|
|
for iface in snap["network"]["interfaces"]:
|
|
iface["rx_bytes_total"] += rx_bytes
|
|
iface["tx_bytes_total"] += tx_bytes
|
|
return snap
|
|
|
|
|
|
def parse_twice(agent_snapshot) -> tuple[dict[str, Any], dict[str, Any]]:
|
|
collector = make_collector()
|
|
first = parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
|
|
second_snap = advance(agent_snapshot, 10.0, read_bytes=4_096_000, write_bytes=2_048_000,
|
|
rx_bytes=100_000, tx_bytes=50_000)
|
|
second = parse_host_snapshot("driveripper", second_snap, collector, now=1_010.0)
|
|
return first.payload, second.payload
|
|
|
|
|
|
def test_first_scrape_has_no_rates_but_reports_instantaneous_values(agent_snapshot) -> None:
|
|
collector = make_collector()
|
|
payload = parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0).payload
|
|
assert payload["cpu"]["usage_pct"] is None, "one sample cannot make a rate"
|
|
assert payload["cpu"]["load1"] is not None, "load is a gauge, available immediately"
|
|
assert payload["memory"]["used_pct"] is not None
|
|
assert payload["disks"]["read_bytes_per_s"] is None
|
|
assert payload["network"]["rx_bytes_per_s"] is None
|
|
assert payload["window_seconds"] is None
|
|
|
|
|
|
def test_second_scrape_measures_the_window_it_claims(agent_snapshot) -> None:
|
|
"""Two scrapes are enough — a dashboard that needs three looks to say anything about
|
|
CPU is a dashboard whose first minute of history is a hole."""
|
|
_first, payload = parse_twice(agent_snapshot)
|
|
assert payload["window_seconds"] == 10.0
|
|
usage = payload["cpu"]["usage_pct"]
|
|
assert usage is not None
|
|
# 1000 ticks total, 900 idle over 10s => exactly 10% busy, whatever the host was doing
|
|
assert usage == 10.0
|
|
|
|
|
|
def test_iowait_counts_as_idle_but_is_still_reported(agent_snapshot) -> None:
|
|
_first, payload = parse_twice(agent_snapshot)
|
|
modes = payload["cpu"]["modes_pct"]
|
|
assert modes["iowait"] is not None and modes["iowait"] > 0
|
|
assert sum(v for v in modes.values() if v is not None) <= 100.0001
|
|
|
|
|
|
def test_per_core_series_is_one_point_per_logical_cpu(agent_snapshot) -> None:
|
|
collector = make_collector()
|
|
parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
|
|
payload = parse_host_snapshot("driveripper", advance(agent_snapshot, 10.0), collector,
|
|
now=1_010.0).payload
|
|
cores = payload["cpu"]["core_usage_pct"]
|
|
assert len(cores) == payload["info"]["cpu_cores_logical"] == 48
|
|
assert all(value is None or 0.0 <= value <= 100.0 for value in cores)
|
|
|
|
|
|
def test_disk_and_network_rates_are_bytes_per_second_over_the_window(agent_snapshot) -> None:
|
|
_first, payload = parse_twice(agent_snapshot)
|
|
devices = payload["disks"]["devices"]
|
|
assert devices, "captured host has block devices"
|
|
# the fixture scrape added the same amount to every device/interface, so the per-device
|
|
# figure is delta/dt and the headline total is the sum of what the payload shows
|
|
assert devices[0]["read_bytes_per_s"] == 4_096_000 / 10.0
|
|
assert payload["disks"]["read_bytes_per_s"] == sum(
|
|
d["read_bytes_per_s"] or 0.0 for d in devices)
|
|
assert payload["disks"]["write_bytes_per_s"] == sum(
|
|
d["write_bytes_per_s"] or 0.0 for d in devices)
|
|
interfaces = payload["network"]["interfaces"]
|
|
assert payload["network"]["rx_bytes_per_s"] == 100_000 / 10.0 * len(interfaces)
|
|
assert payload["network"]["tx_bytes_per_s"] == 50_000 / 10.0 * len(interfaces)
|
|
|
|
|
|
def test_idle_machine_reports_zero_not_none_once_measured(agent_snapshot) -> None:
|
|
"""0 and n/a are different claims: 'nothing happened' vs 'we could not tell'."""
|
|
collector = make_collector()
|
|
parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
|
|
quiet = advance(agent_snapshot, 10.0, total_ticks=1_000.0, idle_ticks=1_000.0)
|
|
payload = parse_host_snapshot("driveripper", quiet, collector, now=1_010.0).payload
|
|
assert payload["cpu"]["usage_pct"] == 0.0
|
|
assert payload["disks"]["read_bytes_per_s"] == 0.0
|
|
assert payload["network"]["rx_bytes_per_s"] == 0.0
|
|
|
|
|
|
def test_missing_sensors_stay_missing(agent_snapshot) -> None:
|
|
snap = copy.deepcopy(agent_snapshot)
|
|
snap["temperatures"] = [] # the agent reports a flat sensor list
|
|
snap["cpu_temperature_celsius"] = None # ...plus its own pick for "the CPU temp"
|
|
snap["cpu"]["freq_mhz"] = None
|
|
for gpu in snap["gpus"]:
|
|
gpu["power_watts"] = None
|
|
gpu["temps"] = {}
|
|
collector = make_collector()
|
|
payload = parse_host_snapshot("driveripper", snap, collector, now=1_000.0).payload
|
|
assert payload["cpu"]["temp_c"] is None
|
|
assert payload["cpu"]["freq_mhz"] is None
|
|
assert payload["temperatures"]["max_c"] is None
|
|
assert payload["temperatures"]["sensors"] == []
|
|
for gpu in payload["gpus"]:
|
|
assert gpu["power_watts"] is None
|
|
assert gpu["temps"] == {}
|
|
assert gpu["temp_c"] is None
|
|
|
|
|
|
def test_no_sample_is_ever_written_as_zero_for_a_missing_value(agent_snapshot) -> None:
|
|
snap = copy.deepcopy(agent_snapshot)
|
|
snap["cpu"]["freq_mhz"] = None
|
|
snap["gpus"] = []
|
|
collector = make_collector()
|
|
result = parse_host_snapshot("driveripper", snap, collector, now=1_000.0)
|
|
for sample in result.samples:
|
|
assert sample.value is not None
|
|
assert math.isfinite(sample.value), f"{sample.metric} is {sample.value}"
|
|
assert not any(s.metric == "cpu.freq_mhz" for s in result.samples)
|
|
|
|
|
|
def test_gpu_pool_percent_accounts_for_vram_and_gtt(agent_snapshot) -> None:
|
|
"""On an APU the model lives in GTT; reporting only vram says '32 MB used' on a 128 GB box."""
|
|
collector = make_collector()
|
|
payload = parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0).payload
|
|
gpu = payload["gpus"][0]
|
|
assert gpu["model_full"] and gpu["model"]
|
|
assert gpu["vram_total_bytes"] > 0
|
|
assert gpu["mem_used_pct"] is not None and 0.0 <= gpu["mem_used_pct"] <= 100.0
|
|
assert "gtt_total_bytes" in gpu
|
|
|
|
|
|
def test_reboot_is_reported_as_an_event(agent_snapshot) -> None:
|
|
collector = make_collector()
|
|
parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
|
|
snap = advance(agent_snapshot, 10.0)
|
|
snap["host"]["boot_ts"] = float(agent_snapshot["host"]["boot_ts"]) + 5_000
|
|
result = parse_host_snapshot("driveripper", snap, collector, now=1_010.0)
|
|
reboots = [e for e in result.events if e.kind == "reboot"]
|
|
assert len(reboots) == 1
|
|
assert reboots[0].severity == "warn"
|
|
assert "rebooted" in reboots[0].message
|
|
|
|
|
|
def test_agent_errors_travel_to_the_ui(agent_snapshot) -> None:
|
|
snap = copy.deepcopy(agent_snapshot)
|
|
snap["errors"] = ["disk_io: permission denied reading /sys/block/sdb"]
|
|
payload = parse_host_snapshot("driveripper", snap, make_collector(), now=1_000.0).payload
|
|
assert payload["agent"]["errors"], "a partial snapshot must be labelled partial"
|
|
|
|
|
|
def test_counter_reset_from_a_reboot_does_not_chart_a_spike(agent_snapshot) -> None:
|
|
collector = make_collector()
|
|
parse_host_snapshot("driveripper", agent_snapshot, collector, now=1_000.0)
|
|
snap = copy.deepcopy(agent_snapshot)
|
|
snap["cpu"]["modes"]["total"] = 1.0 # counter restarted from zero
|
|
payload = parse_host_snapshot("driveripper", snap, collector, now=1_010.0).payload
|
|
assert payload["cpu"]["usage_pct"] is None
|
|
|
|
|
|
def test_fleet_summary_ignores_hosts_that_are_not_answering(agent_snapshot) -> None:
|
|
payload = parse_host_snapshot("driveripper", agent_snapshot, make_collector(),
|
|
now=1_000.0).payload
|
|
live = copy.deepcopy(payload)
|
|
live["cpu"]["usage_pct"] = 40.0
|
|
live["info"]["cpu_cores_logical"] = 48
|
|
dead = copy.deepcopy(payload)
|
|
dead["name"] = "deskwork"
|
|
dead["cpu"]["usage_pct"] = None
|
|
fleet = fleet_summary([live, dead])
|
|
assert fleet["hosts_total"] == 2
|
|
assert fleet["hosts_live"] == 1
|
|
assert fleet["cpu_usage_pct_avg"] == 40.0, "a dead host must not drag the average to 0"
|
|
|
|
|
|
def test_fleet_summary_weights_cpu_by_cores(agent_snapshot) -> None:
|
|
base = parse_host_snapshot("driveripper", agent_snapshot, make_collector(),
|
|
now=1_000.0).payload
|
|
big = copy.deepcopy(base)
|
|
big["cpu"]["usage_pct"] = 10.0
|
|
big["info"]["cpu_cores_logical"] = 100
|
|
small = copy.deepcopy(base)
|
|
small["name"] = "gateway"
|
|
small["cpu"]["usage_pct"] = 90.0
|
|
small["info"]["cpu_cores_logical"] = 2
|
|
fleet = fleet_summary([big, small])
|
|
assert fleet["cpu_usage_pct_avg"] == pytest.approx((10 * 100 + 90 * 2) / 102)
|
|
assert fleet["cpu_cores"] == 102
|
|
|
|
|
|
def test_fleet_summary_worst_disk_and_memory_totals(agent_snapshot) -> None:
|
|
base = parse_host_snapshot("driveripper", agent_snapshot, make_collector(),
|
|
now=1_000.0).payload
|
|
fleet = fleet_summary([base, base])
|
|
assert fleet["disk_worst_used_pct"] == pytest.approx(base["disks"]["worst_used_pct"])
|
|
assert fleet["memory_total_bytes"] == pytest.approx(2 * base["memory"]["total_bytes"])
|
|
assert fleet["memory_used_pct"] == pytest.approx(
|
|
base["memory"]["used_bytes"] / base["memory"]["total_bytes"] * 100)
|
|
|
|
|
|
def test_fleet_summary_of_nothing_is_empty_not_zero() -> None:
|
|
fleet = fleet_summary([])
|
|
assert fleet["hosts_total"] == 0
|
|
assert fleet["cpu_usage_pct_avg"] is None
|
|
assert fleet["memory_used_bytes"] is None
|
|
|