diff options
Diffstat (limited to 'src/api')
| -rw-r--r-- | src/api/peers.rs | 14 | ||||
| -rw-r--r-- | src/api/stats.rs | 8 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 350 |
3 files changed, 346 insertions, 26 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs index 9e907ac..cea3644 100644 --- a/src/api/peers.rs +++ b/src/api/peers.rs @@ -17,6 +17,7 @@ use crate::document::store; use crate::locks::MutexExt; use crate::state::{AppState, HelloState, Incidents, Liveness}; use crate::swarm::peer::forgotten_path; +use crate::transport::Destination; use crate::transport::traffic::PeerTraffic; #[derive(Debug, Serialize)] @@ -50,6 +51,9 @@ pub(crate) struct PeerStatus { alias: Option<String>, /// Where this node reaches it, or would (`Transport::addresses_of`). addresses: Vec<String>, + /// Where messages to it go now: the address, whether the peer proved itself there, + /// and the connection there (libp2p: direct or relayed). Absent if nowhere. + destination: Option<Destination>, validated_swarm: Option<String>, status: PeerLivenessStatus, handshake_complete: bool, @@ -65,9 +69,6 @@ pub(crate) struct PeerStatus { heads: Option<usize>, /// We keep it alive and sync with it on our own; other peers we only answer. neighbour: bool, - /// Effective round-trip time in milliseconds (smoothed, raised by requests still - /// unanswered); absent without a recent sample. - rtt_ms: Option<f64>, /// Messages and payload bytes exchanged with it, by kind, since we started (or /// since it was last listed). traffic: PeerTraffic, @@ -120,10 +121,6 @@ pub(crate) async fn list_peers( for (peer, peer_state_mutex) in snapshot { let neighbour = crate::swarm::topology::is_neighbour(&state, &peer); - let rtt_ms = state - .lock_rtt() - .effective(&peer, now, state.timing.rtt_fresh_for()) - .map(|rtt| rtt.as_secs_f64() * 1000.0); let traffic = state.traffic.peer(&peer).unwrap_or_default(); let sync_throughput = state .throughput @@ -146,11 +143,13 @@ pub(crate) async fn list_peers( let alias = aliases.remove(&peer); let addresses = state.transport.addresses_of(&peer); + let destination = state.transport.destination(&peer); peers.insert( peer, PeerStatus { alias, addresses, + destination, validated_swarm: ps.standing.swarm_id().map(str::to_owned), status, handshake_complete: ps.hello_state == HelloState::Established, @@ -163,7 +162,6 @@ pub(crate) async fn list_peers( same_epoch: ps.epoch.map(|epoch| epoch == local_epoch), heads: ps.reported_heads, neighbour, - rtt_ms, traffic, incidents: ps.incidents, sync_throughput, diff --git a/src/api/stats.rs b/src/api/stats.rs index fa64600..07521ab 100644 --- a/src/api/stats.rs +++ b/src/api/stats.rs @@ -16,6 +16,7 @@ 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)] @@ -44,6 +45,8 @@ pub(crate) struct StatsResponse { 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. @@ -93,6 +96,9 @@ struct DocumentStats { 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)] @@ -242,6 +248,7 @@ async fn collect(state: &AppState) -> StatsResponse { 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, @@ -316,5 +323,6 @@ async fn collect(state: &AppState) -> StatsResponse { (kind.as_str(), stats) }) .collect(), + transports: state.transport.stats(), } } 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.", |