diff options
Diffstat (limited to 'src/api/stats.rs')
| -rw-r--r-- | src/api/stats.rs | 199 |
1 files changed, 199 insertions, 0 deletions
diff --git a/src/api/stats.rs b/src/api/stats.rs new file mode 100644 index 0000000..838cc0f --- /dev/null +++ b/src/api/stats.rs @@ -0,0 +1,199 @@ +//! /admin/stats: how this node is doing, in numbers. JSON, or the Prometheus text +//! format when the client asks for it (`prometheus`). + +mod prometheus; + +use axum::{ + Json, + extract::State, + http::{HeaderMap, header}, + response::{IntoResponse, Response}, +}; +use serde::Serialize; + +use super::peers::EpochInfo; +use crate::state::{AppState, Incidents}; +use crate::swarm::{members, stability, topology}; +use crate::transport::traffic::PeerTraffic; + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct StatsResponse { + public_key: String, + alias: Option<String>, + 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, + /// The peers listed in /peers together (a forgotten peer's are gone). + incidents: Incidents, + propagation: PropagationStats, +} + +/// How long changes written elsewhere took from their commit there until applied here, +/// once caught up. Includes the clock skew between the writer and us. +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct PropagationStats { + count: u64, + sum_ms: u64, + /// Changes stamped later than our clock on arrival (writer's clock ahead). + ahead: u64, + /// Not cumulative; the last bucket has no bound. + buckets: Vec<PropagationBucket>, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +struct PropagationBucket { + le_ms: Option<u64>, + count: u64, +} + +#[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>, headers: HeaderMap) -> Response { + let stats = collect(&state).await; + let accept = headers + .get(header::ACCEPT) + .and_then(|value| value.to_str().ok()); + if prometheus::wanted(accept) { + let body = prometheus::render(&state, &stats).await; + ([(header::CONTENT_TYPE, prometheus::CONTENT_TYPE)], body).into_response() + } else { + Json(stats).into_response() + } +} + +async fn collect(state: &AppState) -> StatsResponse { + let node = state.stats.snapshot(); + + let (heads, tier) = { + let replica = state.doc.read().await; + let heads = replica.heads().len(); + (heads, members::tier(&replica.doc, state.local_id())) + }; + let saved = state.stats.saved_size(); + + 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); + } + + StatsResponse { + public_key: state.local_id().to_owned(), + alias: state.alias.clone(), + 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(), + incidents, + propagation: PropagationStats { + count: node.propagation.count, + sum_ms: node.propagation.sum_ms, + ahead: node.propagation.ahead, + buckets: node + .propagation + .buckets() + .map(|(le_ms, count)| PropagationBucket { le_ms, count }) + .collect(), + }, + } +} |