expose stats via the API

GET /admin/stats to retrieve all stats.

2 files changed, 136 insertions, 0 deletions

diff --git a/src/api/mod.rs b/src/api/mod.rs
index bf74de5..d7f14b3 100644
--- a/src/api/mod.rs
+++ b/src/api/mod.rs
@@ -4,6 +4,7 @@ pub(crate) mod auth;
 mod debug;
 mod log;
 mod peers;
+mod stats;
 mod store;
 
 use axum::{
@@ -42,6 +43,7 @@ pub(crate) fn router(state: AppState, auth_state: AuthState, debug_api: bool) ->
             "/admin/log",
             get(log::get_log_filter).put(log::put_log_filter),
         )
+        .route("/admin/stats", get(stats::get_stats))
         .merge(crate::remake::router::<AppState>());
 
     if debug_api {
diff --git a/src/api/stats.rs b/src/api/stats.rs
new file mode 100644
index 0000000..f4a368b
--- /dev/null
+++ b/src/api/stats.rs
@@ -0,0 +1,134 @@
+//! /admin/stats: how this node is doing, in numbers.
+
+use axum::{Json, extract::State};
+use serde::Serialize;
+
+use super::peers::EpochInfo;
+use crate::state::AppState;
+use crate::swarm::{members, stability, topology};
+use crate::transport::traffic::PeerTraffic;
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub(crate) struct StatsResponse {
+    uptime_secs: u64,
+    ready: bool,
+    epoch: EpochInfo,
+    document: DocumentStats,
+    catch_up: CatchUpStats,
+    topology: TopologyStats,
+    stability: StabilityStats,
+    /// HELLO_REPLYs to an earlier run of ours, ignored.
+    stale_hello_replies: u64,
+    /// All peers together, including ones no longer listed in /peers.
+    traffic: PeerTraffic,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct DocumentStats {
+    heads: usize,
+    /// Most heads after applying a sync message (concurrent changes not yet merged).
+    max_heads: usize,
+    /// Size as saved (compressed): about what a joining node downloads. Measured in the
+    /// background, at least a minute apart (longer for large documents); absent until
+    /// first measured.
+    saved_bytes: Option<u64>,
+    /// How long ago `saved_bytes` was measured.
+    saved_bytes_secs_ago: Option<u64>,
+    /// How long the last measurement took (copy and save), in milliseconds.
+    saved_bytes_took_ms: Option<u64>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct CatchUpStats {
+    /// Catch-up sources chosen (more than one: it switched).
+    sources: u32,
+    /// Seconds from start until the initial sync was done; absent until then (and on
+    /// the genesis node).
+    done_after_secs: Option<f64>,
+    /// Sync payload received until then.
+    sync_bytes: Option<u64>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct TopologyStats {
+    k: usize,
+    /// Absent until ready with a member list.
+    view: Option<TopologyView>,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct TopologyView {
+    /// Registered members, us included.
+    members: usize,
+    /// Members other than us not judged unreachable.
+    reachable: usize,
+    stable: usize,
+    neighbours: usize,
+    stable_neighbours: usize,
+    full_mesh: bool,
+}
+
+#[derive(Debug, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct StabilityStats {
+    /// As published in the member registry.
+    tier: &'static str,
+    /// Fraction of the observed time (up to a day) with a live peer.
+    connected_fraction: f64,
+    observed_secs: u64,
+}
+
+/// `GET /admin/stats`.
+pub(crate) async fn get_stats(State(state): State<AppState>) -> Json<StatsResponse> {
+    let node = state.stats.snapshot();
+
+    let (heads, tier) = {
+        let mut replica = state.doc.write().await;
+        let heads = replica.doc.get_heads().len();
+        (heads, members::tier(&replica.doc, state.local_id()))
+    };
+    let saved = state.stats.saved_size();
+
+    let (connected_fraction, observed) = stability::measured(&state);
+
+    Json(StatsResponse {
+        uptime_secs: node.uptime.as_secs(),
+        ready: state.is_ready().await,
+        epoch: state.local_epoch().into(),
+        document: DocumentStats {
+            heads,
+            max_heads: node.max_heads.max(heads),
+            saved_bytes: saved.map(|s| s.bytes),
+            saved_bytes_secs_ago: saved.map(|s| s.at.elapsed().as_secs()),
+            saved_bytes_took_ms: saved.map(|s| s.took.as_millis() as u64),
+        },
+        catch_up: CatchUpStats {
+            sources: node.catch_up_sources,
+            done_after_secs: node.caught_up.map(|c| c.after.as_secs_f64()),
+            sync_bytes: node.caught_up.map(|c| c.sync_bytes),
+        },
+        topology: TopologyStats {
+            k: state.neighbours_k,
+            view: topology::summary(&state).map(|s| TopologyView {
+                members: s.members,
+                reachable: s.reachable,
+                stable: s.stable,
+                neighbours: s.neighbours,
+                stable_neighbours: s.stable_neighbours,
+                full_mesh: s.full_mesh,
+            }),
+        },
+        stability: StabilityStats {
+            tier: tier.as_str(),
+            connected_fraction,
+            observed_secs: observed.as_secs(),
+        },
+        stale_hello_replies: node.stale_hello_replies,
+        traffic: state.traffic.total(),
+    })
+}