summaryrefslogtreecommitdiffstats
path: root/src/api/stats/prometheus.rs
diff options
context:
space:
mode:
Diffstat (limited to 'src/api/stats/prometheus.rs')
-rw-r--r--src/api/stats/prometheus.rs522
1 files changed, 522 insertions, 0 deletions
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs
new file mode 100644
index 0000000..f68ca85
--- /dev/null
+++ b/src/api/stats/prometheus.rs
@@ -0,0 +1,522 @@
+//! /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"
+ );
+ }
+}
generated by cgit v1.3.1 (git 2.54.0) at 2026-10-24 16:25:12 +0000