//! /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::locks::MutexExt; use crate::state::{AppState, Incidents, Liveness}; use crate::transport::Destination; use crate::transport::stats::{Reachability, TransportStats}; 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); } } const DISK_COUNTERS: usize = 7; /// Name, help and value of each counter of one kind of file. fn disk_counters(c: &super::DiskStats) -> [(&'static str, &'static str, u64); DISK_COUNTERS] { [ ( "disk_reads_total", "Reads of this node's own files.", c.reads, ), ( "disk_read_bytes_total", "Bytes read from them.", c.read_bytes, ), ("disk_writes_total", "Whole files written.", c.writes), ( "disk_written_bytes_total", "Bytes of them.", c.written_bytes, ), ( "disk_appends_total", "Appends to this node's own files.", c.appends, ), ( "disk_appended_bytes_total", "Bytes appended.", c.appended_bytes, ), ( "disk_errors_total", "Failed reads, writes and appends.", c.errors, ), ] } fn escape(value: &str) -> String { value .replace('\\', "\\\\") .replace('"', "\\\"") .replace('\n', "\\n") } /// The transport of an address. fn scheme(address: &str) -> &str { address.split_once(':').map_or("", |(scheme, _)| scheme) } fn flag(value: bool) -> f64 { if value { 1.0 } else { 0.0 } } /// What we show per peer. struct Peer { key: String, alias: Option, neighbour: bool, dormant: bool, last_seen_secs: Option, last_sync_secs: Option, diverged_secs: Option, traffic: PeerTraffic, incidents: Incidents, rtt_secs: Option, throughput: Option, destination: Option, } async fn peers(state: &AppState) -> Vec { 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 rtt_secs = state .lock_rtt() .effective(&key, now, state.timing.rtt_fresh_for()) .map(|rtt| rtt.as_secs_f64()); 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, 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, rtt_secs, throughput, alias: aliases.remove(&key), destination: state.transport.destination(&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, "status", by.status), (direction, "status_reply", by.status_reply), (direction, "blob", by.blob), (direction, "identity", by.identity), (direction, "paths", by.paths), ] }; 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 } /// The transports' figures (`TransportStats`). fn transports(out: &mut Out, stats: &TransportStats) { if let Some(identity) = &stats.identity { out.family( "identity_proven_nodes", "gauge", "Nodes proven at an address, by transport.", ); for (transport, nodes) in &identity.proven { out.sample( "identity_proven_nodes", &[("transport", transport)], *nodes as f64, ); } out.single( "identity_hinted_nodes", "gauge", "Nodes known only from hints, proven nowhere.", identity.hinted as f64, ); let counts = &identity.counts; for (name, help, value) in [ ( "identity_proofs_rejected_total", "Identity proofs that didn't hold.", counts.proofs_rejected, ), ( "identity_address_changes_total", "Nodes that proved themselves at another address on a transport than before.", counts.address_changes, ), ( "identity_dropped_unproven_total", "Messages dropped because no node proved itself at their address.", counts.dropped_unproven, ), ( "identity_refused_unproven_total", "Sends refused because the peer hadn't proven where it is yet.", counts.refused_unproven, ), ( "identity_seeds_unanswered_total", "Seeds no node answered at in time.", counts.seeds_unanswered, ), ] { out.single(name, "counter", help, value as f64); } } if let Some(paths) = &stats.paths { let counts = &paths.counts; out.family( "paths_probes_total", "counter", "PINGs to paths of peers with several, by result.", ); for (result, count) in [ ("sent", counts.probes_sent), ("answered", counts.probes_answered), ("unanswered", counts.probes_unanswered), ] { out.sample("paths_probes_total", &[("result", result)], count as f64); } for (name, help, value) in [ ( "paths_send_failures_total", "Sends that failed on a path, each tried at the next.", counts.send_failures, ), ( "paths_failovers_total", "Sends that went over another path after one failed.", counts.failovers, ), ( "paths_switches_total", "Changes of the path in use for a faster one.", counts.switches, ), ] { out.single(name, "counter", help, value as f64); } } if let Some(libp2p) = &stats.libp2p { out.family( "libp2p_reachability", "gauge", "What AutoNAT says: 1 for the current status.", ); for status in Reachability::ALL { out.sample( "libp2p_reachability", &[("status", status.as_str())], flag(libp2p.reachability == status), ); } out.family( "libp2p_connections", "gauge", "Open libp2p connections, direct or through a relay.", ); for (kind, count) in [ ("direct", libp2p.connections.direct), ("relayed", libp2p.connections.relayed), ] { out.sample("libp2p_connections", &[("kind", kind)], count as f64); } let slots = &libp2p.relay_slots; out.single( "libp2p_relay_slots", "gauge", "Slots held on relays, to be reachable through them.", slots.held as f64, ); out.single( "libp2p_relays_paused", "gauge", "Relays picked that never granted a slot, not picked again for a while.", slots.paused as f64, ); out.family( "libp2p_relay_slot_events_total", "counter", "Relay slots granted, lost after being granted, and never granted.", ); for (event, count) in [ ("granted", slots.granted), ("lost", slots.lost), ("not_granted", slots.not_granted), ] { out.sample( "libp2p_relay_slot_events_total", &[("event", event)], count as f64, ); } out.family( "libp2p_hole_punches_total", "counter", "Direct connections tried over a relayed one (DCUtR), by result.", ); for (result, count) in [ ("succeeded", libp2p.hole_punches.succeeded), ("failed", libp2p.hole_punches.failed), ] { out.sample( "libp2p_hole_punches_total", &[("result", result)], count as f64, ); } let failures = &libp2p.dial_failures; out.family( "libp2p_dial_failures_total", "counter", "libp2p connections that couldn't be made, by cause.", ); for (cause, count) in [ ("unreachable", failures.unreachable), ("wrong_peer_id", failures.wrong_peer_id), ("no_addresses", failures.no_addresses), ("other", failures.other), ] { out.sample( "libp2p_dial_failures_total", &[("cause", cause)], count as f64, ); } } } /// 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 { 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 node ID and alias (always 1)."); out.sample( "info", &[ ("node_id", &stats.node_id), ("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 remakes = &stats.remakes; out.single( "remakes_undecided", "gauge", "Remakes of the local epoch not decided yet (expired ones included).", remakes.undecided as f64, ); out.single( "remakes_committed", "gauge", "Committed remakes of the local epoch: 1 while switching; more is a double commit.", remakes.committed as f64, ); out.single( "rejoining", "gauge", "1 while syncing a newer epoch from scratch.", flag(remakes.rejoining), ); out.single( "catch_up_attempts", "gauge", "Attempts at catching up with a newer epoch (0 when not catching up).", remakes.catch_up_attempts as f64, ); out.single( "quarantined", "gauge", "Quarantined change sets (in memory only).", remakes.quarantined as f64, ); if let Some(secs) = remakes.oldest_quarantine_secs { out.single( "oldest_quarantine_seconds", "gauge", "Age of the oldest quarantined change set.", secs as f64, ); } out.single( "archived_epochs", "gauge", "Old epochs' documents kept in memory.", 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( "largest_sync_received_bytes", "gauge", "The largest sync message received since start; larger transfers come as deltas.", stats.sync.largest_received_bytes 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", "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, ); for counter in 0..DISK_COUNTERS { for (i, (kind, counts)) in stats.disk.iter().enumerate() { let (name, help, value) = disk_counters(counts)[counter]; if i == 0 { out.family(name, "counter", help); } out.sample(name, &[("file", *kind)], value 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, ); } transports(&mut out, &stats.transports); 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_destination_info", "gauge", "Where messages to the peer go: its transport, whether it proved itself there, and the connection (always 1).", ); for peer in &peers { if let Some(destination) = &peer.destination { let transport = scheme(&destination.address); out.sample( "peer_destination_info", &[ ("peer", &peer.key), ("transport", transport), ("proven", if destination.proven { "true" } else { "false" }), ( "connection", destination.connection.map_or("", |c| c.as_str()), ), ], 1.0, ); } } out.family( "peer_path_rtt_seconds", "gauge", "Smoothed round-trip time of the PINGs on each path to a peer with several.", ); for peer in &peers { for path in peer.destination.iter().flat_map(|d| &d.paths) { if let Some(ms) = path.rtt_ms { out.sample( "peer_path_rtt_seconds", &[("peer", &peer.key), ("transport", scheme(&path.address))], ms / 1000.0, ); } } } out.family( "peer_path_down", "gauge", "1 while a path to the peer is down (PINGs unanswered, or a send failed).", ); for peer in &peers { for path in peer.destination.iter().flat_map(|d| &d.paths) { out.sample( "peer_path_down", &[("peer", &peer.key), ("transport", scheme(&path.address))], flag(path.down), ); } } 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_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 an unanswered HELLO).", ); 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: [IncidentSeries; 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" ); } }