path: root/src/api/stats.rs

//! /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 std::collections::BTreeMap;

use serde::Serialize;

use super::peers::EpochInfo;
use crate::state::{AppState, Incidents};
use crate::swarm::{members, stability, topology};
use crate::transport::stats::TransportStats;
use crate::transport::traffic::PeerTraffic;

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct StatsResponse {
    node_id: String,
    /// Where this node can be reached (`Transport::local_addresses`).
    addresses: Vec<String>,
    alias: Option<String>,
    uptime_secs: u64,
    ready: bool,
    epoch: EpochInfo,
    document: DocumentStats,
    catch_up: CatchUpStats,
    remakes: RemakeStats,
    sync: SyncStats,
    relay: RelayStats,
    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,
    /// This node's own files (`files`), by kind.
    disk: BTreeMap<&'static str, DiskStats>,
    /// How the transports find and reach peers (`Transport::stats`).
    transports: TransportStats,
}

/// Reads and writes of one kind of file, since the node started.
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
struct DiskStats {
    reads: u64,
    read_bytes: u64,
    writes: u64,
    written_bytes: u64,
    appends: u64,
    appended_bytes: u64,
    errors: u64,
}

/// 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>,
    /// How long takers of the document's lock waited for it, since start: the node's
    /// work on the one document queues there.
    lock_wait: crate::document::lock::LockWaits,
}

#[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>,
}

/// 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,
    /// The largest sync message received since start (its Automerge part). Larger
    /// transfers come as deltas (blobs).
    largest_received_bytes: u64,
}

/// 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")]
struct RemakeStats {
    /// Remakes of the local epoch not decided yet (expired ones included).
    undecided: usize,
    /// Committed remakes of the local epoch: 1 while switching; more is a double commit.
    committed: usize,
    /// Syncing a newer epoch from scratch.
    rejoining: bool,
    /// Attempts at catching up with a newer epoch (0 when not catching up).
    catch_up_attempts: u32,
    /// Quarantined change sets (`GET /admin/quarantine`), and the oldest one's age.
    quarantined: usize,
    oldest_quarantine_secs: Option<u64>,
    /// Old epochs' documents kept in memory.
    archived_epochs: usize,
}

#[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,
}

/// What `/admin/stats` of a node that only relays (`--relay-only`) says.
#[derive(Clone)]
pub(crate) struct RelayOnly {
    pub(crate) transport: std::sync::Arc<dyn crate::transport::Transport>,
    pub(crate) started: tokio::time::Instant,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct RelayOnlyStats {
    /// The relay's libp2p peer ID.
    peer_id: String,
    /// Where nodes reach it, as addresses to list (`_continuum/relays`).
    addresses: Vec<String>,
    uptime_secs: u64,
    transports: TransportStats,
}

/// `GET /admin/stats` of a node that only relays.
#[cfg_attr(not(feature = "libp2p"), allow(dead_code))]
pub(crate) async fn get_relay_only_stats(
    State(relay): State<RelayOnly>,
    headers: HeaderMap,
) -> Response {
    let stats = RelayOnlyStats {
        peer_id: relay.transport.local_id().to_owned(),
        addresses: relay
            .transport
            .local_addresses()
            .into_iter()
            .map(|address| format!("libp2p:{address}"))
            .collect(),
        uptime_secs: relay.started.elapsed().as_secs(),
        transports: relay.transport.stats(),
    };
    let accept = headers
        .get(header::ACCEPT)
        .and_then(|value| value.to_str().ok());
    if prometheus::wanted(accept) {
        let body = prometheus::render_relay_only(&stats);
        ([(header::CONTENT_TYPE, prometheus::CONTENT_TYPE)], body).into_response()
    } else {
        Json(stats).into_response()
    }
}

/// `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, remakes) = {
        let replica = state.doc.read().await;
        let heads = replica.heads().len();
        let remakes = crate::remake::summary(&state.remake, &replica.doc, replica.epoch);
        (
            heads,
            members::tier(&replica.doc, state.local_id()),
            remakes,
        )
    };
    let now_ms = state.clock.now_millis();
    let saved = state.stats.saved_size();

    let (connected_fraction, observed) = stability::measured(state);

    let mut incidents = Incidents::default();
    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 {
        node_id: state.local_id().to_owned(),
        addresses: state.transport.local_addresses(),
        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),
            lock_wait: state.doc.waits(),
        },
        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),
        },
        remakes: RemakeStats {
            undecided: remakes.undecided,
            committed: remakes.committed,
            rejoining: remakes.rejoining,
            catch_up_attempts: remakes.catch_up_attempts,
            quarantined: remakes.quarantined,
            oldest_quarantine_secs: remakes
                .oldest_quarantine
                .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,
            largest_received_bytes: node.largest_sync_received,
        },
        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 {
                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(),
        },
        disk: state
            .files
            .counts()
            .into_iter()
            .map(|(kind, counts)| {
                let stats = DiskStats {
                    reads: counts.reads,
                    read_bytes: counts.read_bytes,
                    writes: counts.writes,
                    written_bytes: counts.written_bytes,
                    appends: counts.appends,
                    appended_bytes: counts.appended_bytes,
                    errors: counts.errors,
                };
                (kind.as_str(), stats)
            })
            .collect(),
        transports: state.transport.stats(),
    }
}