feat(snmp): Devices widget gives a verdict, status in words and readable key readings (#5255)
CI / lint (push) Successful in 3s
CI / integration (push) Successful in 51s
CI / unit (push) Successful in 1m17s
CI / publish (push) Successful in 1m16s

#5255 "SNMP — Devices: status in words and key readings".

- Verdict first: "All N devices answering", "ups: no reply since 14m ago",
  "N of M devices not answering", or "N not polled yet".
- Devices that stopped answering come first, then those never polled, then
  the rest.
- One header row: Device / Readings / Polled.
- The name is the link; the separate "detail →" is gone.
- The hollow dot is replaced by words: "no data yet", or "no reply since …;
  readings are from then".
- Key readings: the first two in config order, gauges ahead of counters.
  Units come from the label suffix: _centisecs as a duration, _pct, _dV as
  volts, _octets as bytes total.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-07 13:27:26 -04:00
co-authored by Claude Opus 5.5
parent a720403b47
commit 0d603bc7dd
3 changed files with 138 additions and 59 deletions
+47 -10
View File
@@ -7,6 +7,7 @@ from sqlalchemy import select
from steward.auth.middleware import require_role
from steward.models.users import UserRole
from steward.core.formatting import human_bytes, human_duration
from steward.core.metric_series import latest_by_resource
from steward.models.metrics import PluginMetric
@@ -160,28 +161,64 @@ async def host_panel(host_id: str):
return await render_template("snmp/host_panel.html", devices=devices)
# Labels by suffix: how a reading reads ("uptime_centisecs" → uptime 12d 4h).
# Counters (octets, packets) only ever grow, so the raw number says little;
# a device's key readings prefer the others.
_COUNTER_SUFFIXES = ("_octets", "_packets", "_pkts", "_errors", "_discards")
def snmp_reading(label: str, value: float) -> tuple[str, str]:
"""(name, value) as a reader would say them: units taken from the label's
suffix, underscores as spaces."""
if label.endswith("_centisecs"):
return label[:-len("_centisecs")].replace("_", " "), human_duration(value / 100)
if label.endswith("_pct"):
return label[:-len("_pct")].replace("_", " "), f"{value:.0f}%"
if label.endswith("_dV"):
return label[:-len("_dV")].replace("_", " "), f"{value / 10:.1f} V"
if label.endswith("_octets"):
return label[:-len("_octets")].replace("_", " "), f"{human_bytes(value)} total"
return label.replace("_", " "), f"{value:,.0f}" if float(value).is_integer() else f"{value:.4g}"
def key_readings(oids: list[dict], readings: dict, limit: int = 2) -> list[tuple[str, str]]:
"""The first `limit` readings in config order, gauges before counters."""
labels = [o.get("label") or o.get("oid") for o in oids if isinstance(o, dict)]
present = [lb for lb in labels if lb in readings]
ordered = ([lb for lb in present if not lb.endswith(_COUNTER_SUFFIXES)]
+ [lb for lb in present if lb.endswith(_COUNTER_SUFFIXES)])
return [snmp_reading(lb, readings[lb]) for lb in ordered[:limit]]
_REACH_ORDER = {"stale": 0, "pending": 1, "up": 2}
DEVICES_SHOWN = 10
@snmp_bp.get("/widget")
@require_role(UserRole.viewer)
async def widget():
"""The dashboard SNMP widget (#5255), in the widget-shape order: a
verdict, devices not answering first with since when, then the rest, each
with its status in words and one or two readable readings."""
devices_cfg: list[dict] = current_app.config["PLUGINS"]["snmp"].get("devices", [])
device_names = [d.get("name") or d.get("host", "?") for d in devices_cfg]
async with current_app.db_sessionmaker() as db:
latest = await _latest_readings(db, device_names[:10])
latest = await _latest_readings(db, device_names)
now = datetime.now(timezone.utc)
devices = []
for cfg in devices_cfg[:10]:
for cfg in devices_cfg:
name = cfg.get("name") or cfg.get("host", "?")
devices.append({
"name": name,
"oids": cfg.get("oids", []),
**_device_state(latest.get(name), _poll_seconds(), now),
})
state = _device_state(latest.get(name), _poll_seconds(), now)
devices.append({"name": name, **state,
"key": key_readings(cfg.get("oids", []), state["readings"])})
devices.sort(key=lambda d: _REACH_ORDER[d["reach"]])
counts = {r: sum(1 for d in devices if d["reach"] == r) for r in _REACH_ORDER}
return await render_template("snmp/widget.html",
devices=devices,
total=len(devices_cfg))
devices=devices[:DEVICES_SHOWN],
more=max(0, len(devices) - DEVICES_SHOWN),
counts=counts, total=len(devices))
@snmp_bp.get("/device/<device_name>")