1 files changed, 142 insertions, 92 deletions

diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 690effa..0a1d75d 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -4,12 +4,12 @@
 
 use std::fmt::Write;
 
-use tokio::time::Instant;
-
 use super::StatsResponse;
-use crate::locks::MutexExt;
-use crate::state::{AppState, Incidents, Liveness};
+use crate::api::media_type;
+use crate::api::peers::{PeerListResponse, PeerLivenessStatus};
+use crate::state::Incidents;
 use crate::transport::Destination;
+use crate::transport::paths::ClockOffset;
 use crate::transport::stats::{Reachability, TransportStats};
 use crate::transport::traffic::{ByKind, Counter, PeerTraffic};
 
@@ -19,10 +19,10 @@ pub(super) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8"
 /// plain text; browsers and curl don't name either.
 pub(super) fn wanted(accept: Option<&str>) -> bool {
     accept.is_some_and(|accept| {
-        accept.split(',').any(|item| {
-            let media = item.split(';').next().unwrap_or("").trim();
-            media == "application/openmetrics-text" || media == "text/plain"
-        })
+        accept
+            .split(',')
+            .map(media_type)
+            .any(|media| media == "application/openmetrics-text" || media == "text/plain")
     })
 }
 
@@ -50,6 +50,32 @@ impl Out {
         let _ = writeln!(self.0, " {value}");
     }
 
+    /// One histogram's series (its family declared before): each bucket counting what
+    /// fell into it and every one before (upper bounds in seconds, the last unbounded),
+    /// the sum (seconds) and the count.
+    fn histogram(
+        &mut self,
+        name: &str,
+        labels: &[(&str, &str)],
+        buckets: impl Iterator<Item = (Option<f64>, u64)>,
+        sum: f64,
+        count: u64,
+    ) {
+        let mut cumulative = 0;
+        for (le, in_bucket) in buckets {
+            cumulative += in_bucket;
+            let le = match le {
+                Some(seconds) => seconds.to_string(),
+                None => "+Inf".to_owned(),
+            };
+            let mut with_le = labels.to_vec();
+            with_le.push(("le", &le));
+            self.sample(&format!("{name}_bucket"), &with_le, cumulative as f64);
+        }
+        self.sample(&format!("{name}_sum"), labels, sum);
+        self.sample(&format!("{name}_count"), labels, count as f64);
+    }
+
     /// A family with one unlabelled sample.
     fn single(&mut self, name: &str, kind: &str, help: &str, value: f64) {
         self.family(name, kind, help);
@@ -126,45 +152,30 @@ struct Peer {
     incidents: Incidents,
     throughput: Option<f64>,
     destination: Option<Destination>,
+    clock: Option<ClockOffset>,
 }
 
-async fn peers(state: &AppState) -> Vec<Peer> {
-    let now = Instant::now();
-    let snapshot = state.peers_snapshot().await;
-    let mut aliases = crate::swarm::members::aliases(&state.doc.read().await.doc);
-    let mut peers = Vec::with_capacity(snapshot.len());
-    for (key, peer_state) in snapshot {
-        let neighbour = crate::swarm::topology::is_neighbour(state, &key);
-        let throughput = state
-            .throughput
-            .lock_or_recover()
-            .get(&key)
-            .map(|estimate| estimate.bytes_per_sec);
-        let traffic = state.traffic.peer(&key).unwrap_or_default();
-        let peer_state = peer_state.lock().await;
-        peers.push(Peer {
-            neighbour,
-            unknown: peer_state.liveness == Liveness::Unknown,
-            dormant: matches!(peer_state.liveness, Liveness::Dormant { .. }),
-            last_seen_secs: peer_state
-                .last_seen
-                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
-            last_sync_secs: peer_state
-                .last_sync_rx
-                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
-            diverged_secs: peer_state
-                .diverged_since
-                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
-            traffic,
-            incidents: peer_state.incidents,
-            throughput,
-            alias: aliases.remove(&key),
-            destination: state.transport.destination(&key),
-            key,
-        });
-    }
-    peers.sort_by(|a, b| a.key.cmp(&b.key));
-    peers
+/// The rows of `/peers` (`peers::peer_list`), as the series need them.
+fn peers(list: &PeerListResponse) -> Vec<Peer> {
+    let secs = |secs: Option<u64>| secs.map(|secs| secs as f64);
+    list.peers
+        .iter()
+        .map(|(key, peer)| Peer {
+            key: key.clone(),
+            alias: peer.alias.clone(),
+            neighbour: peer.neighbour,
+            unknown: matches!(peer.status, PeerLivenessStatus::Unknown),
+            dormant: matches!(peer.status, PeerLivenessStatus::Dormant),
+            last_seen_secs: secs(peer.last_seen_secs_ago),
+            last_sync_secs: secs(peer.last_sync_secs_ago),
+            diverged_secs: secs(peer.diverged_for_secs),
+            traffic: peer.traffic,
+            incidents: peer.incidents,
+            throughput: peer.sync_throughput.as_ref().map(|t| t.bytes_per_sec),
+            destination: peer.destination.clone(),
+            clock: peer.clock,
+        })
+        .collect()
 }
 
 /// Every (direction, kind, counter) of `traffic`.
@@ -179,6 +190,7 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou
             (direction, "blob", by.blob),
             (direction, "identity", by.identity),
             (direction, "paths", by.paths),
+            (direction, "links", by.links),
         ]
     };
     let mut series = Vec::new();
