time to establish connection

misc/metrics/src/swarm.rs

@@ -20,15 +20,19 @@
 
 use crate::protocol_stack;
 use prometheus_client::encoding::text::Encode;
-use prometheus_client::metrics::counter::Counter;
-use prometheus_client::metrics::family::Family;
+use prometheus_client::metrics::{
+    counter::Counter,
+    family::Family,
+    histogram::{exponential_buckets, Histogram},
+};
 use prometheus_client::registry::Registry;
 
 pub struct Metrics {
     connections_incoming: Family<AddressLabels, Counter>,
     connections_incoming_error: Family<IncomingConnectionErrorLabels, Counter>,
 
     connections_established: Family<ConnectionEstablishedLabels, Counter>,
+    connection_establishment_duration: Family<ConnectionEstablishedLabels, Histogram>,
     connections_closed: Family<ConnectionClosedLabels, Counter>,
 
     new_listen_addr: Family<AddressLabels, Counter>,
@@ -123,6 +127,15 @@ impl Metrics {
             Box::new(connections_closed.clone()),
         );
 
+        let connection_establishment_duration = Family::new_with_constructor(
+            create_connection_establishment_duration_histogram as fn() -> Histogram,
+        );
+        sub_registry.register(
+            "connection_establishment_duration",
+            "Time it took (locally) to establish connections",
+            Box::new(connection_establishment_duration.clone()),
+        );
+
         Self {
             connections_incoming,
             connections_incoming_error,
@@ -135,6 +148,7 @@ impl Metrics {
             dial_attempt,
             outgoing_connection_error,
             connected_to_banned_peer,
+            connection_establishment_duration,
         }
     }
 }
@@ -143,13 +157,19 @@ impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleE
     fn record(&self, event: &libp2p_swarm::SwarmEvent<TBvEv, THandleErr>) {
         match event {
             libp2p_swarm::SwarmEvent::Behaviour(_) => {}
-            libp2p_swarm::SwarmEvent::ConnectionEstablished { endpoint, .. } => {
-                self.connections_established
-                    .get_or_create(&ConnectionEstablishedLabels {
-                        role: endpoint.into(),
-                        protocols: protocol_stack::as_string(endpoint.get_remote_address()),
-                    })
-                    .inc();
+            libp2p_swarm::SwarmEvent::ConnectionEstablished {
+                endpoint,
+                time_taken,
+                ..
+            } => {
+                let labels = ConnectionEstablishedLabels {
+                    role: endpoint.into(),
+                    protocols: protocol_stack::as_string(endpoint.get_remote_address()),
+                };
+                self.connections_established.get_or_create(&labels).inc();
+                self.connection_establishment_duration
+                    .get_or_create(&labels)
+                    .observe(time_taken.as_secs_f64());
             }
             libp2p_swarm::SwarmEvent::ConnectionClosed { endpoint, .. } => {
                 self.connections_closed
@@ -372,3 +392,7 @@ impl<TTransErr> From<&libp2p_swarm::PendingInboundConnectionError<TTransErr>>
         }
     }
 }
+
+fn create_connection_establishment_duration_histogram() -> Histogram {
+    Histogram::new(exponential_buckets(1e-3, 2., 10))
+}

protocols/autonat/tests/test_server.rs

@@ -195,6 +195,7 @@ async fn test_dial_back() {
                         },
                     num_established,
                     concurrent_dial_errors,
+                    time_taken: _,
                 } => {
                     assert_eq!(peer_id, client_id);
                     assert_eq!(num_established, NonZeroU32::new(2).unwrap());

swarm/src/connection/pool.rs

@@ -38,6 +38,7 @@ use futures::{
     ready,
     stream::FuturesUnordered,
 };
+use instant::Instant;
 use libp2p_core::connection::{ConnectionId, Endpoint, PendingPoint};
 use libp2p_core::muxing::{StreamMuxerBox, StreamMuxerExt};
 use std::{
@@ -195,6 +196,8 @@ struct PendingConnection<THandler> {
     endpoint: PendingPoint,
     /// When dropped, notifies the task which then knows to terminate.
     abort_notifier: Option<oneshot::Sender<Void>>,
+    /// The moment we became aware of this possible connection, useful for timing metrics.
+    accepted_at: Instant,
 }
 
 impl<THandler> PendingConnection<THandler> {
@@ -237,6 +240,8 @@ where
         /// Addresses are dialed in parallel. Contains the addresses and errors
         /// of dial attempts that failed before the one successful dial.
         concurrent_dial_errors: Option<Vec<(Multiaddr, TransportError<TTrans::Error>)>>,
+        /// How long it took to establish this connection.
+        established_in: std::time::Duration,
     },
 
     /// An established connection was closed.
@@ -493,6 +498,7 @@ where
                 handler,
                 endpoint,
                 abort_notifier: Some(abort_notifier),
+                accepted_at: Instant::now(),
             },
         );
         Ok(connection_id)
@@ -540,6 +546,7 @@ where
                 handler,
                 endpoint: endpoint.into(),
                 abort_notifier: Some(abort_notifier),
+                accepted_at: Instant::now(),
             },
         );
         Ok(connection_id)
@@ -634,6 +641,7 @@ where
                         handler,
                         endpoint,
                         abort_notifier: _,
+                        accepted_at,
                     } = self
                         .pending
                         .remove(&id)
@@ -783,13 +791,14 @@ where
                         )
                         .boxed(),
                     );
-
+                    let time_taken = Instant::now() - accepted_at;
                     return Poll::Ready(PoolEvent::ConnectionEstablished {
                         peer_id: obtained_peer_id,
                         endpoint,
                         id,
                         other_established_connection_ids,
                         concurrent_dial_errors,
+                        established_in: time_taken,
                     });
                 }
                 task::PendingConnectionEvent::PendingFailed { id, error } => {
@@ -798,6 +807,7 @@ where
                         handler,
                         endpoint,
                         abort_notifier: _,
+                        accepted_at: _, // Ignoring the time it took for the connection to fail.
                     }) = self.pending.remove(&id)
                     {
                         self.counters.dec_pending(&endpoint);

swarm/src/lib.rs

@@ -191,6 +191,8 @@ pub enum SwarmEvent<TBehaviourOutEvent, THandlerErr> {
         /// Addresses are dialed concurrently. Contains the addresses and errors
         /// of dial attempts that failed before the one successful dial.
         concurrent_dial_errors: Option<Vec<(Multiaddr, TransportError<io::Error>)>>,
+        /// How long it took to establish this connection
+        time_taken: std::time::Duration,
     },
     /// A connection with the given peer has been closed,
     /// possibly as a result of an error.
@@ -821,6 +823,7 @@ where
                 endpoint,
                 other_established_connection_ids,
                 concurrent_dial_errors,
+                established_in: time_taken,
             } => {
                 if self.banned_peers.contains(&peer_id) {
                     // Mark the connection for the banned peer as banned, thus withholding any
@@ -861,6 +864,7 @@ where
                         num_established,
                         endpoint,
                         concurrent_dial_errors,
+                        time_taken,
                     });
                 }
             }