From f65ecfdf313658a3979a2d0add53e342ce4b97df Mon Sep 17 00:00:00 2001 From: tv Date: Fri, 30 Oct 2026 17:42:26 +0100 Subject: add dashboard --- src/api/dashboard/dashboard.html | 422 +++++++++++++++++++++++++++++++++++++++ src/api/dashboard/mod.rs | 394 ++++++++++++++++++++++++++++++++++++ 2 files changed, 816 insertions(+) create mode 100644 src/api/dashboard/dashboard.html create mode 100644 src/api/dashboard/mod.rs (limited to 'src/api/dashboard') diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html new file mode 100644 index 0000000..c87341f --- /dev/null +++ b/src/api/dashboard/dashboard.html @@ -0,0 +1,422 @@ + + + + + +continuum + + + +

continuum

+ + connecting… +

Swarm

+ +

Peers

Figures

+ + + diff --git a/src/api/dashboard/mod.rs b/src/api/dashboard/mod.rs new file mode 100644 index 0000000..adaed10 --- /dev/null +++ b/src/api/dashboard/mod.rs @@ -0,0 +1,394 @@ +//! The dashboard (`--features dashboard`): a page showing the node's figures +//! (`/admin/stats`), its peers (`/peers`) and a map of the swarm, kept up to date +//! through Server-Sent Events. +//! +//! One producer per node serves every client: while at least one is connected (or +//! asked for the state within `LINGER`), it builds the snapshot every `TICK` and, if +//! something changed, publishes it, and a JSON merge patch (RFC 7386) from the one +//! before to every client connected. Nothing changed, nothing is sent (but SSE's +//! keep-alive comments). Ages (`…SecsAgo`, `uptimeSecs` and the like, `GROWING`) grow by +//! themselves: they alone changing is no change, and the page counts them on. +//! +//! The map: the graph the member list makes (`topology::intended_graph`, as this node +//! sees it), and the links every member measures, asked for with LINKS every +//! `LINKS_EVERY` while anyone looks (`swarm::links`), the same answers for all clients. + +use std::collections::{BTreeMap, BTreeSet}; +use std::convert::Infallible; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use axum::http::{HeaderValue, header::CONTENT_TYPE}; +use axum::response::sse::{Event, KeepAlive, Sse}; +use axum::response::{Html, IntoResponse, Response}; +use axum::routing::get; +use axum::{Extension, Router}; +use serde_json::{Map, Value as JsonValue, json}; +use tokio::sync::{broadcast, watch}; +use tokio::time::Instant; + +use crate::locks::MutexExt; +use crate::state::AppState; +use crate::swarm::{links, members, topology}; +use crate::transport::Link; + +/// How often the snapshot is built and compared, while anyone looks. +const TICK: Duration = Duration::from_secs(2); +/// How often members are asked for their links, and the graph recomputed. +const LINKS_EVERY: Duration = Duration::from_secs(30); +/// A member's links older than this are left out (it didn't answer since). +const LINKS_KEPT_FOR: Duration = Duration::from_secs(90); +/// The producer runs on this long after the last request for the state. +const LINGER: Duration = Duration::from_secs(60); +/// Members asked for their links at most, each round. +const MAX_ASKED: usize = 512; + +/// A snapshot or a patch as published: its number, and its JSON. +struct Published { + seq: u64, + json: String, +} + +/// The producer and what it published, shared by all clients. +struct Hub { + state: AppState, + /// The latest snapshot; None while the producer hasn't made one (since it started). + latest: watch::Sender>>, + /// Each patch, to the clients connected. + patches: broadcast::Sender>, + control: Mutex, +} + +struct Control { + running: bool, + /// The producer stops once no client is connected and this is past. + wanted_until: Instant, +} + +impl Hub { + fn new(state: AppState) -> Arc { + let (latest, _) = watch::channel(None); + let (patches, _) = broadcast::channel(64); + Arc::new(Self { + state, + latest, + patches, + control: Mutex::new(Control { + running: false, + wanted_until: Instant::now(), + }), + }) + } + + /// Someone looks: the producer runs (started if it wasn't), for `LINGER` at least. + fn want(self: &Arc) { + let mut control = self.control.lock_or_recover(); + control.wanted_until = Instant::now() + LINGER; + if !control.running { + control.running = true; + // What it published before it stopped is stale. + self.latest.send_replace(None); + tokio::spawn(Arc::clone(self).run()); + } + } + + /// The latest snapshot, once there is one. + async fn current(&self) -> Arc { + let mut latest = self.latest.subscribe(); + loop { + // The borrow (a lock guard, not Send) ends here, before any later await. + let published = latest + .wait_for(Option::is_some) + .await + .ok() + .and_then(|current| current.clone()); + if let Some(published) = published { + return published; + } + // The sender lives as long as the hub: not reached. + tokio::time::sleep(TICK).await; + } + } + + async fn run(self: Arc) { + let mut last: Option = None; + let mut seq = 0u64; + let mut graph = Graph::default(); + let mut graph_due = Instant::now(); + loop { + if Instant::now() >= graph_due { + graph.refresh(&self.state).await; + graph_due = Instant::now() + LINKS_EVERY; + } + let snapshot = build(&self.state, &graph).await; + let changed = last + .as_ref() + .is_none_or(|old| without_ages(old) != without_ages(&snapshot)); + if changed { + seq += 1; + if let Some(old) = &last + && let Some(patch) = merge_patch(old, &snapshot) + { + let _ = self.patches.send(Arc::new(Published { + seq, + json: patch.to_string(), + })); + } + self.latest.send_replace(Some(Arc::new(Published { + seq, + json: snapshot.to_string(), + }))); + last = Some(snapshot); + } + tokio::time::sleep(TICK).await; + { + let mut control = self.control.lock_or_recover(); + if self.patches.receiver_count() == 0 && Instant::now() >= control.wanted_until { + control.running = false; + return; + } + } + } + } +} + +/// The intended graph, recomputed every `LINKS_EVERY` (it costs about `members²`), when +/// members are asked for their links too. +#[derive(Default)] +struct Graph { + intended: BTreeMap>, +} + +impl Graph { + async fn refresh(&mut self, state: &AppState) { + self.intended = topology::intended_graph(state).unwrap_or_default(); + let members: Vec = { + let replica = state.doc.read().await; + members::members(&replica.doc) + }; + for member in members + .into_iter() + .filter(|member| member != state.local_id()) + .take(MAX_ASKED) + { + let state = state.clone(); + tokio::spawn(async move { links::ask(&state, &member).await }); + } + } +} + +/// The snapshot: the node, its figures, its peers, and the map. +async fn build(state: &AppState, graph: &Graph) -> JsonValue { + let stats = serde_json::to_value(super::stats::collect(state).await).unwrap_or_default(); + let peers = serde_json::to_value(super::peers::peer_list(state).await).unwrap_or_default(); + let aliases = { + let replica = state.doc.read().await; + members::aliases(&replica.doc) + }; + let local = state.local_id(); + let intended: Map = graph + .intended + .iter() + .map(|(member, neighbours)| (member.clone(), json!(neighbours))) + .collect(); + let mut measured = Map::new(); + measured.insert( + local.to_owned(), + json!({ "secsAgo": 0, "links": links_json(&links::own_links(state).await) }), + ); + for (member, age, member_links) in state.links.replies() { + if age <= LINKS_KEPT_FOR { + measured.insert( + member, + json!({ "secsAgo": age.as_secs(), "links": links_json(&member_links) }), + ); + } + } + // Everyone on the map: in the member list's graph, or measured. + let mut members_json = Map::new(); + for member in graph.intended.keys().chain(measured.keys()) { + members_json.insert( + member.clone(), + json!({ "alias": aliases.get(member), "self": member == local }), + ); + } + json!({ + "node": { "id": local, "alias": state.alias }, + "stats": stats, + "peers": peers, + "graph": { "members": members_json, "intended": intended, "measured": measured }, + }) +} + +fn links_json(links: &[Link]) -> JsonValue { + links + .iter() + .map(|link| { + json!({ + "peer": link.peer, + "transport": link.transport, + "relayed": link.relayed, + "rttMs": link.rtt_us.map(|us| f64::from(us) / 1000.0), + "dormant": link.dormant, + }) + }) + .collect() +} + +/// Fields that grow by themselves besides those named `…SecsAgo` (the page knows +/// them too). +const GROWING: &[&str] = &[ + "uptimeSecs", + "dormantForSecs", + "observedSecs", + "lastFromNeighbourSecs", + "oldestQuarantineSecs", +]; + +/// Whether the field `key` is an age, which grows by itself. +fn is_age(key: &str) -> bool { + key == "secsAgo" || key.ends_with("SecsAgo") || GROWING.contains(&key) +} + +/// `value` without its ages, to tell whether anything else changed. +fn without_ages(value: &JsonValue) -> JsonValue { + match value { + JsonValue::Object(object) => JsonValue::Object( + object + .iter() + .filter(|(key, _)| !is_age(key)) + .map(|(key, value)| (key.clone(), without_ages(value))) + .collect(), + ), + JsonValue::Array(items) => JsonValue::Array(items.iter().map(without_ages).collect()), + other => other.clone(), + } +} + +/// The JSON merge patch (RFC 7386) from `old` to `new`, if they differ: objects field +/// by field (a field gone is null), anything else replaced whole. A null in `new` reads +/// as the field gone, as a merge patch can't tell them apart. +fn merge_patch(old: &JsonValue, new: &JsonValue) -> Option { + match (old, new) { + (JsonValue::Object(old), JsonValue::Object(new)) => { + let mut patch = Map::new(); + for key in old.keys() { + if !new.contains_key(key) { + patch.insert(key.clone(), JsonValue::Null); + } + } + for (key, value) in new { + match old.get(key) { + Some(previous) => { + if let Some(change) = merge_patch(previous, value) { + patch.insert(key.clone(), change); + } + } + None => { + patch.insert(key.clone(), value.clone()); + } + } + } + (!patch.is_empty()).then_some(JsonValue::Object(patch)) + } + (old, new) if old == new => None, + (_, new) => Some(new.clone()), + } +} + +/// The dashboard's routes (authenticated, as the rest of `/admin`), with their hub. +pub(crate) fn router(state: AppState) -> Router { + Router::new() + .route("/admin/dashboard", get(page)) + .route("/admin/dashboard/state", get(current_state)) + .route("/admin/dashboard/events", get(events)) + .layer(Extension(Hub::new(state))) +} + +/// `GET /admin/dashboard`: the page. +async fn page() -> Html<&'static str> { + Html(include_str!("dashboard.html")) +} + +/// `GET /admin/dashboard/state`: the latest snapshot, for a client that wants it whole. +async fn current_state(Extension(hub): Extension>) -> Response { + hub.want(); + let current = hub.current().await; + ( + [(CONTENT_TYPE, HeaderValue::from_static("application/json"))], + current.json.clone(), + ) + .into_response() +} + +/// `GET /admin/dashboard/events`: Server-Sent Events: a `snapshot` first, then a +/// `patch` whenever something changed (each with its number as the event's ID; a +/// client that fell behind gets a `snapshot` again). +async fn events(Extension(hub): Extension>) -> impl IntoResponse { + // Subscribed before the snapshot is read: no patch after it is missed. + let mut patches = hub.patches.subscribe(); + hub.want(); + let first = hub.current().await; + let stream = async_stream::stream! { + let mut seen = first.seq; + yield Ok::<_, Infallible>(snapshot_event(&first)); + loop { + match patches.recv().await { + Ok(patch) if patch.seq <= seen => {} + Ok(patch) => { + seen = patch.seq; + yield Ok(Event::default() + .event("patch") + .id(patch.seq.to_string()) + .data(patch.json.clone())); + } + Err(broadcast::error::RecvError::Lagged(_)) => { + let current = hub.current().await; + seen = current.seq; + yield Ok(snapshot_event(¤t)); + } + Err(broadcast::error::RecvError::Closed) => break, + } + } + }; + Sse::new(stream).keep_alive(KeepAlive::default()) +} + +fn snapshot_event(published: &Published) -> Event { + Event::default() + .event("snapshot") + .id(published.seq.to_string()) + .data(published.json.clone()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn merge_patches() { + let old = json!({ "a": 1, "b": { "c": 2, "d": [1, 2] }, "gone": true }); + let new = json!({ "a": 1, "b": { "c": 3, "d": [1, 2] }, "added": "x" }); + assert_eq!( + merge_patch(&old, &new), + Some(json!({ "b": { "c": 3 }, "gone": null, "added": "x" })) + ); + assert_eq!(merge_patch(&old, &old), None); + assert_eq!( + merge_patch(&json!([1]), &json!([1, 2])), + Some(json!([1, 2])) + ); + } + + #[test] + fn ages_alone_are_no_change() { + let old = + json!({ "uptimeSecs": 1, "peers": { "p": { "lastSeenSecsAgo": 3, "rttMs": 2 } } }); + let older = + json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 2 } } }); + assert_eq!(without_ages(&old), without_ages(&older)); + let moved = + json!({ "uptimeSecs": 9, "peers": { "p": { "lastSeenSecsAgo": 7, "rttMs": 5 } } }); + assert_ne!(without_ages(&old), without_ages(&moved)); + } +} -- cgit v1.3.1