path: root/src/api/stats/prometheus.rs

//! /admin/stats in the Prometheus text exposition format (version 0.0.4), for a
//! scraper asking for it. Adds per-peer series (the figures /peers shows), labelled
//! with the peer's public key.

use std::fmt::Write;

use tokio::time::Instant;

use super::StatsResponse;
use crate::state::{AppState, Incidents, Liveness};
use crate::transport::traffic::{ByKind, Counter, PeerTraffic};

pub(super) const CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8";

/// Whether the client prefers this format to JSON: Prometheus asks for OpenMetrics or
/// plain text; browsers and curl don't name either.
pub(super) fn wanted(accept: Option<&str>) -> bool {
    accept.is_some_and(|accept| {
        accept.split(',').any(|item| {
            let media = item.split(';').next().unwrap_or("").trim();
            media == "application/openmetrics-text" || media == "text/plain"
        })
    })
}

#[derive(Default)]
struct Out(String);

impl Out {
    fn family(&mut self, name: &str, kind: &str, help: &str) {
        let _ = writeln!(self.0, "# HELP continuum_{name} {help}");
        let _ = writeln!(self.0, "# TYPE continuum_{name} {kind}");
    }

    fn sample(&mut self, name: &str, labels: &[(&str, &str)], value: f64) {
        let _ = write!(self.0, "continuum_{name}");
        if !labels.is_empty() {
            self.0.push('{');
            for (i, (key, value)) in labels.iter().enumerate() {
                if i > 0 {
                    self.0.push(',');
                }
                let _ = write!(self.0, "{key}=\"{}\"", escape(value));
            }
            self.0.push('}');
        }
        let _ = writeln!(self.0, " {value}");
    }

    /// A family with one unlabelled sample.
    fn single(&mut self, name: &str, kind: &str, help: &str, value: f64) {
        self.family(name, kind, help);
        self.sample(name, &[], value);
    }
}

fn escape(value: &str) -> String {
    value
        .replace('\\', "\\\\")
        .replace('"', "\\\"")
        .replace('\n', "\\n")
}

fn flag(value: bool) -> f64 {
    if value { 1.0 } else { 0.0 }
}

/// What we show per peer.
struct Peer {
    key: String,
    alias: Option<String>,
    neighbour: bool,
    dormant: bool,
    last_seen_secs: Option<f64>,
    traffic: PeerTraffic,
    incidents: Incidents,
    rtt_secs: Option<f64>,
    throughput: Option<f64>,
}

async fn peers(state: &AppState) -> Vec<Peer> {
    let now = Instant::now();
    let snapshot: Vec<_> = state
        .peer_states
        .read()
        .await
        .iter()
        .map(|(key, peer_state)| (key.clone(), peer_state.clone()))
        .collect();
    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 rtt_secs = state
            .lock_rtt()
            .effective(&key, now, state.timing.rtt_fresh_for())
            .map(|rtt| rtt.as_secs_f64());
        let throughput = state
            .throughput
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner())
            .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,
            dormant: matches!(peer_state.liveness, Liveness::Dormant { .. }),
            last_seen_secs: peer_state
                .last_seen
                .map(|t| now.saturating_duration_since(t).as_secs_f64()),
            traffic,
            incidents: peer_state.incidents,
            rtt_secs,
            throughput,
            alias: aliases.remove(&key),
            key,
        });
    }
    peers.sort_by(|a, b| a.key.cmp(&b.key));
    peers
}

/// Every (direction, kind, counter) of `traffic`.
fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Counter)> {
    let by_kind = |direction: &'static str, by: &ByKind| {
        [
            (direction, "hello", by.hello),
            (direction, "hello_reply", by.hello_reply),
            (direction, "sync", by.sync),
            (direction, "ping", by.ping),
            (direction, "pong", by.pong),
            (direction, "blob", by.blob),
        ]
    };
    let mut series = Vec::new();
    series.extend(by_kind("sent", &traffic.sent));
    series.extend(by_kind("received", &traffic.received));
    series.extend(by_kind("refused", &traffic.refused));
    series
}

pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String {
    let mut out = Out::default();

    // Names for joining (`* on(peer) group_left(alias) continuum_peer_info`), kept out
    // of the other series so a rename doesn't start new ones.
    out.family(
        "info",
        "gauge",
        "This node's public key and alias (always 1).",
    );
    out.sample(
        "info",
        &[
            ("public_key", &stats.public_key),
            ("alias", stats.alias.as_deref().unwrap_or("")),
        ],
        1.0,
    );

    out.single(
        "uptime_seconds",
        "gauge",
        "Seconds since this node started.",
        stats.uptime_secs as f64,
    );
    out.single(
        "ready",
        "gauge",
        "1 once the node holds the swarm's document.",
        flag(stats.ready),
    );
    out.single(
        "epoch_generation",
        "gauge",
        "Generation of the local epoch (remakes so far).",
        state.local_epoch().generation as f64,
    );

    let document = &stats.document;
    out.single(
        "document_heads",
        "gauge",
        "Heads of the local document.",
        document.heads as f64,
    );
    out.single(
        "document_max_heads",
        "gauge",
        "Most heads after applying a sync message, since start.",
        document.max_heads as f64,
    );
    if let Some(bytes) = document.saved_bytes {
        out.single(
            "document_saved_bytes",
            "gauge",
            "Size of the document as saved (compressed); measured in the background.",
            bytes as f64,
        );
    }

    let catch_up = &stats.catch_up;
    out.single(
        "catch_up_sources_total",
        "counter",
        "Catch-up sources chosen (more than one: switched).",
        catch_up.sources as f64,
    );
    if let Some(secs) = catch_up.done_after_secs {
        out.single(
            "catch_up_duration_seconds",
            "gauge",
            "Seconds from start until the initial sync was done.",
            secs,
        );
    }
    if let Some(bytes) = catch_up.sync_bytes {
        out.single(
            "catch_up_sync_bytes",
            "gauge",
            "Sync payload received until the initial sync was done.",
            bytes as f64,
        );
    }

    let topology = &stats.topology;
    out.single(
        "neighbours_target",
        "gauge",
        "Target number of neighbours (k).",
        topology.k as f64,
    );
    if let Some(view) = &topology.view {
        out.single(
            "members",
            "gauge",
            "Registered members, this node included.",
            view.members as f64,
        );
        out.single(
            "reachable_members",
            "gauge",
            "Members other than this node not judged unreachable.",
            view.reachable as f64,
        );
        out.single(
            "stable_members",
            "gauge",
            "Members with the stable tier.",
            view.stable as f64,
        );
        out.single(
            "neighbours",
            "gauge",
            "Current neighbours.",
            view.neighbours as f64,
        );
        out.single(
            "stable_neighbours",
            "gauge",
            "Current neighbours with the stable tier.",
            view.stable_neighbours as f64,
        );
        out.single(
            "full_mesh",
            "gauge",
            "1 while every reachable member is a neighbour.",
            flag(view.full_mesh),
        );
    }

    let stability = &stats.stability;
    out.single(
        "stable",
        "gauge",
        "1 while this node's published tier is stable.",
        flag(stability.tier == "stable"),
    );
    out.single(
        "connected_ratio",
        "gauge",
        "Fraction of the observed time (up to a day) with a live peer.",
        stability.connected_fraction,
    );
    out.single(
        "connectivity_observed_seconds",
        "gauge",
        "Time the connected ratio is based on.",
        stability.observed_secs as f64,
    );

    out.single(
        "stale_hello_replies_total",
        "counter",
        "HELLO_REPLYs to an earlier run of this node, ignored.",
        stats.stale_hello_replies as f64,
    );

    let propagation = &stats.propagation;
    out.family(
        "propagation_delay_seconds",
        "histogram",
        "Time from a change's commit elsewhere until applied here (includes clock skew).",
    );
    let mut cumulative = 0;
    for bucket in &propagation.buckets {
        cumulative += bucket.count;
        let le = match bucket.le_ms {
            Some(ms) => (ms as f64 / 1000.0).to_string(),
            None => "+Inf".to_owned(),
        };
        out.sample(
            "propagation_delay_seconds_bucket",
            &[("le", &le)],
            cumulative as f64,
        );
    }
    out.sample(
        "propagation_delay_seconds_sum",
        &[],
        propagation.sum_ms as f64 / 1000.0,
    );
    out.sample(
        "propagation_delay_seconds_count",
        &[],
        propagation.count as f64,
    );
    out.single(
        "propagation_clock_ahead_total",
        "counter",
        "Changes stamped later than this node's clock on arrival.",
        propagation.ahead as f64,
    );

    // Node totals, including peers no longer listed.
    let totals = traffic_series(&stats.traffic);
    out.family(
        "messages_total",
        "counter",
        "Messages by direction and kind, all peers.",
    );
    for (direction, kind, counter) in &totals {
        out.sample(
            "messages_total",
            &[("direction", direction), ("kind", kind)],
            counter.messages as f64,
        );
    }
    out.family(
        "payload_bytes_total",
        "counter",
        "Payload bytes by direction and kind, all peers.",
    );
    for (direction, kind, counter) in &totals {
        out.sample(
            "payload_bytes_total",
            &[("direction", direction), ("kind", kind)],
            counter.bytes as f64,
        );
    }

    let peers = peers(state).await;

    out.family(
        "peer_info",
        "gauge",
        "A peer's alias from the member list (always 1).",
    );
    for peer in &peers {
        out.sample(
            "peer_info",
            &[
                ("peer", &peer.key),
                ("alias", peer.alias.as_deref().unwrap_or("")),
            ],
            1.0,
        );
    }

    out.family("peer_neighbour", "gauge", "1 if the peer is a neighbour.");
    for peer in &peers {
        out.sample(
            "peer_neighbour",
            &[("peer", &peer.key)],
            flag(peer.neighbour),
        );
    }
    out.family("peer_dormant", "gauge", "1 while the peer is dormant.");
    for peer in &peers {
        out.sample("peer_dormant", &[("peer", &peer.key)], flag(peer.dormant));
    }
    out.family(
        "peer_last_seen_seconds",
        "gauge",
        "Seconds since the peer's last message.",
    );
    for peer in &peers {
        if let Some(secs) = peer.last_seen_secs {
            out.sample("peer_last_seen_seconds", &[("peer", &peer.key)], secs);
        }
    }
    out.family(
        "peer_rtt_seconds",
        "gauge",
        "Effective round-trip time (smoothed, raised by unanswered requests).",
    );
    for peer in &peers {
        if let Some(rtt) = peer.rtt_secs {
            out.sample("peer_rtt_seconds", &[("peer", &peer.key)], rtt);
        }
    }
    out.family(
        "peer_sync_throughput_bytes_per_second",
        "gauge",
        "Smoothed sync throughput in large exchanges.",
    );
    for peer in &peers {
        if let Some(throughput) = peer.throughput {
            out.sample(
                "peer_sync_throughput_bytes_per_second",
                &[("peer", &peer.key)],
                throughput,
            );
        }
    }

    out.family(
        "peer_messages_total",
        "counter",
        "Messages by direction and kind.",
    );
    for peer in &peers {
        for (direction, kind, counter) in traffic_series(&peer.traffic) {
            out.sample(
                "peer_messages_total",
                &[
                    ("peer", &peer.key),
                    ("direction", direction),
                    ("kind", kind),
                ],
                counter.messages as f64,
            );
        }
    }
    out.family(
        "peer_payload_bytes_total",
        "counter",
        "Payload bytes by direction and kind.",
    );
    for peer in &peers {
        for (direction, kind, counter) in traffic_series(&peer.traffic) {
            out.sample(
                "peer_payload_bytes_total",
                &[
                    ("peer", &peer.key),
                    ("direction", direction),
                    ("kind", kind),
                ],
                counter.bytes as f64,
            );
        }
    }

    let incidents: [(&str, &str, fn(&Incidents) -> f64); 4] = [
        (
            "peer_reply_timeouts_total",
            "Sync messages unanswered until the reply timeout.",
            |i| i.reply_timeouts as f64,
        ),
        (
            "peer_heals_total",
            "Re-handshakes because documents differed while sync was idle.",
            |i| i.heals as f64,
        ),
        (
            "peer_dormancies_total",
            "Times the peer went dormant.",
            |i| i.dormancies as f64,
        ),
        (
            "peer_dormant_seconds_total",
            "Time spent dormant, up to the last return.",
            |i| i.dormant_ms as f64 / 1000.0,
        ),
    ];
    for (name, help, value) in incidents {
        out.family(name, "counter", help);
        for peer in &peers {
            out.sample(name, &[("peer", &peer.key)], value(&peer.incidents));
        }
    }

    out.0
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn negotiation() {
        let prometheus = "application/openmetrics-text;version=1.0.0,application/openmetrics-text;version=0.0.1;q=0.75,text/plain;version=0.0.4;q=0.5,*/*;q=0.1";
        assert!(wanted(Some(prometheus)));
        assert!(wanted(Some("text/plain")));
        assert!(!wanted(Some("*/*")));
        assert!(!wanted(Some("application/json")));
        assert!(!wanted(Some("text/html,application/xhtml+xml,*/*;q=0.8")));
        assert!(!wanted(None));
    }

    #[test]
    fn sample_lines() {
        let mut out = Out::default();
        out.single("x_total", "counter", "An x.", 3.0);
        out.sample("y", &[("peer", "a\"b"), ("kind", "sync")], 0.25);
        assert_eq!(
            out.0,
            "# HELP continuum_x_total An x.\n# TYPE continuum_x_total counter\ncontinuum_x_total 3\ncontinuum_y{peer=\"a\\\"b\",kind=\"sync\"} 0.25\n"
        );
    }
}