diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/peers.rs | 56 | ||||
| -rw-r--r-- | src/api/stats.rs | 3 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 70 |
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", |