From 1e491642f11dbc2e3e3d62d1d72d9ec7541ac579 Mon Sep 17 00:00:00 2001 From: Jonas Bengtsson Date: Sun, 20 Sep 2026 07:55:18 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=9A=A7=20Add=20broker=20status=20page?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 7 +- backend/app/dynsec.py | 24 +++++ backend/app/main.py | 15 ++++ backend/app/metrics.py | 123 ++++++++++++++++++++++++++ backend/app/routers/broker.py | 14 +++ backend/app/schemas.py | 37 ++++++++ backend/tests/test_api.py | 45 ++++++++++ frontend/src/App.tsx | 42 +++++++-- frontend/src/pages/Status.tsx | 160 ++++++++++++++++++++++++++++++++++ 9 files changed, 461 insertions(+), 6 deletions(-) create mode 100644 backend/app/metrics.py create mode 100644 frontend/src/pages/Status.tsx diff --git a/README.md b/README.md index e7078a8..0ddcfe8 100644 --- a/README.md +++ b/README.md @@ -179,12 +179,17 @@ Hardening done beyond the plan: accepts them on create) - [x] Administration → System page: broker status + restart (the §28 restart gap is closed end-to-end from the UI) +- [x] Broker Status dashboard (§33/§55): $SYS telemetry collected via + the existing dynsec MQTT connection (extra $SYS/# subscription + + sys-observe role grant at startup), exposed at + GET /broker/status and shown on Broker → Status (version, + uptime, connected/stored clients, subscriptions, retained + messages, message/publish counters, 1-min load averages, bytes) - [ ] Audit browsing UI (events are recorded; no viewer page yet) - [ ] Forced password-change UI (backend flag + API exist; the frontend does not enforce `must_change_password` yet) - [ ] Client metadata editing in the UI - [ ] ACL simulator (§15) -- [ ] Dashboard ($SYS monitoring, broker status, §33) - [ ] Certificate expiry monitoring - [ ] Backup and restore (logical, encrypted, §45) - [ ] Configuration drift detection (§48) diff --git a/backend/app/dynsec.py b/backend/app/dynsec.py index 6b93eda..764f2f3 100644 --- a/backend/app/dynsec.py +++ b/backend/app/dynsec.py @@ -61,6 +61,9 @@ class DynsecClient: self._lock = threading.Lock() self._responses: dict[str, queue.Queue[dict[str, Any]]] = {} self._connected = threading.Event() + # Optional tap for $SYS telemetry (app.metrics registers itself + # at startup). Receives (topic, payload) for every $SYS message. + self._event_sink = None # -- connection lifecycle ------------------------------------------------ def connect(self) -> None: @@ -118,12 +121,25 @@ class DynsecClient: return self._teardown_locked() + def set_event_sink(self, sink) -> None: + """Register a callable receiving every $SYS message (topic, payload). + + Used by app.metrics for the broker dashboard (spec sections 33, + 55). The sink must never raise: telemetry must not be able to + disturb the management channel. + """ + self._event_sink = sink + # -- paho callbacks ------------------------------------------------------- def _on_connect(self, client, userdata, flags, reason_code, properties) -> None: if reason_code == 0: # FIXED response topic — all commands reply here; requests are # matched by the correlationData echo inside the payload. client.subscribe(RESPONSE_TOPIC, qos=1) + # $SYS telemetry for the dashboard (spec sections 33, 55). + # qos 0: telemetry is periodically republished, losing one + # is harmless. + client.subscribe("$SYS/#", qos=0) self._connected.set() else: logger.error("dynsec client connect failed: %s", reason_code) @@ -149,6 +165,14 @@ class DynsecClient: def _on_message(self, client, userdata, msg) -> None: if msg.topic != RESPONSE_TOPIC: + # Not a dynsec reply: forward $SYS telemetry to the sink, if + # one is registered. Swallow sink errors — telemetry must + # never disturb the management channel. + if msg.topic.startswith("$SYS/") and self._event_sink is not None: + try: + self._event_sink(msg.topic, msg.payload) + except Exception: # noqa: BLE001 + logger.warning("$SYS event sink raised; message dropped") return try: payload = json.loads(msg.payload.decode("utf-8")) diff --git a/backend/app/main.py b/backend/app/main.py index 27c163f..d2a04ee 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -66,6 +66,21 @@ async def lifespan(app: FastAPI): try: dynsec_client.connect() logger.info("dynamic security channel connected") + from app import metrics + + dynsec_client.set_event_sink(metrics.handle_sys_message) + # Best effort: ensure the management identity can read $SYS — + # the 2.1 auto-generated admin role set does not guarantee it + # (see the compose healthcheck note). sys-observe is one of the + # plugin's default roles (read-only $SYS/#); addClientRole is + # idempotent broker-side. + try: + dynsec_client.add_client_role( + settings.mosquitto_dynsec_username, "sys-observe", priority=1 + ) + logger.info("sys-observe role granted to the management identity") + except Exception as exc: # noqa: BLE001 + logger.warning("could not grant sys-observe: %s", exc) except Exception as exc: # noqa: BLE001 - startup should not crash the API logger.warning("dynsec channel not ready yet: %s", exc) yield diff --git a/backend/app/metrics.py b/backend/app/metrics.py new file mode 100644 index 0000000..3e6ccc7 --- /dev/null +++ b/backend/app/metrics.py @@ -0,0 +1,123 @@ +"""$SYS broker telemetry (spec sections 33, 55). + +The dynsec channel's MQTT connection subscribes to $SYS/# and forwards +every message here (see DynsecClient.set_event_sink). Values are parsed +ONCE at ingest; snapshot() returns the latest known state. The broker +publishes $SYS on the configured sys_interval (10 s in the bootstrap +config), so a fresh backend needs a few seconds before the first +snapshot is populated — every field is therefore Optional and the +`available` flag says whether ANY $SYS message has been seen. +""" + +from __future__ import annotations + +import threading +from datetime import UTC, datetime +from typing import Any, Callable + +SYS_PREFIX = "$SYS/broker/" + +_lock = threading.Lock() +_values: dict[str, Any] = {} +_last_update: datetime | None = None + + +def _as_int(payload: str) -> object: + try: + return int(payload) + except ValueError: + return payload + + +def _as_float(payload: str) -> object: + try: + return float(payload) + except ValueError: + return payload + + +# $SYS topic -> (snapshot field, parser). Field names mirror +# BrokerStatusOut (schemas.py). NOTE: "retained messages/count" +# genuinely contains a space in the Mosquitto topic name. +_TOPIC_FIELDS: dict[str, tuple[str, Callable[[str], object]]] = { + f"{SYS_PREFIX}version": ("version", str), + f"{SYS_PREFIX}uptime": ("uptime", str), + f"{SYS_PREFIX}timestamp": ("build_timestamp", str), + # clients/disconnected = persistent clients currently offline, + # i.e. the "stored sessions" of the dashboard (spec section 33). + f"{SYS_PREFIX}clients/connected": ("clients_connected", _as_int), + f"{SYS_PREFIX}clients/disconnected": ("clients_disconnected", _as_int), + f"{SYS_PREFIX}clients/maximum": ("clients_maximum", _as_int), + f"{SYS_PREFIX}clients/total": ("clients_total", _as_int), + f"{SYS_PREFIX}clients/expired": ("clients_expired", _as_int), + f"{SYS_PREFIX}messages/received": ("messages_received", _as_int), + f"{SYS_PREFIX}messages/sent": ("messages_sent", _as_int), + f"{SYS_PREFIX}messages/stored": ("messages_stored", _as_int), + f"{SYS_PREFIX}messages/dropped": ("messages_dropped", _as_int), + f"{SYS_PREFIX}publish/messages/received": ("publish_received", _as_int), + f"{SYS_PREFIX}publish/messages/sent": ("publish_sent", _as_int), + f"{SYS_PREFIX}publish/messages/dropped": ("publish_dropped", _as_int), + f"{SYS_PREFIX}bytes/received": ("bytes_received", _as_int), + f"{SYS_PREFIX}bytes/sent": ("bytes_sent", _as_int), + f"{SYS_PREFIX}subscriptions/count": ("subscriptions_count", _as_int), + f"{SYS_PREFIX}retained messages/count": ("retained_count", _as_int), + # load/* are per-second averages over the last minute ("x.xx"). + f"{SYS_PREFIX}load/messages/received/1min": ( + "load_messages_received_1min", + _as_float, + ), + f"{SYS_PREFIX}load/messages/sent/1min": ( + "load_messages_sent_1min", + _as_float, + ), + f"{SYS_PREFIX}load/publish/received/1min": ( + "load_publish_received_1min", + _as_float, + ), + f"{SYS_PREFIX}load/publish/sent/1min": ( + "load_publish_sent_1min", + _as_float, + ), + f"{SYS_PREFIX}load/bytes/received/1min": ( + "load_bytes_received_1min", + _as_float, + ), + f"{SYS_PREFIX}load/bytes/sent/1min": ( + "load_bytes_sent_1min", + _as_float, + ), +} + +_FIELDS = [field for field, _ in _TOPIC_FIELDS.values()] + + +def handle_sys_message(topic: str, payload: bytes) -> None: + """Ingest one $SYS message (wired as the dynsec client's event sink).""" + entry = _TOPIC_FIELDS.get(topic) + if entry is None: + return # unknown $SYS topic — deliberately ignored + field, parser = entry + text = payload.decode("utf-8", errors="replace").strip() + global _last_update + with _lock: + _values[field] = parser(text) + _last_update = datetime.now(UTC) + + +def reset() -> None: + """Clear all telemetry (used by tests).""" + global _last_update + with _lock: + _values.clear() + _last_update = None + + +def snapshot() -> dict[str, Any]: + """Latest known $SYS state; every field None until first publish.""" + with _lock: + values = dict(_values) + last_update = _last_update + snap: dict[str, Any] = {field: values.get(field) for field in _FIELDS} + snap["available"] = last_update is not None + snap["last_update"] = last_update + return snap \ No newline at end of file diff --git a/backend/app/routers/broker.py b/backend/app/routers/broker.py index 5b3615d..b0e391f 100644 --- a/backend/app/routers/broker.py +++ b/backend/app/routers/broker.py @@ -33,6 +33,7 @@ from app.models import ( from app.schemas import ( BrokerSettingsIn, BrokerSettingsOut, + BrokerStatusOut, ConfigPreviewOut, DeployRequest, ListenerIn, @@ -316,6 +317,19 @@ def _publish_certificates(db: Session) -> None: ) +@router.get("/status", response_model=BrokerStatusOut) +def broker_status(_user: AnyReader): + """Live $SYS telemetry for the dashboard (spec sections 33, 55). + + Read-only: every management role may view (spec section 9). Fields + are None until the first $SYS publish arrives — the broker publishes + on sys_interval (10 s), so a fresh backend needs a few seconds. + """ + from app import metrics + + return BrokerStatusOut(**metrics.snapshot()) + + @router.get("/config/preview", response_model=ConfigPreviewOut) def preview_config( db: Annotated[Session, Depends(get_db)], _user: AnyReader diff --git a/backend/app/schemas.py b/backend/app/schemas.py index cdd477f..c8e5b4d 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -230,6 +230,43 @@ class ConfigPreviewOut(BaseModel): warnings: list[str] +class BrokerStatusOut(BaseModel): + """$SYS snapshot for the dashboard (spec sections 33, 55). + + Every metric is None until the first $SYS publish arrives (the + broker publishes on sys_interval, 10 s in the bootstrap config); + `available` says whether ANY $SYS message has been ingested. + """ + + available: bool + last_update: datetime | None + version: str | None = None + uptime: str | None = None + build_timestamp: str | None = None + clients_connected: int | None = None + clients_disconnected: int | None = None + clients_maximum: int | None = None + clients_total: int | None = None + clients_expired: int | None = None + messages_received: int | None = None + messages_sent: int | None = None + messages_stored: int | None = None + messages_dropped: int | None = None + publish_received: int | None = None + publish_sent: int | None = None + publish_dropped: int | None = None + bytes_received: int | None = None + bytes_sent: int | None = None + subscriptions_count: int | None = None + retained_count: int | None = None + load_messages_received_1min: float | None = None + load_messages_sent_1min: float | None = None + load_publish_received_1min: float | None = None + load_publish_sent_1min: float | None = None + load_bytes_received_1min: float | None = None + load_bytes_sent_1min: float | None = None + + class DeployRequest(BaseModel): comment: str | None = None diff --git a/backend/tests/test_api.py b/backend/tests/test_api.py index c2c5d57..8862d0a 100644 --- a/backend/tests/test_api.py +++ b/backend/tests/test_api.py @@ -194,6 +194,51 @@ def test_system_endpoints_rbac(auditor_client): ) +def test_broker_status_endpoint(admin_client, auditor_client): + """GET /broker/status returns the $SYS snapshot (spec 33/55). + + No $SYS messages are ingested in tests (no real broker), so the + endpoint must report available: False with all metrics None; after + ingesting fake $SYS messages the snapshot reflects them. Read-only: + the auditor role may view (spec section 9). + """ + from app import metrics + + metrics.reset() + + empty = admin_client.get("/api/v1/broker/status") + assert empty.status_code == 200 + body = empty.json() + assert body["available"] is False + assert body["version"] is None + assert body["clients_connected"] is None + + # Ingest fake $SYS telemetry and verify the snapshot reflects it. + metrics.handle_sys_message("$SYS/broker/version", b"2.1.2") + metrics.handle_sys_message("$SYS/broker/clients/connected", b"4") + metrics.handle_sys_message("$SYS/broker/retained messages/count", b"7") + metrics.handle_sys_message("$SYS/broker/load/messages/received/1min", b"1.25") + + body = admin_client.get("/api/v1/broker/status").json() + assert body["available"] is True + assert body["version"] == "2.1.2" + assert body["clients_connected"] == 4 + assert body["retained_count"] == 7 # topic contains a literal space + assert body["load_messages_received_1min"] == 1.25 + assert body["last_update"] is not None + + # Unknown $SYS topics are ignored without breaking the snapshot. + metrics.handle_sys_message("$SYS/broker/something/unknown", b"x") + body = admin_client.get("/api/v1/broker/status").json() + assert body["available"] is True + + # Read-only for every management role (spec section 9: auditor has + # read access to configuration/status). + assert auditor_client.get("/api/v1/broker/status").status_code == 200 + + metrics.reset() + + def test_role_acl_lifecycle(admin_client): admin_client.post("/api/v1/mqtt/roles", json={"rolename": "telemetry"}) diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index a763b2b..b773626 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -10,6 +10,7 @@ import { GroupsPage } from "./pages/Groups"; import { BrokerConfigPage } from "./pages/BrokerConfig"; import { CertificatesPage } from "./pages/Certificates"; import { SystemPage } from "./pages/System"; +import { StatusPage } from "./pages/Status"; // Navigation structure per spec section 57. Entries with a `to` route are // implemented; entries without one are planned pages, rendered greyed out @@ -21,7 +22,7 @@ const NAV_SECTIONS: { { title: "Broker", items: [ - { label: "Status", hint: "Dashboard / $SYS monitoring (Release 3)" }, + { label: "Status", to: "/broker/status" }, { label: "Configuration", to: "/broker" }, { label: "Listeners", to: "/broker#listeners" }, { label: "Bridges", hint: "Bridge management (post-MVP)" }, @@ -82,6 +83,19 @@ function RestartIndicator() { ); } +// Hash-aware active matching: an item is active only when BOTH its path +// AND its hash (if any) match the current location. NavLink's default +// prefix matching lights up every /broker* item on /broker/status, and +// hash items (#listeners, #revisions) all match on pathname alone. +function isItemActive( + to: string, + location: { pathname: string; hash: string }, +): boolean { + const [path, hash] = to.split("#"); + if (path !== location.pathname) return false; + return hash ? location.hash === `#${hash}` : location.hash === ""; +} + function Layout({ children }: { children: React.ReactNode }) { const { session, logout } = useAuth(); const location = useLocation(); @@ -92,9 +106,7 @@ function Layout({ children }: { children: React.ReactNode }) { const activeSection = NAV_SECTIONS.find((section) => section.items.some( - (item) => - item.to && - item.to.split("#")[0] === location.pathname, + (item) => item.to && isItemActive(item.to, location), ), ); @@ -129,7 +141,19 @@ function Layout({ children }: { children: React.ReactNode }) {
{section.items.map((item) => item.to ? ( - + + isItemActive(item.to!, location) + ? "active" + : "" + } + > {item.label} ) : ( @@ -206,6 +230,14 @@ export function App() { } /> + + + + } + /> +
{label}
+
+ {value} +
+
+ ); +} + +export function StatusPage() { + const status = useQuery({ + queryKey: ["broker-status"], + queryFn: () => api.get("/broker/status"), + // Match the broker's sys_interval (10 s in the bootstrap config). + refetchInterval: 10000, + }); + + const s = status.data; + + return ( +
+

Broker status

+

+ Live telemetry from the broker's $SYS topics. The broker + publishes every 10 seconds; this page refreshes to match. +

+ + {status.error && ( +
{(status.error as Error).message}
+ )} + + {!s?.available && !status.error && ( +
+ No $SYS telemetry received yet. This happens for a few seconds after + backend startup, or if the management identity lacks $SYS read access + (the sys-observe grant is applied at backend startup). +
+ )} + + {s?.available && ( + <> +
+ + + + + + +
+ +

Messages

+
+ + + + + + +
+ +

Throughput (1 min avg / s)

+
+ + + + + + +
+ +

Bytes (total)

+
+ + +
+ +

+ Last $SYS update:{" "} + {s.last_update + ? new Date(s.last_update).toLocaleTimeString() + : "—"} +

+ + )} +
+ ); +} \ No newline at end of file