diff options
Diffstat (limited to 'src/api/stats/prometheus.rs')
| -rw-r--r-- | src/api/stats/prometheus.rs | 350 |
1 files changed, 332 insertions, 18 deletions
diff --git a/src/api/stats/prometheus.rs b/src/api/stats/prometheus.rs index 01cd271..9cbeff4 100644 --- a/src/api/stats/prometheus.rs +++ b/src/api/stats/prometheus.rs @@ -9,6 +9,8 @@ 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"; @@ -101,6 +103,11 @@ fn escape(value: &str) -> String { .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 } } @@ -116,8 +123,8 @@ struct Peer { diverged_secs: Option<f64>, traffic: PeerTraffic, incidents: Incidents, - rtt_secs: Option<f64>, throughput: Option<f64>, + destination: Option<Destination>, } async fn peers(state: &AppState) -> Vec<Peer> { @@ -127,10 +134,6 @@ async fn peers(state: &AppState) -> Vec<Peer> { 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() @@ -152,9 +155,9 @@ async fn peers(state: &AppState) -> Vec<Peer> { .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, }); } @@ -169,10 +172,11 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou (direction, "hello", by.hello), (direction, "hello_reply", by.hello_reply), (direction, "sync", by.sync), - (direction, "ping", by.ping), - (direction, "pong", by.pong), + (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(); @@ -182,6 +186,181 @@ fn traffic_series(traffic: &PeerTraffic) -> Vec<(&'static str, &'static str, Cou 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); @@ -241,6 +420,94 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { ); } + out.family( + "document_lock_wait_seconds", + "histogram", + "Time waited for the document's lock, by mode (read, write).", + ); + let lock_wait = &document.lock_wait; + for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] { + let mut cumulative = 0; + for (le_us, count) in waits.buckets() { + cumulative += count; + let le = match le_us { + Some(us) => (us as f64 / 1e6).to_string(), + None => "+Inf".to_owned(), + }; + out.sample( + "document_lock_wait_seconds_bucket", + &[("mode", mode), ("le", &le)], + cumulative as f64, + ); + } + out.sample( + "document_lock_wait_seconds_sum", + &[("mode", mode)], + waits.sum_us as f64 / 1e6, + ); + out.sample( + "document_lock_wait_seconds_count", + &[("mode", mode)], + waits.count as f64, + ); + } + out.family( + "document_lock_wait_max_seconds", + "gauge", + "Longest wait for the document's lock since start, by mode.", + ); + for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] { + out.sample( + "document_lock_wait_max_seconds", + &[("mode", mode)], + waits.max_us as f64 / 1e6, + ); + } + + out.family( + "document_lock_wait_seconds", + "histogram", + "Time waited for the document's lock, by mode (read, write).", + ); + let lock_wait = &document.lock_wait; + for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] { + let mut cumulative = 0; + for (le_us, count) in waits.buckets() { + cumulative += count; + let le = match le_us { + Some(us) => (us as f64 / 1e6).to_string(), + None => "+Inf".to_owned(), + }; + out.sample( + "document_lock_wait_seconds_bucket", + &[("mode", mode), ("le", &le)], + cumulative as f64, + ); + } + out.sample( + "document_lock_wait_seconds_sum", + &[("mode", mode)], + waits.sum_us as f64 / 1e6, + ); + out.sample( + "document_lock_wait_seconds_count", + &[("mode", mode)], + waits.count as f64, + ); + } + out.family( + "document_lock_wait_max_seconds", + "gauge", + "Longest wait for the document's lock since start, by mode.", + ); + for (mode, waits) in [("read", &lock_wait.read), ("write", &lock_wait.write)] { + out.sample( + "document_lock_wait_max_seconds", + &[("mode", mode)], + waits.max_us as f64 / 1e6, + ); + } + let catch_up = &stats.catch_up; out.single( "catch_up_sources_total", @@ -503,6 +770,8 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { ); } + transports(&mut out, &stats.transports); + let peers = peers(state).await; out.family( @@ -521,6 +790,61 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { ); } + 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( @@ -564,16 +888,6 @@ pub(super) async fn render(state: &AppState, stats: &StatsResponse) -> String { } } 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.", |