feat(swarm): expose ConnectionId and add conn duration metric

misc/metrics/CHANGELOG.md

@@ -22,7 +22,12 @@
 - Raise MSRV to 1.65.
   See [PR 3715].
 
+- Replace `libp2p_swarm_connections_closed` `Counter` with `libp2p_swarm_connections_duration` `Histogram` which additionally tracks the duration of a connection.
+  Note that you can use the `_count` metric of the `Histogram` as a replacement for the `Counter`.
+  See [PR 3927].
+
 [PR 3715]: https://github.com/libp2p/rust-libp2p/pull/3715
+[PR 3927]: https://github.com/libp2p/rust-libp2p/pull/3927
 [PR 3325]: https://github.com/libp2p/rust-libp2p/pull/3325
 
 ## 0.12.0

misc/metrics/Cargo.toml

@@ -19,6 +19,7 @@ relay = ["libp2p-relay"]
 dcutr = ["libp2p-dcutr"]
 
 [dependencies]
+instant = "0.1.11"
 libp2p-core = { workspace = true }
 libp2p-dcutr =  { workspace = true, optional = true }
 libp2p-identify = { workspace = true, optional = true }
@@ -27,7 +28,7 @@ libp2p-ping = { workspace = true, optional = true }
 libp2p-relay =  { workspace = true, optional = true }
 libp2p-swarm = { workspace = true }
 libp2p-identity = { workspace = true }
-prometheus-client = { version = "0.21.0" }
+prometheus-client = { version = "0.21.1"}
 once_cell = "1.16.0"
 
 [target.'cfg(not(target_os = "unknown"))'.dependencies]

misc/metrics/src/swarm.rs

@@ -18,20 +18,25 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
 use crate::protocol_stack;
+use instant::Instant;
+use libp2p_swarm::ConnectionId;
 use prometheus_client::encoding::{EncodeLabelSet, EncodeLabelValue};
 use prometheus_client::metrics::counter::Counter;
 use prometheus_client::metrics::family::Family;
 use prometheus_client::metrics::histogram::{exponential_buckets, Histogram};