@@ -203,6 +215,19 @@ fn transports(out: &mut Out, stats: &TransportStats) {
                 *nodes as f64,
             );
         }
+        out.family(
+            "identity_relayed_nodes",
+            "gauge",
+            "Proven nodes a transport reaches only through a relay, by transport.",
+        );
+        for transport in identity.proven.keys() {
+            let relayed = identity.relayed.get(transport).copied().unwrap_or(0);
+            out.sample(
+                "identity_relayed_nodes",
+                &[("transport", transport)],
+                relayed as f64,
+            );
+        }
         out.single(
             "identity_hinted_nodes",
             "gauge",
@@ -245,7 +270,7 @@ fn transports(out: &mut Out, stats: &TransportStats) {
         out.family(
             "paths_probes_total",
             "counter",
-            "PINGs to paths of peers with several, by result.",
+            "PINGs to paths in use, by result.",
         );
         for (result, count) in [
             ("sent", counts.probes_sent),
@@ -385,7 +410,8 @@ pub(super) fn render_relay_only(stats: &super::RelayOnlyStats) -> String {
 /// A per-peer incident counter: its name, its help and how to read it.
 type IncidentSeries = (&'static str, &'static str, fn(&Incidents) -> f64);
 
-pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
+/// `stats` and `list` (`/peers`) as series: nothing read anew.
+pub(super) fn render(stats: &StatsResponse, list: &PeerListResponse) -> String {
     let mut out = Out::default();
 
     // Names for joining (`* on(peer) group_left(alias) continuum_peer_info`), kept out
@@ -416,7 +442,7 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
         "epoch_generation",
         "gauge",
         "Generation of the local epoch (remakes so far).",
-        state.local_epoch().generation as f64,
+        stats.epoch.generation as f64,
     );
 
     let document = &stats.document;
@@ -448,28 +474,14 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
     );
     let lock_wait = &document.lock_wait;
     for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] {
-        let mut cumulative = 0;
-        for (le_us, count) in waits.buckets() {
-            cumulative += count;
-            let le = match le_us {
-                Some(us) => (us as f64 / 1e6).to_string(),
-                None => "+Inf".to_owned(),
-            };
-            out.sample(
-                "document_lock_wait_seconds_bucket",
-                &[("mode", mode), ("le", &le)],
-                cumulative as f64,
-            );
-        }
-        out.sample(
-            "document_lock_wait_seconds_sum",
+        out.histogram(
+            "document_lock_wait_seconds",
             &[("mode", mode)],
+            waits
+                .buckets()
+                .map(|(le_us, count)| (le_us.map(|us| us as f64 / 1e6), count)),
             waits.sum_us as f64 / 1e6,
-        );
-        out.sample(
-            "document_lock_wait_seconds_count",
-            &[("mode", mode)],
-            waits.count as f64,
+            waits.count,
         );
     }
     out.family(
@@ -732,38 +744,47 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
     out.family(
         "propagation_delay_seconds",
         "histogram",
-        "Time from a change's commit elsewhere until applied here (includes clock skew).",
+        "Time from a change's commit elsewhere until applied here, by the peers' clocks (includes the writer's clock skew).",
     );
-    let mut cumulative = 0;
-    for bucket in &propagation.buckets {
-        cumulative += bucket.count;
-        let le = match bucket.le_ms {
-            Some(ms) => (ms as f64 / 1000.0).to_string(),
-            None => "+Inf".to_owned(),
-        };
-        out.sample(
-            "propagation_delay_seconds_bucket",
-            &[("le", &le)],
-            cumulative as f64,
-        );
-    }
-    out.sample(
-        "propagation_delay_seconds_sum",
+    out.histogram(
+        "propagation_delay_seconds",
         &[],
+        propagation
+            .buckets
+            .iter()
+            .map(|bucket| (bucket.le_ms.map(|ms| ms as f64 / 1000.0), bucket.count)),
         propagation.sum_ms as f64 / 1000.0,
-    );
-    out.sample(
-        "propagation_delay_seconds_count",
-        &[],
-        propagation.count as f64,
+        propagation.count,
     );
     out.single(
         "propagation_clock_ahead_total",
         "counter",
-        "Changes stamped later than this node's clock on arrival.",
+        "Changes stamped later than their arrival, by the peers' clocks.",
         propagation.ahead as f64,
     );
 
+    let clock = &stats.clock;
+    out.single(
+        "clock_peers_compared",
+        "gauge",
+        "Active peers whose clocks were compared with ours (from their PONGs).",
+        clock.peers as f64,
+    );
+    if let Some(offset_ms) = clock.offset_ms {
+        out.single(
+            "clock_offset_seconds",
+            "gauge",
+            "How far this node's clock is ahead of its peers' (behind: negative), the median.",
+            offset_ms as f64 / 1000.0,
+        );
+    }
+    out.single(
+        "clock_off",
+        "gauge",
+        "1 while this node's clock differs from most of its peers' by more than --clock-skew-warn.",
+        flag(clock.off),
+    );
+
     // Node totals, including peers no longer listed.
     let totals = traffic_series(&stats.traffic);
     out.family(
@@ -793,7 +814,7 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
 
     transports(&mut out, &stats.transports);
 
-    let peers = peers(state).await;
+    let peers = peers(list);
 
     out.family(
         "peer_info",
@@ -838,7 +859,7 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
     out.family(
         "peer_path_rtt_seconds",
         "gauge",
-        "Smoothed round-trip time of the PINGs on each path to a peer with several.",
+        "Smoothed round-trip time of the PINGs on each path to a peer.",
     );
     for peer in &peers {
         for path in peer.destination.iter().flat_map(|d| &d.paths) {
@@ -932,6 +953,35 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
     }
 
     out.family(
+        "peer_clock_offset_seconds",
+        "gauge",
+        "How far the peer's clock is ahead of ours (behind: negative), from its PONGs.",
+    );
+    for peer in &peers {
+        if let Some(clock) = peer.clock {
+            out.sample(
+                "peer_clock_offset_seconds",
+                &[("peer", &peer.key)],
+                clock.offset_ms as f64 / 1000.0,
+            );
+        }
+    }
+    out.family(
+        "peer_clock_offset_within_seconds",
+        "gauge",
+        "Give or take this much: half the round trip it was measured over.",
+    );
+    for peer in &peers {
+        if let Some(clock) = peer.clock {
+            out.sample(
+                "peer_clock_offset_within_seconds",
+                &[("peer", &peer.key)],
+                clock.within_ms as f64 / 1000.0,
+            );
+        }
+    }
+
+    out.family(
         "peer_messages_total",
         "counter",
         "Messages by direction and kind.",