prometheus: build per-peer series from /peers

3 files changed, 56 insertions, 73 deletions

diff --git a/src/api/peers.rs b/src/api/peers.rs
index 2f93a43..b425ba6 100644
--- a/src/api/peers.rs
+++ b/src/api/peers.rs
@@ -32,9 +32,9 @@ pub(crate) enum PeerLivenessStatus {
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct EpochInfo {
-    generation: u64,
+    pub(crate) generation: u64,
     /// Hex-encoded snapshot hash; all zeros for the genesis epoch.
-    id: String,
+    pub(crate) id: String,
 }
 
 impl From<Epoch> for EpochInfo {
@@ -50,63 +50,63 @@ impl From<Epoch> for EpochInfo {
 #[serde(rename_all = "camelCase")]
 pub(crate) struct PeerStatus {
     /// The name the peer gave itself in the member list, if any (not unique).
-    alias: Option<String>,
+    pub(crate) alias: Option<String>,
     /// Where this node reaches it, or would (`Transport::addresses_of`).
-    addresses: Vec<String>,
+    pub(crate) addresses: Vec<String>,
     /// Where messages to it go now: the address, whether the peer proved itself there,
     /// and the connection there (libp2p: direct or relayed). Absent if nowhere.
-    destination: Option<Destination>,
-    validated_swarm: Option<String>,
-    status: PeerLivenessStatus,
-    handshake_complete: bool,
-    last_seen_secs_ago: Option<u64>,
-    dormant_for_secs: Option<u64>,
+    pub(crate) destination: Option<Destination>,
+    pub(crate) validated_swarm: Option<String>,
+    pub(crate) status: PeerLivenessStatus,
+    pub(crate) handshake_complete: bool,
+    pub(crate) last_seen_secs_ago: Option<u64>,
+    pub(crate) dormant_for_secs: Option<u64>,
     /// Since a sync message from it was applied (or the epoch changed).
-    last_sync_secs_ago: Option<u64>,
+    pub(crate) last_sync_secs_ago: Option<u64>,
     /// How long its heads have differed from ours with no sync message from it in
     /// between; absent while they match.
-    diverged_for_secs: Option<u64>,
-    has_pending_delivery: bool,
+    pub(crate) diverged_for_secs: Option<u64>,
+    pub(crate) has_pending_delivery: bool,
     /// Epoch the peer last advertised; absent until it has sent a message.
-    epoch: Option<EpochInfo>,
+    pub(crate) epoch: Option<EpochInfo>,
     /// Whether that epoch equals ours, i.e. whether we sync with this peer.
-    same_epoch: Option<bool>,
+    pub(crate) same_epoch: Option<bool>,
     /// Number of heads in its last keepalive; absent before one, or when it had more
     /// than 32 (then only their digest and some of them are sent).
-    heads: Option<usize>,
+    pub(crate) heads: Option<usize>,
     /// We keep it alive and sync with it on our own; other peers we only answer.
-    neighbour: bool,
+    pub(crate) neighbour: bool,
     /// Messages and payload bytes exchanged with it, by kind, since we started (or
     /// since it was last listed).
-    traffic: PeerTraffic,
-    incidents: Incidents,
+    pub(crate) traffic: PeerTraffic,
+    pub(crate) incidents: Incidents,
     /// Smoothed sync throughput (`swarm::throughput`); absent before the first large
     /// exchange.
-    sync_throughput: Option<ThroughputInfo>,
+    pub(crate) sync_throughput: Option<ThroughputInfo>,
 }
 
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct ThroughputInfo {
-    bytes_per_sec: f64,
-    samples: u32,
+    pub(crate) bytes_per_sec: f64,
+    pub(crate) samples: u32,
     /// Age of the last sample.
-    secs_ago: u64,
+    pub(crate) secs_ago: u64,
 }
 
 #[derive(Debug, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub(crate) struct PeerListResponse {
     /// Our own alias (`--alias`).
-    local_alias: Option<String>,
-    local_epoch: EpochInfo,
+    pub(crate) local_alias: Option<String>,
+    pub(crate) local_epoch: EpochInfo,
     /// Number of heads of our document. Several heads are concurrent changes no later
     /// change has merged yet.
-    local_heads: usize,
+    pub(crate) local_heads: usize,
     /// Where this node can be reached (`Transport::local_addresses`).
-    local_addresses: Vec<String>,
+    pub(crate) local_addresses: Vec<String>,
     /// By node ID.
-    peers: BTreeMap<String, PeerStatus>,
+    pub(crate) peers: BTreeMap<String, PeerStatus>,
 }
 
 #[derive(Debug, Deserialize)]
diff --git a/src/api/stats.rs b/src/api/stats.rs
index 114f644..b9314e2 100644
--- a/src/api/stats.rs
+++ b/src/api/stats.rs
@@ -242,7 +242,8 @@ pub(crate) async fn get_stats(State(state): State<AppState>, headers: HeaderMap)
         .get(header::ACCEPT)
         .and_then(|value| value.to_str().ok());
     if prometheus::wanted(accept) {
-        let body = prometheus::render(&state, &stats).await;
+        let list = super::peers::peer_list(&state).await;
+        let body = prometheus::render(&stats, &list);
         ([(header::CONTENT_TYPE, prometheus::CONTENT_TYPE)], body).into_response()
     } else {
         Json(stats).into_response()
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
index 01f1619..9180d9d 100644
--- a/src/api/stats/prometheus.rs
+++ b/src/api/stats/prometheus.rs
@@ -4,11 +4,9 @@
 
 use std::fmt::Write;
 
-use tokio::time::Instant;
-
 use super::StatsResponse;
-use crate::locks::MutexExt;
-use crate::state::{AppState, Incidents, Liveness};
+use crate::api::peers::{PeerListResponse, PeerLivenessStatus};
+use crate::state::Incidents;
 use crate::transport::Destination;
 use crate::transport::stats::{Reachability, TransportStats};
 use crate::transport::traffic::{ByKind, Counter, PeerTraffic};
@@ -128,43 +126,26 @@ struct Peer {
     destination: Option<Destination>,
 }
 
-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(),
+        })
+        .collect()
 }
 
 /// Every (direction, kind, counter) of `traffic`.
@@ -399,7 +380,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
@@ -430,7 +412,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;
@@ -807,7 +789,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",