diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/api/peers.rs | 6 | ||||
| -rw-r--r-- | src/api/stats.rs | 4 | ||||
| -rw-r--r-- | src/api/stats/prometheus.rs | 174 | ||||
| -rw-r--r-- | src/simtest/identity.rs | 19 | ||||
| -rw-r--r-- | src/transport/firewall.rs | 12 | ||||
| -rw-r--r-- | src/transport/identity.rs | 77 | ||||
| -rw-r--r-- | src/transport/mod.rs | 51 | ||||
| -rw-r--r-- | src/transport/p2p.rs | 212 | ||||
| -rw-r--r-- | src/transport/stats.rs | 141 | ||||
| -rw-r--r-- | src/transport/traffic.rs | 12 |
10 files changed, 684 insertions, 24 deletions
diff --git a/src/api/peers.rs b/src/api/peers.rs index 9e907ac..ac73296 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, @@ -146,11 +150,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, diff --git a/src/api/stats.rs b/src/api/stats.rs index fa64600..45d91da 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. @@ -316,5 +319,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..8c6e3e6 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"; @@ -118,6 +120,7 @@ struct Peer { incidents: Incidents, rtt_secs: Option<f64>, throughput: Option<f64>, + destination: Option<Destination>, } async fn peers(state: &AppState) -> Vec<Peer> { @@ -155,6 +158,7 @@ async fn peers(state: &AppState) -> Vec<Peer> { rtt_secs, throughput, alias: aliases.remove(&key), + destination: state.transport.destination(&key), key, }); } @@ -182,6 +186,147 @@ 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(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); @@ -503,6 +648,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 +668,33 @@ 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 = destination + .address + .split_once(':') + .map_or("", |(scheme, _)| scheme); + 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_neighbour", "gauge", "1 if the peer is a neighbour."); for peer in &peers { out.sample( diff --git a/src/simtest/identity.rs b/src/simtest/identity.rs index ed35389..ec5d665 100644 --- a/src/simtest/identity.rs +++ b/src/simtest/identity.rs @@ -14,7 +14,9 @@ use crate::swarm::bootstrap::Start; use crate::swarm::members; use crate::transport::identity::{WHO_LABEL, signed}; use crate::transport::sim::SimNet; -use crate::transport::{HeadsSummary, IdentityProof, Keepalive, ParsedMessage, Transport}; +use crate::transport::{ + Destination, HeadsSummary, IdentityProof, Keepalive, ParsedMessage, Transport, +}; const SOON: Duration = Duration::from_secs(20); @@ -41,6 +43,18 @@ async fn nodes_are_their_node_ids() { for node in [&a, &b, &c] { peers_are_node_ids(node).await; } + // Where a sends to b: where b proved itself, on the sim's transport. + assert_eq!( + a.transport.destination(b.local_id()), + Some(Destination { + address: "sim:b".into(), + proven: true, + connection: None, + }) + ); + let stats = a.transport.stats(); + assert_eq!(stats.identity.unwrap().proven["sim"], 2); + assert!(stats.libp2p.is_none()); let (b_id, c_id) = (b.local_id().to_owned(), c.local_id().to_owned()); let a = &a; sim.wait_until( @@ -118,6 +132,9 @@ async fn a_proof_holds_only_at_its_address() { sim.net.inject("ghost", "a", &ping(&a)).await.unwrap(); assert!(a.get_peer_state(&ghost.id()).await.is_none()); } + let counts = a.transport.stats().identity.unwrap().counts; + assert_eq!(counts.proofs_rejected, 2); + assert!(counts.dropped_unproven >= 2); sim.net .inject("ghost", "a", &who(&ghost, "sim:ghost", "sim:a")) diff --git a/src/transport/firewall.rs b/src/transport/firewall.rs index 78d00fc..846ae1b 100644 --- a/src/transport/firewall.rs +++ b/src/transport/firewall.rs @@ -96,6 +96,18 @@ impl Transport for FirewallTransport { self.inner.transports() } + fn destination(&self, peer: &str) -> Option<crate::transport::Destination> { + self.inner.destination(peer) + } + + fn connection(&self, target: &str) -> Option<crate::transport::Connection> { + self.inner.connection(target) + } + + fn stats(&self) -> crate::transport::stats::TransportStats { + self.inner.stats() + } + async fn resolve_seed(&self, seed: &str) -> Result<String> { self.inner.resolve_seed(seed).await } diff --git a/src/transport/identity.rs b/src/transport/identity.rs index af0fe74..17c7795 100644 --- a/src/transport/identity.rs +++ b/src/transport/identity.rs @@ -38,9 +38,11 @@ use tracing::{debug, info}; use crate::identity::{self, NodeKey}; use crate::locks::MutexExt; +use crate::transport::stats::{IdentityCounts, IdentityStats, TransportStats}; use crate::transport::traffic::Traffic; use crate::transport::{ - EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind, ParsedMessage, Transport, + Destination, EncodedMessage, HandlerFn, IdentityProof, MessageHandler, MessageKind, + ParsedMessage, Transport, }; pub(crate) const WHO_LABEL: &[u8] = b"continuum/who/1"; @@ -132,6 +134,8 @@ struct Book { asked: HashMap<String, Instant>, /// Seeds being resolved: who waits for a ME from where. waiting: HashMap<String, Vec<oneshot::Sender<String>>>, + /// Since start (`stats`). + counts: IdentityCounts, } /// The scheme of `address`. @@ -151,6 +155,7 @@ impl Book { if *slot != address { let previous = std::mem::replace(slot, address.to_owned()); self.at.remove(&previous); + self.counts.address_changes += 1; } } else { addresses.push(address.to_owned()); @@ -185,6 +190,25 @@ impl Book { } } + /// What the books hold now, and the counts. + fn stats(&self) -> IdentityStats { + let mut proven = std::collections::BTreeMap::new(); + for addresses in self.proven.values() { + for address in addresses { + *proven.entry(scheme_of(address).to_owned()).or_insert(0) += 1; + } + } + IdentityStats { + proven, + hinted: self + .hints + .keys() + .filter(|node| !self.proven.contains_key(*node)) + .count(), + counts: self.counts, + } + } + /// Whether to ask `address` WHO now (always, if `force`), noting that it was asked. fn ask(&mut self, address: &str, now: Instant, force: bool) -> bool { if !force @@ -294,6 +318,7 @@ impl Shared { ) -> Option<String> { let message = signed(label, address, &link.own_address()); if !identity::verify(&proof.key, &message, &proof.signature) { + self.lock().counts.proofs_rejected += 1; debug!(from = %address, "Ignoring an identity proof that doesn't hold"); return None; } @@ -379,6 +404,7 @@ impl Shared { match node { Some(node) => handler(node, message, payload_len).await, None => { + self.lock().counts.dropped_unproven += 1; debug!( from = %address, kind = ?message.kind(), @@ -512,6 +538,38 @@ impl Transport for IdentityTransport { .collect() } + /// The address `encode` picks, and the transport's connection there. + fn destination(&self, peer: &str) -> Option<Destination> { + let address = self.shared.destination(peer)?; + let proven = self + .shared + .lock() + .proven + .get(peer) + .is_some_and(|addresses| addresses.contains(&address)); + let connection = self + .shared + .split(&address) + .and_then(|(link, target)| link.inner.connection(&target)); + Some(Destination { + address, + proven, + connection, + }) + } + + /// The identity layer's, and each transport's. + fn stats(&self) -> TransportStats { + let mut stats = TransportStats { + identity: Some(self.shared.lock().stats()), + ..TransportStats::default() + }; + for link in &self.shared.links { + stats.merge(link.inner.stats()); + } + stats + } + /// Each transport's, with its scheme. fn discovered_seeds(&self) -> Vec<String> { self.shared @@ -594,9 +652,12 @@ impl Transport for IdentityTransport { info!(seed = %seed, node = %node, "Seed is node"); Ok(node) } - _ => Err(anyhow!( - "no node answered at {address} within {RESOLVE_TIMEOUT:?}" - )), + _ => { + self.shared.lock().counts.seeds_unanswered += 1; + Err(anyhow!( + "no node answered at {address} within {RESOLVE_TIMEOUT:?}" + )) + } } } @@ -617,7 +678,7 @@ impl Transport for IdentityTransport { async fn send_encoded(&self, message: &EncodedMessage) -> Result<()> { let hints = { - let book = self.shared.lock(); + let mut book = self.shared.lock(); let proven = book .proven .get(&message.peer) @@ -625,6 +686,7 @@ impl Transport for IdentityTransport { if proven { None } else { + book.counts.refused_unproven += 1; Some(book.hints.get(&message.peer).cloned().unwrap_or_default()) } }; @@ -721,6 +783,11 @@ mod tests { assert!(book.bind("s:y", "n2", now)); assert_eq!(book.proven["n1"], ["t:x"]); assert_eq!(book.at["s:y"].0, "n2"); + // Only n1's move from s:x to s:y changed an address. + assert_eq!(book.counts.address_changes, 1); + let stats = book.stats(); + assert_eq!(stats.proven["s"], 1); + assert_eq!(stats.proven["t"], 1); } #[test] diff --git a/src/transport/mod.rs b/src/transport/mod.rs index 3f77306..10a8ae5 100644 --- a/src/transport/mod.rs +++ b/src/transport/mod.rs @@ -18,6 +18,7 @@ pub(crate) mod mycelium; pub(crate) mod p2p; #[cfg(test)] pub(crate) mod sim; +pub(crate) mod stats; pub(crate) mod traffic; pub(crate) mod wire; @@ -330,6 +331,37 @@ pub(crate) struct ListedPeer { pub(crate) addresses: Vec<String>, } +/// Where messages to a peer go now (`Transport::destination`). +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Destination { + /// As the identity layer keeps it: `<scheme>:<transport identity>`. + pub(crate) address: String, + /// The peer proved itself there; else it is only said to be there, and sends wait + /// for its proof. + pub(crate) proven: bool, + /// The transport's connection to it, if the transport has connections. + pub(crate) connection: Option<Connection>, +} + +/// How a transport with connections (libp2p) is connected to a peer. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) enum Connection { + Direct, + /// Only through a relay. + Relayed, +} + +impl Connection { + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Direct => "direct", + Self::Relayed => "relayed", + } + } +} + #[derive(Clone, Serialize, Deserialize)] pub(crate) struct HelloReply { pub(crate) swarm_id: String, @@ -584,6 +616,25 @@ pub(crate) trait Transport: Send + Sync { let _ = (peer, addresses); } + /// Where messages to `peer` go now. None if nowhere, and by default, for a + /// transport whose identities are addresses (`peer` is where they go). + fn destination(&self, peer: &str) -> Option<Destination> { + let _ = peer; + None + } + + /// How this node is connected to `target` (a transport identity), if it is. By + /// default not at all, for a transport without connections (Mycelium). + fn connection(&self, target: &str) -> Option<Connection> { + let _ = target; + None + } + + /// Figures about the transport itself (`stats`). By default none. + fn stats(&self) -> stats::TransportStats { + stats::TransportStats::default() + } + /// Resolves a seed (anything that leads to a node, e.g. an address) to that node's /// identity, as `local_id` gives it on that node. async fn resolve_seed(&self, seed: &str) -> Result<String>; diff --git a/src/transport/p2p.rs b/src/transport/p2p.rs index ebda922..c03a0c4 100644 --- a/src/transport/p2p.rs +++ b/src/transport/p2p.rs @@ -58,7 +58,12 @@ use tokio::sync::{Semaphore, mpsc, oneshot, watch}; use tracing::{debug, info, warn}; use crate::locks::MutexExt; -use crate::transport::{EncodedMessage, MessageHandler, ParsedMessage, Transport, wire}; +use crate::transport::stats::{ + Connections, DialFailures, HolePunches, Libp2pStats, Reachability, RelaySlots, TransportStats, +}; +use crate::transport::{ + Connection, EncodedMessage, MessageHandler, ParsedMessage, Transport, wire, +}; const PROTOCOL: StreamProtocol = StreamProtocol::new("/continuum/1"); /// The largest message read: anyone who can connect can send anything, so this bounds @@ -193,6 +198,15 @@ struct Book { public_relays: HashMap<PeerId, Multiaddr>, /// Relays picked that never granted a slot, and when (`RELAY_PAUSE`). failed_relays: HashMap<PeerId, tokio::time::Instant>, + /// Relays that granted us a slot we hold. + slots: std::collections::HashSet<PeerId>, + /// Open connections: to whom, and whether through a relay. + connections: HashMap<ConnectionId, (PeerId, bool)>, + /// What AutoNAT says, and the address it confirmed while public. + reachability: Reachability, + public_address: Option<Multiaddr>, + /// Since start (`stats`). + counts: Counts, /// When each peer was last mentioned (`STALE_AFTER`). seen: HashMap<PeerId, tokio::time::Instant>, /// Peers found on the LAN (mDNS) not known yet to speak continuum. @@ -204,7 +218,72 @@ struct Book { discovered: Vec<Multiaddr>, } +/// What `Libp2pStats` counts since start. +#[derive(Default)] +struct Counts { + slots_granted: u64, + slots_lost: u64, + slots_not_granted: u64, + hole_punches: HolePunches, + dial_failures: DialFailures, +} + +/// Counts a dial that failed with `error` (not one that wasn't needed). +fn count_dial_failure(failures: &mut DialFailures, error: &::libp2p::swarm::DialError) { + use ::libp2p::swarm::DialError; + match error { + DialError::DialPeerConditionFalse(_) => {} + DialError::Transport(_) => failures.unreachable += 1, + DialError::WrongPeerId { .. } => failures.wrong_peer_id += 1, + DialError::NoAddresses => failures.no_addresses += 1, + _ => failures.other += 1, + } +} + impl Book { + /// How we are connected to `peer`: directly if any connection is direct. + fn connection(&self, peer: &PeerId) -> Option<Connection> { + let mut found = None; + for (connected, relayed) in self.connections.values() { + if connected == peer { + if !relayed { + return Some(Connection::Direct); + } + found = Some(Connection::Relayed); + } + } + found + } + + fn stats(&self) -> Libp2pStats { + let mut connections = Connections::default(); + for (_, relayed) in self.connections.values() { + if *relayed { + connections.relayed += 1; + } else { + connections.direct += 1; + } + } + Libp2pStats { + reachability: self.reachability, + public_address: self.public_address.as_ref().map(Multiaddr::to_string), + connections, + relay_slots: RelaySlots { + held: self.slots.len(), + paused: self + .failed_relays + .values() + .filter(|at| at.elapsed() < RELAY_PAUSE) + .count(), + granted: self.counts.slots_granted, + lost: self.counts.slots_lost, + not_granted: self.counts.slots_not_granted, + }, + hole_punches: self.counts.hole_punches, + dial_failures: self.counts.dial_failures, + } + } + /// `peer` was mentioned just now. fn touch(&mut self, peer: PeerId) { self.seen.insert(peer, tokio::time::Instant::now()); @@ -675,8 +754,10 @@ fn pick_relays( ) { if nat.is_public() && !picked.is_empty() { info!("Reachable directly (AutoNAT); letting the relays picked go"); - for listener in std::mem::take(picked).into_keys() { + let mut book = lock(book); + for (listener, relay) in std::mem::take(picked) { swarm.remove_listener(listener); + book.slots.remove(&relay); } return; } @@ -735,15 +816,14 @@ async fn run( ) { let serving = relaying.serving; let mut nat = ::libp2p::autonat::NatStatus::Unknown; - // Relays picked among the members, by listener, and those that granted a slot. + // Relays picked among the members, by listener. let mut picked: HashMap<::libp2p::core::transport::ListenerId, PeerId> = HashMap::new(); - let mut granted: std::collections::HashSet<PeerId> = std::collections::HashSet::new(); let mut check = tokio::time::interval(RELAY_CHECK); check.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); let mut dht_bootstrap = tokio::time::interval(DHT_BOOTSTRAP); dht_bootstrap.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); - // Relays we listen through, by listener, as configured. - let mut reservations: HashMap<::libp2p::core::transport::ListenerId, Multiaddr> = + // Relays we listen through, by listener: as configured, and their peer. + let mut reservations: HashMap<::libp2p::core::transport::ListenerId, (Multiaddr, PeerId)> = HashMap::new(); let reserve_later = |configured: Multiaddr| { let again = again.clone(); @@ -805,6 +885,7 @@ async fn run( opts = opts.extend_addresses_through_behaviour(); } if let Err(error) = swarm.dial(opts.build()) { + count_dial_failure(&mut lock(&book).counts.dial_failures, &error); for done in connecting.remove(&peer).into_iter().flatten() { let _ = done.send(Err(error.to_string())); } @@ -837,10 +918,13 @@ async fn run( }); } Some(Command::Reserve { relay, configured }) => { + let Some(relay_peer) = split_peer(&relay) else { + continue; // The arm above takes those. + }; match swarm.listen_on(relay.clone().with(Protocol::P2pCircuit)) { Ok(listener) => { debug!(relay = %relay, "Reserving a slot on a libp2p relay"); - reservations.insert(listener, configured); + reservations.insert(listener, (configured, relay_peer)); } Err(error) => { warn!(relay = %relay, error = %error, "Couldn't use a libp2p relay; trying again later"); @@ -862,8 +946,13 @@ async fn run( } }, event = swarm.select_next_some() => match event { - SwarmEvent::ConnectionEstablished { peer_id, connection_id, .. } => { - lock(&book).touch(peer_id); + SwarmEvent::ConnectionEstablished { peer_id, connection_id, endpoint, .. } => { + { + let mut book = lock(&book); + book.touch(peer_id); + let relayed = endpoint.is_relayed(); + book.connections.insert(connection_id, (peer_id, relayed)); + } for done in connecting.remove(&peer_id).into_iter().flatten() { let _ = done.send(Ok(())); } @@ -872,7 +961,11 @@ async fn run( let _ = done.send(Ok(peer_id)); } } + SwarmEvent::ConnectionClosed { connection_id, .. } => { + lock(&book).connections.remove(&connection_id); + } SwarmEvent::OutgoingConnectionError { connection_id, peer_id, error } => { + count_dial_failure(&mut lock(&book).counts.dial_failures, &error); if let Some((_, done)) = identifying.remove(&connection_id) { let _ = done.send(Err(error.to_string())); } @@ -909,14 +1002,33 @@ async fn run( lock(&book) .listening .retain(|known| !addresses.contains(known)); - if let Some(relay) = reservations.remove(&listener_id) { + if let Some((relay, relay_peer)) = reservations.remove(&listener_id) { + { + let mut book = lock(&book); + if book.slots.remove(&relay_peer) { + book.counts.slots_lost += 1; + } else { + book.counts.slots_not_granted += 1; + } + } warn!(relay = %relay, reason = ?reason, "Lost the slot on a libp2p relay; reserving again later"); reserve_later(relay); } // A relay picked among the members: the next check picks again, another // one if this one never granted the slot. if let Some(relay) = picked.remove(&listener_id) { - if granted.remove(&relay) { + let lost = { + let mut book = lock(&book); + let lost = book.slots.remove(&relay); + if lost { + book.counts.slots_lost += 1; + } else { + book.counts.slots_not_granted += 1; + book.failed_relays.insert(relay, tokio::time::Instant::now()); + } + lost + }; + if lost { debug!(relay = %relay, reason = ?reason, "Lost the slot on a member relaying"); } else { info!( @@ -924,9 +1036,6 @@ async fn run( pause = ?RELAY_PAUSE, "A relay didn't grant a slot (unreachable, or another peer at its address); picking others for a while" ); - lock(&book) - .failed_relays - .insert(relay, tokio::time::Instant::now()); } } } @@ -936,8 +1045,14 @@ async fn run( SwarmEvent::Behaviour(BehaviourEvent::RelayClient( ::libp2p::relay::client::Event::ReservationReqAccepted { relay_peer_id, renewal, .. }, )) => { - granted.insert(relay_peer_id); - lock(&book).failed_relays.remove(&relay_peer_id); + { + let mut book = lock(&book); + book.slots.insert(relay_peer_id); + book.failed_relays.remove(&relay_peer_id); + if !renewal { + book.counts.slots_granted += 1; + } + } if !renewal { info!(relay = %relay_peer_id, "Reachable through a libp2p relay"); } @@ -946,14 +1061,31 @@ async fn run( ::libp2p::autonat::Event::StatusChanged { old, new }, )) => { info!(old = ?old, new = ?new, "libp2p reachability (AutoNAT)"); + { + use ::libp2p::autonat::NatStatus; + let mut book = lock(&book); + (book.reachability, book.public_address) = match &new { + NatStatus::Unknown => (Reachability::Unknown, None), + NatStatus::Private => (Reachability::Private, None), + NatStatus::Public(address) => { + (Reachability::Public, Some(address.clone())) + } + }; + } nat = new; } SwarmEvent::Behaviour(BehaviourEvent::Relay(event)) => { debug!(event = ?event, "libp2p relay"); } SwarmEvent::Behaviour(BehaviourEvent::Dcutr(event)) => match event.result { - Ok(_) => info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay"), - Err(error) => debug!(peer = %event.remote_peer_id, error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on"), + Ok(_) => { + lock(&book).counts.hole_punches.succeeded += 1; + info!(peer = %event.remote_peer_id, "Connected directly to a peer behind a relay"); + } + Err(error) => { + lock(&book).counts.hole_punches.failed += 1; + debug!(peer = %event.remote_peer_id, error = %error, "Couldn't connect directly to a peer behind a relay; the relay carries on"); + } }, SwarmEvent::Behaviour(BehaviourEvent::Identify( ::libp2p::identify::Event::Received { peer_id, info, .. }, @@ -1124,6 +1256,18 @@ impl Transport for Libp2pTransport { Some(peer.to_string()) } + fn connection(&self, target: &str) -> Option<Connection> { + let peer = PeerId::from_str(target).ok()?; + lock(&self.book).connection(&peer) + } + + fn stats(&self) -> TransportStats { + TransportStats { + libp2p: Some(lock(&self.book).stats()), + ..TransportStats::default() + } + } + /// Nodes found on the LAN (mDNS) that speak continuum. fn discovered_seeds(&self) -> Vec<String> { std::mem::take(&mut lock(&self.book).discovered) @@ -1352,6 +1496,12 @@ mod tests { a.send_message(b.local_id(), &ping()).await.unwrap(); let (from, _) = at_b.recv().await.unwrap(); assert_eq!(from, a.local_id()); + + assert_eq!(a.connection(b.local_id()), Some(Connection::Direct)); + assert_eq!(b.connection(a.local_id()), Some(Connection::Direct)); + let stats = a.stats().libp2p.unwrap(); + assert!(stats.connections.direct >= 1); + assert_eq!(stats.connections.relayed, 0); } #[tokio::test] @@ -1474,11 +1624,16 @@ mod tests { "{relayed}" ); + assert!(eventually(|| b.stats().libp2p.unwrap().relay_slots.held == 1).await); + let slots = b.stats().libp2p.unwrap().relay_slots; + assert_eq!((slots.granted, slots.not_granted), (1, 0)); + let (a, _) = memory_node(); assert_eq!(a.resolve_seed(&relayed).await.unwrap(), b.local_id()); a.send_message(b.local_id(), &ping()).await.unwrap(); let (from, _) = at_b.recv().await.unwrap(); assert_eq!(from, a.local_id()); + assert!(a.connection(b.local_id()).is_some()); } fn peer() -> PeerId { @@ -1603,5 +1758,26 @@ mod tests { .to_string(); assert!(b.send_message(&stranger, &ping()).await.is_err()); assert!(b.resolve_seed(&stranger).await.is_err()); + assert_eq!(b.connection(&stranger), None); + } + + #[test] + fn a_direct_connection_wins_over_a_relayed_one() { + let mut book = Book::default(); + let (relayed_only, both) = (peer(), peer()); + let mut id = 0; + let mut connect = |book: &mut Book, peer: PeerId, relayed: bool| { + id += 1; + book.connections + .insert(ConnectionId::new_unchecked(id), (peer, relayed)); + }; + connect(&mut book, relayed_only, true); + connect(&mut book, both, true); + connect(&mut book, both, false); + assert_eq!(book.connection(&relayed_only), Some(Connection::Relayed)); + assert_eq!(book.connection(&both), Some(Connection::Direct)); + assert_eq!(book.connection(&peer()), None); + let connections = book.stats().connections; + assert_eq!((connections.direct, connections.relayed), (1, 2)); } } diff --git a/src/transport/stats.rs b/src/transport/stats.rs new file mode 100644 index 0000000..35238f5 --- /dev/null +++ b/src/transport/stats.rs @@ -0,0 +1,141 @@ +//! Figures about the transports themselves (`Transport::stats`), for /admin/stats +//! (`transports`, `docs/measurements.md`): how they find and reach peers, besides the +//! messages counted per peer (`traffic`). + +use std::collections::BTreeMap; + +use serde::Serialize; + +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct TransportStats { + /// The identity layer's (`transport::identity`); absent without one. + pub(crate) identity: Option<IdentityStats>, + /// The libp2p transport's; absent if it isn't one of this node's. + pub(crate) libp2p: Option<Libp2pStats>, +} + +impl TransportStats { + /// Takes what `other` has. + pub(crate) fn merge(&mut self, other: TransportStats) { + if other.identity.is_some() { + self.identity = other.identity; + } + if other.libp2p.is_some() { + self.libp2p = other.libp2p; + } + } +} + +/// Who is where (`transport::identity`): the books now, and counts since start. +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct IdentityStats { + /// Nodes proven at an address, by transport (scheme); a node on several counts on + /// each. + pub(crate) proven: BTreeMap<String, usize>, + /// Nodes known only from hints (the member list, HELLO_REPLYs), proven nowhere. + pub(crate) hinted: usize, + #[serde(flatten)] + pub(crate) counts: IdentityCounts, +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct IdentityCounts { + /// Proofs that didn't hold (a bad signature, or made for other addresses): normally + /// none. + pub(crate) proofs_rejected: u64, + /// Nodes that proved themselves at another address on a transport than before: a + /// node restarted with a new transport identity (a new libp2p key or Mycelium key). + pub(crate) address_changes: u64, + /// Messages dropped because no node proved itself at their address: some after a + /// restart are normal, a steady stream is not. + pub(crate) dropped_unproven: u64, + /// Sends refused because the peer hadn't proven where it is yet (`NotProven`). + pub(crate) refused_unproven: u64, + /// Seeds no node answered at in time. + pub(crate) seeds_unanswered: u64, +} + +/// Whether others can dial this node, as AutoNAT found out. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) enum Reachability { + #[default] + Unknown, + /// Others can't dial it: it needs a relay. + Private, + Public, +} + +impl Reachability { + pub(crate) const ALL: [Reachability; 3] = [Self::Unknown, Self::Private, Self::Public]; + + pub(crate) fn as_str(self) -> &'static str { + match self { + Self::Unknown => "unknown", + Self::Private => "private", + Self::Public => "public", + } + } +} + +#[derive(Debug, Clone, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Libp2pStats { + pub(crate) reachability: Reachability, + /// The address AutoNAT confirmed others can dial, while public. + pub(crate) public_address: Option<String>, + /// Open connections, now. + pub(crate) connections: Connections, + pub(crate) relay_slots: RelaySlots, + /// Direct connections tried over a relayed one (DCUtR), since start. + pub(crate) hole_punches: HolePunches, + /// Connections that couldn't be made, by cause, since start. + pub(crate) dial_failures: DialFailures, +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct Connections { + pub(crate) direct: usize, + /// Through a relay (a circuit). + pub(crate) relayed: usize, +} + +/// Slots this node holds on relays, to be reachable through them. +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct RelaySlots { + /// Held now. + pub(crate) held: usize, + /// Relays picked that never granted a slot, not picked again for a while now. + pub(crate) paused: usize, + /// Since start: slots granted (not counting renewals), lost after being granted, and + /// asked for but never granted (a relay unreachable, or another peer at its + /// address: a relay restarted without its libp2p key). + pub(crate) granted: u64, + pub(crate) lost: u64, + pub(crate) not_granted: u64, +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct HolePunches { + pub(crate) succeeded: u64, + /// The relay carries on. + pub(crate) failed: u64, +} + +#[derive(Debug, Clone, Copy, Default, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct DialFailures { + /// No address of the peer could be reached (refused, timed out, no route). + pub(crate) unreachable: u64, + /// Another peer answered at the address: one restarted without its libp2p key. + pub(crate) wrong_peer_id: u64, + /// No address known for the peer. + pub(crate) no_addresses: u64, + pub(crate) other: u64, +} diff --git a/src/transport/traffic.rs b/src/transport/traffic.rs index b332d3d..e912302 100644 --- a/src/transport/traffic.rs +++ b/src/transport/traffic.rs @@ -173,6 +173,18 @@ impl Transport for CountingTransport { self.inner.transports() } + fn destination(&self, peer: &str) -> Option<crate::transport::Destination> { + self.inner.destination(peer) + } + + fn connection(&self, target: &str) -> Option<crate::transport::Connection> { + self.inner.connection(target) + } + + fn stats(&self) -> crate::transport::stats::TransportStats { + self.inner.stats() + } + async fn resolve_seed(&self, seed: &str) -> Result<String> { self.inner.resolve_seed(seed).await } |