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
218 lines
8.8 KiB
Python
218 lines
8.8 KiB
Python
"""The poll loop: isolation, last-good-value, and events that mark transitions only.
|
|
|
|
These three behaviours are why the dashboard can be trusted when something is broken, so
|
|
they are asserted directly rather than through HTTP:
|
|
|
|
* a source that throws must not take the loop, the hub, or its neighbours down;
|
|
* a failed poll keeps the previous payload and is marked stale — never zeroed;
|
|
* ok→fail and fail→ok each write exactly one event; 500 more failures write nothing;
|
|
* maintenance failing is logged, because a silent rollup failure means every long-window
|
|
chart is quietly empty (that bug is what the logging was added for).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import time
|
|
from collections.abc import Iterator
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from dashboards.poller import Hub, Job, Result
|
|
from dashboards.rates import RateEstimator
|
|
from dashboards.store import Event, Sample, SampleStore
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path: Any) -> Iterator[SampleStore]:
|
|
instance = SampleStore(tmp_path / "poller.sqlite3")
|
|
try:
|
|
yield instance
|
|
finally:
|
|
instance.close()
|
|
|
|
|
|
def ok_result(value: float = 1.0) -> Result:
|
|
return Result(payload={"value": value},
|
|
samples=[Sample("host", "gw", "cpu.usage_pct", value, time.time())])
|
|
|
|
|
|
async def failing() -> Result:
|
|
raise ConnectionError("agent is down")
|
|
|
|
|
|
async def test_a_successful_poll_stores_payload_samples_and_health(store) -> None:
|
|
job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result(7.0)))
|
|
outcome = await job.run_once(store)
|
|
assert outcome is True, job.state.error
|
|
assert job.state.ok is True and job.state.error is None
|
|
assert job.state.payload == {"value": 7.0}
|
|
assert job.state.successes == 1 and job.state.samples_written == 1
|
|
assert job.state.latency_ms is not None
|
|
assert store.recent_events(limit=5) == [], "a first successful poll is not a recovery"
|
|
|
|
|
|
async def test_a_failing_poll_keeps_the_last_good_payload(store) -> None:
|
|
"""The UI shows `stale 12s` against real numbers; it must not show zeros or n/a."""
|
|
job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result(7.0)))
|
|
await job.run_once(store)
|
|
job.fn = failing
|
|
assert await job.run_once(store) is False
|
|
assert job.state.ok is False
|
|
assert job.state.payload == {"value": 7.0}, "a dead host must not blank its last values"
|
|
assert job.state.error == "ConnectionError: agent is down"
|
|
assert job.state.as_json()["age_seconds"] is not None
|
|
|
|
|
|
async def test_transitions_write_one_event_each_and_spam_writes_none(store) -> None:
|
|
job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result()),
|
|
grace_failures=1)
|
|
await job.run_once(store)
|
|
job.fn = failing
|
|
for _ in range(5):
|
|
await job.run_once(store)
|
|
events = store.recent_events(limit=50)
|
|
down = [event for event in events if event.kind == "unreachable"]
|
|
assert len(down) == 1, "five consecutive failures must not log five events"
|
|
assert down[0].host == "gw", "events carry the display label, not the internal job key"
|
|
assert down[0].severity == "error"
|
|
|
|
job.fn = lambda: asyncio.sleep(0, ok_result())
|
|
await job.run_once(store)
|
|
events = store.recent_events(limit=50)
|
|
up = [event for event in events if event.kind == "reachable"]
|
|
assert len(up) == 1
|
|
assert up[0].message == "gw is answering again after 5 failures"
|
|
|
|
|
|
async def test_one_failure_reads_as_a_failure_not_failures(store) -> None:
|
|
job = Job("host:gw", "system", "gw", 0.5, lambda: asyncio.sleep(0, ok_result()))
|
|
await job.run_once(store)
|
|
job.fn = failing
|
|
await job.run_once(store)
|
|
job.fn = lambda: asyncio.sleep(0, ok_result())
|
|
await job.run_once(store)
|
|
up = [event for event in store.recent_events(limit=50) if event.kind == "reachable"]
|
|
assert up[0].message.endswith("after 1 failure")
|
|
|
|
|
|
async def test_grace_allows_a_blip_without_an_event(store) -> None:
|
|
"""One failure inside the grace window must not page anybody."""
|
|
job = Job("host:gw", "system", "gw", 0.5, failing, grace_failures=2)
|
|
await job.run_once(store)
|
|
assert store.recent_events(limit=10) == []
|
|
await job.run_once(store)
|
|
events = store.recent_events(limit=10)
|
|
assert [event.kind for event in events] == ["unreachable"]
|
|
assert events[0].severity == "warn", "never answered is a config problem, not an outage"
|
|
assert "never answered" in events[0].message
|
|
for _ in range(4):
|
|
await job.run_once(store)
|
|
assert len(store.recent_events(limit=10)) == 1, "one event per outage, not per poll"
|
|
|
|
|
|
async def test_a_slow_poll_never_overlaps_itself(store) -> None:
|
|
calls = 0
|
|
gate = asyncio.Event()
|
|
|
|
async def slow() -> Result:
|
|
nonlocal calls
|
|
calls += 1
|
|
await gate.wait()
|
|
return ok_result()
|
|
|
|
job = Job("host:gw", "system", "gw", 0.5, slow)
|
|
first = asyncio.create_task(job.run_once(store))
|
|
await asyncio.sleep(0.01)
|
|
assert await job.run_once(store) is False, "an overlapping poll double-counts rates"
|
|
gate.set()
|
|
assert await first is True
|
|
assert calls == 1
|
|
|
|
|
|
async def test_a_counter_that_resets_is_reported_not_silently_negative() -> None:
|
|
"""The reset path is what ships: an agent restart must not read as negative throughput."""
|
|
rates = RateEstimator(span=10.0)
|
|
first, _ = rates.rate("cpu.total", 1000.0, now=100.0)
|
|
assert first is None, "one sample cannot produce a rate"
|
|
second, window = rates.rate("cpu.total", 2000.0, now=110.0)
|
|
assert second == pytest.approx(100.0) and window == pytest.approx(10.0)
|
|
after, _ = rates.rate("cpu.total", 5.0, now=120.0) # agent restarted
|
|
assert after is None, "a reset counter must read n/a, never as negative throughput"
|
|
|
|
|
|
async def test_the_hub_runs_every_job_and_stops_cleanly(store) -> None:
|
|
counts = {"a": 0, "b": 0}
|
|
|
|
async def good(key: str) -> Result:
|
|
counts[key] += 1
|
|
return ok_result()
|
|
|
|
hub = Hub(store)
|
|
hub.add(Job("host:a", "system", "a", 0.5, lambda: good("a")))
|
|
hub.add(Job("host:b", "system", "b", 0.5, failing))
|
|
hub.add(Job("llm:off", "llm", "off", 0.5, failing, enabled=False))
|
|
hub.start()
|
|
try:
|
|
await asyncio.sleep(1.2)
|
|
finally:
|
|
await hub.stop()
|
|
assert counts["a"] >= 2, "a neighbour that throws must not slow the healthy one"
|
|
states = {state["key"]: state for state in hub.states()}
|
|
assert states["host:a"]["ok"] is True
|
|
assert states["host:b"]["ok"] is False
|
|
assert states["llm:off"]["configured"] is False, "not configured is not the same as broken"
|
|
assert hub.jobs["host:b"].state.failures >= 1
|
|
|
|
|
|
async def test_states_can_be_filtered_to_one_dashboard(store) -> None:
|
|
hub = Hub(store)
|
|
hub.add(Job("host:a", "system", "a", 0.5, lambda: asyncio.sleep(0, ok_result())))
|
|
hub.add(Job("nut", "ups", "NUT server", 0.5, lambda: asyncio.sleep(0, ok_result())))
|
|
assert [s["key"] for s in hub.states("ups")] == ["nut"]
|
|
|
|
|
|
async def test_maintenance_failure_is_logged_not_swallowed(store, monkeypatch,
|
|
caplog: pytest.LogCaptureFixture) -> None:
|
|
"""A broken rollup means every chart past the raw window is empty. Retrying quietly
|
|
turned a data-loss bug into nothing at all — this is the regression that added the log."""
|
|
def boom() -> dict[str, int]:
|
|
raise RuntimeError("no such table: sample_rollup")
|
|
|
|
monkeypatch.setattr(store, "maintain", boom)
|
|
hub = Hub(store)
|
|
task = asyncio.create_task(hub._maintain_loop(interval=0.02))
|
|
try:
|
|
with caplog.at_level(logging.ERROR, logger="dashboards.poller"):
|
|
await asyncio.sleep(0.15)
|
|
finally:
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
assert "store maintenance failed" in caplog.text
|
|
assert "no such table" in caplog.text, "the log line must carry the real reason"
|
|
|
|
|
|
async def test_maintenance_runs_from_the_hub_and_records_its_pass(store) -> None:
|
|
hub = Hub(store)
|
|
calls = []
|
|
store.maintain = lambda: calls.append(1) or {"to_minute": 0, "to_hour": 0, # type: ignore[method-assign]
|
|
"pruned": 0}
|
|
task = asyncio.create_task(hub._maintain_loop(interval=0.02))
|
|
try:
|
|
await asyncio.sleep(0.1)
|
|
finally:
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
assert calls, "the maintenance loop must actually call maintain"
|
|
|
|
|
|
async def test_events_written_by_a_job_land_in_the_store(store) -> None:
|
|
event = Event(1_700_000_000.0, "ups", "cyberpower", "status", "warn", "OL → OB")
|
|
job = Job("nut", "ups", "NUT server", 0.5,
|
|
lambda: asyncio.sleep(0, Result(payload={}, events=[event])))
|
|
await job.run_once(store)
|
|
assert [e.message for e in store.recent_events(limit=5)] == ["OL → OB"]
|