-use prometheus_client::registry::Registry;
+use prometheus_client::registry::{Registry, Unit};
 
 pub(crate) struct Metrics {
     connections_incoming: Family<AddressLabels, Counter>,
     connections_incoming_error: Family<IncomingConnectionErrorLabels, Counter>,
 
-    connections_established: Family<ConnectionEstablishedLabels, Counter>,
-    connections_establishment_duration: Family<ConnectionEstablishmentDurationLabels, Histogram>,
-    connections_closed: Family<ConnectionClosedLabels, Counter>,
+    connections_established: Family<ConnectionLabels, Counter>,
+    connections_establishment_duration: Family<ConnectionLabels, Histogram>,
+    connections_duration: Family<ConnectionClosedLabels, Histogram>,
 
     new_listen_addr: Family<AddressLabels, Counter>,
     expired_listen_addr: Family<AddressLabels, Counter>,
@@ -41,6 +46,8 @@ pub(crate) struct Metrics {
 
     dial_attempt: Counter,
     outgoing_connection_error: Family<OutgoingConnectionErrorLabels, Counter>,
+
+    connections: Arc<Mutex<HashMap<ConnectionId, Instant>>>,
 }
 
 impl Metrics {
@@ -110,34 +117,42 @@ impl Metrics {
             connections_established.clone(),
         );
 
-        let connections_closed = Family::default();
+        let connections_establishment_duration = {
+            let constructor: fn() -> Histogram =
+                || Histogram::new(exponential_buckets(0.01, 1.5, 20));
+            Family::new_with_constructor(constructor)
+        };
         sub_registry.register(
-            "connections_closed",
-            "Number of connections closed",
-            connections_closed.clone(),
+            "connections_establishment_duration",
+            "Time it took (locally) to establish connections",
+            connections_establishment_duration.clone(),
         );
 
-        let connections_establishment_duration = Family::new_with_constructor(
-            create_connection_establishment_duration_histogram as fn() -> Histogram,
-        );
-        sub_registry.register(
+        let connections_duration = {
+            let constructor: fn() -> Histogram =
+                || Histogram::new(exponential_buckets(0.01, 3.0, 20));
+            Family::new_with_constructor(constructor)
+        };
+        sub_registry.register_with_unit(
             "connections_establishment_duration",
             "Time it took (locally) to establish connections",
+            Unit::Seconds,
             connections_establishment_duration.clone(),
         );
 
         Self {
             connections_incoming,
             connections_incoming_error,
             connections_established,
-            connections_closed,
             new_listen_addr,
             expired_listen_addr,
             listener_closed,
             listener_error,
             dial_attempt,
             outgoing_connection_error,
             connections_establishment_duration,
+            connections_duration,
+            connections: Default::default(),
         }
     }
 }
@@ -149,24 +164,44 @@ impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleE
             libp2p_swarm::SwarmEvent::ConnectionEstablished {
                 endpoint,
                 established_in: time_taken,
+                connection_id,
                 ..
             } => {
-                let labels = ConnectionEstablishedLabels {
+                let labels = ConnectionLabels {
                     role: endpoint.into(),
                     protocols: protocol_stack::as_string(endpoint.get_remote_address()),
                 };
                 self.connections_established.get_or_create(&labels).inc();
                 self.connections_establishment_duration
                     .get_or_create(&labels)
                     .observe(time_taken.as_secs_f64());
+                self.connections
+                    .lock()
+                    .expect("lock not to be poisoned")
+                    .insert(*connection_id, Instant::now());
             }
-            libp2p_swarm::SwarmEvent::ConnectionClosed { endpoint, .. } => {
-                self.connections_closed
-                    .get_or_create(&ConnectionClosedLabels {
+            libp2p_swarm::SwarmEvent::ConnectionClosed {
+                endpoint,
+                connection_id,
+                cause,
+                ..
+            } => {
+                let labels = ConnectionClosedLabels {
+                    connection: ConnectionLabels {
                         role: endpoint.into(),
                         protocols: protocol_stack::as_string(endpoint.get_remote_address()),
-                    })
-                    .inc();
+                    },
+                    cause: cause.as_ref().expect("TODO remove see definition").into(),
+                };
+                self.connections_duration.get_or_create(&labels).observe(
+                    self.connections
+                        .lock()
+                        .expect("lock not to be poisoned")
+                        .remove(connection_id)
+                        .expect("closed connection to previously be established")
+                        .elapsed()
+                        .as_secs_f64(),
+                );
             }
             libp2p_swarm::SwarmEvent::IncomingConnection { send_back_addr, .. } => {
                 self.connections_incoming
@@ -187,7 +222,7 @@ impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleE
                     })
                     .inc();
             }
-            libp2p_swarm::SwarmEvent::OutgoingConnectionError { error, peer_id } => {
+            libp2p_swarm::SwarmEvent::OutgoingConnectionError { error, peer_id, .. } => {
                 let peer = match peer_id {
                     Some(_) => PeerStatus::Known,
                     None => PeerStatus::Unknown,
@@ -261,25 +296,42 @@ impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleE
             libp2p_swarm::SwarmEvent::ListenerError { .. } => {
                 self.listener_error.inc();
             }
-            libp2p_swarm::SwarmEvent::Dialing(_) => {
+            libp2p_swarm::SwarmEvent::Dialing { .. } => {
                 self.dial_attempt.inc();
             }
         }
     }
 }
 
 #[derive(EncodeLabelSet, Hash, Clone, Eq, PartialEq, Debug)]
-struct ConnectionEstablishedLabels {
+struct ConnectionLabels {
     role: Role,
     protocols: String,
 }
 
-type ConnectionEstablishmentDurationLabels = ConnectionEstablishedLabels;
-
 #[derive(EncodeLabelSet, Hash, Clone, Eq, PartialEq, Debug)]
 struct ConnectionClosedLabels {
-    role: Role,
-    protocols: String,
+    // TODO: Should be Option<ConnectionError>. Needs https://github.com/prometheus/client_rust/pull/137
+    cause: ConnectionError,
+    #[prometheus(flatten)]
+    connection: ConnectionLabels,
+}
+
+#[derive(EncodeLabelValue, Hash, Clone, Eq, PartialEq, Debug)]
+enum ConnectionError {
+    Io,
+    KeepAliveTimeout,
+    Handler,
+}
+
+impl<E> From<&libp2p_swarm::ConnectionError<E>> for ConnectionError {
+    fn from(value: &libp2p_swarm::ConnectionError<E>) -> Self {
+        match value {
+            libp2p_swarm::ConnectionError::IO(_) => ConnectionError::Io,
+            libp2p_swarm::ConnectionError::KeepAliveTimeout => ConnectionError::KeepAliveTimeout,
+            libp2p_swarm::ConnectionError::Handler(_) => ConnectionError::Handler,
+        }
+    }
 }
 
 #[derive(EncodeLabelSet, Hash, Clone, Eq, PartialEq, Debug)]
@@ -359,7 +411,3 @@ impl From<&libp2p_swarm::ListenError> for IncomingConnectionError {
         }
     }
 }
-
-fn create_connection_establishment_duration_histogram() -> Histogram {
-    Histogram::new(exponential_buckets(0.01, 1.5, 20))
-}

swarm/CHANGELOG.md

@@ -50,6 +50,10 @@
 - Remove deprecated `NetworkBehaviourAction` type.
   See [PR 3919].
 
+- Expose `ConnectionId` on `SwarmEvent::{ConnectionEstablished,ConnectionClosed,IncomingConnection,IncomingConnectionError,OutgoingConnectionError,Dialing}`.
+  Also emit `SwarmEvent::Dialing` for dials with unknown `PeerId`.
+  See [PR 3927].
+
 [PR 3605]: https://github.com/libp2p/rust-libp2p/pull/3605
 [PR 3651]: https://github.com/libp2p/rust-libp2p/pull/3651
 [PR 3715]: https://github.com/libp2p/rust-libp2p/pull/3715
@@ -62,6 +66,7 @@
 [PR 3886]: https://github.com/libp2p/rust-libp2p/pull/3886
 [PR 3912]: https://github.com/libp2p/rust-libp2p/pull/3912
 [PR 3919]: https://github.com/libp2p/rust-libp2p/pull/3919
+[PR 3927]: https://github.com/libp2p/rust-libp2p/pull/3927
 
 ## 0.42.2