diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/stats.rs | 49 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 68 |
2 files changed, 114 insertions, 3 deletions
diff --git a/src/api/stats.rs b/src/api/stats.rs index 5d32d82..a34d4ed 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -29,6 +29,8 @@ pub(crate) struct StatsResponse { document: DocumentStats, catch_up: CatchUpStats, remakes: RemakeStats, + sync: SyncStats, + relay: RelayStats, topology: TopologyStats, stability: StabilityStats, /// HELLO_REPLYs to an earlier run of ours, ignored. @@ -103,6 +105,29 @@ struct CatchUpStats { sync_bytes: Option<u64>, } +/// How sync with the neighbours goes. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct SyncStats { + /// Since a sync message from any neighbour was applied; absent if none ever was. A + /// node whose writes go nowhere shows it growing. + last_from_neighbour_secs: Option<u64>, + /// Peers whose heads differ from ours while sync is idle (`peers.md`, healing). + diverged_peers: usize, +} + +/// Passing changes on (`swarm::relay`) and serving blobs, since start. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct RelayStats { + announce_rounds: u64, + pulls: u64, + /// Announcements of heads we lacked that waited for the pull in progress. + pulls_queued: u64, + /// Blob requests not answered: too many were being answered. + blob_requests_dropped: u64, +} + /// Remakes and their aftermath here (`remake::Summary`). #[derive(Debug, Serialize)] #[serde(rename_all = "camelCase")] @@ -186,9 +211,17 @@ async fn collect(state: &AppState) -> StatsResponse { let (connected_fraction, observed) = stability::measured(state); let mut incidents = Incidents::default(); - let peers: Vec<_> = state.peer_states.read().await.values().cloned().collect(); - for peer_state in peers { - incidents.add(&peer_state.lock().await.incidents); + let mut last_from_neighbour: Option<tokio::time::Instant> = None; + let mut diverged_peers = 0; + for (peer, peer_state) in state.peers_snapshot().await { + let peer_state = peer_state.lock().await; + incidents.add(&peer_state.incidents); + if peer_state.diverged_since.is_some() { + diverged_peers += 1; + } + if topology::is_neighbour(state, &peer) { + last_from_neighbour = last_from_neighbour.max(peer_state.last_sync_rx); + } } StatsResponse { @@ -220,6 +253,16 @@ async fn collect(state: &AppState) -> StatsResponse { .map(|created| now_ms.ms_since(created) / 1000), archived_epochs: remakes.archived_epochs, }, + sync: SyncStats { + last_from_neighbour_secs: last_from_neighbour.map(|at| at.elapsed().as_secs()), + diverged_peers, + }, + relay: RelayStats { + announce_rounds: node.relay.announce_rounds, + pulls: node.relay.pulls, + pulls_queued: node.relay.pulls_queued, + blob_requests_dropped: node.blob_requests_dropped, + }, topology: TopologyStats { k: state.neighbours_k, view: topology::summary(state).map(|s| TopologyView { diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 6fd37a0..3baa125 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -112,6 +112,8 @@ struct Peer { neighbour: bool, dormant: bool, last_seen_secs: Option<f64>, + last_sync_secs: Option<f64>, + diverged_secs: Option<f64>, traffic: PeerTraffic, incidents: Incidents, rtt_secs: Option<f64>, @@ -142,6 +144,12 @@ async fn peers(state: &AppState) -> Vec<Peer> { 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, rtt_secs, @@ -303,6 +311,46 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { remakes.archived_epochs as f64, ); + if let Some(secs) = stats.sync.last_from_neighbour_secs { + out.single( + "last_sync_from_neighbour_seconds", + "gauge", + "Seconds since a sync message from any neighbour was applied.", + secs as f64, + ); + } + out.single( + "diverged_peers", + "gauge", + "Peers whose heads differ from ours while sync is idle.", + stats.sync.diverged_peers as f64, + ); + let relay = &stats.relay; + out.single( + "announce_rounds_total", + "counter", + "Announcement rounds (our new heads to the neighbours).", + relay.announce_rounds as f64, + ); + out.single( + "pulls_total", + "counter", + "Pulls of changes we lacked, started.", + relay.pulls as f64, + ); + out.single( + "pulls_queued_total", + "counter", + "Announcements of heads we lacked that waited for the pull in progress.", + relay.pulls_queued as f64, + ); + out.single( + "blob_requests_dropped_total", + "counter", + "Blob requests not answered: too many were being answered.", + relay.blob_requests_dropped as f64, + ); + let topology = &stats.topology; out.single( "neighbours_target", @@ -490,6 +538,26 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { } } out.family( + "peer_last_sync_seconds", + "gauge", + "Seconds since a sync message from the peer was applied (or the epoch changed).", + ); + for peer in &peers { + if let Some(secs) = peer.last_sync_secs { + out.sample("peer_last_sync_seconds", &[("peer", &peer.key)], secs); + } + } + out.family( + "peer_diverged_seconds", + "gauge", + "Seconds the peer's heads have differed from ours while sync was idle.", + ); + for peer in &peers { + if let Some(secs) = peer.diverged_secs { + out.sample("peer_diverged_seconds", &[("peer", &peer.key)], secs); + } + } + out.family( "peer_rtt_seconds", "gauge", "Effective round-trip time (smoothed, raised by unanswered requests).", |