feat(libp2p): track bandwidth per transport protocol stack

Cargo.toml

@@ -85,7 +85,7 @@ libp2p-identity = { version = "0.2.7" }
 libp2p-kad = { version = "0.45.1", path = "protocols/kad" }
 libp2p-mdns = { version = "0.45.0", path = "protocols/mdns" }
 libp2p-memory-connection-limits = { version = "0.2.0", path = "misc/memory-connection-limits" }
-libp2p-metrics = { version = "0.14.0", path = "misc/metrics" }
+libp2p-metrics = { version = "0.14.1", path = "misc/metrics" }
 libp2p-mplex = { version = "0.41.0", path = "muxers/mplex" }
 libp2p-muxer-test-harness = { path = "muxers/test-harness" }
 libp2p-noise = { version = "0.44.0", path = "transports/noise" }
@@ -97,7 +97,7 @@ libp2p-quic = { version = "0.10.0", path = "transports/quic" }
 libp2p-relay = { version = "0.17.0", path = "protocols/relay" }
 libp2p-rendezvous = { version = "0.14.0", path = "protocols/rendezvous" }
 libp2p-request-response = { version = "0.26.0", path = "protocols/request-response" }
-libp2p-server = { version = "0.12.3", path = "misc/server" }
+libp2p-server = { version = "0.12.4", path = "misc/server" }
 libp2p-swarm = { version = "0.44.0", path = "swarm" }
 libp2p-swarm-derive = { version = "=0.34.0", path = "swarm-derive" } # `libp2p-swarm-derive` may not be compatible with different `libp2p-swarm` non-breaking releases. E.g. `libp2p-swarm` might introduce a new enum variant `FromSwarm` (which is `#[non-exhaustive]`) in a non-breaking release. Older versions of `libp2p-swarm-derive` would not forward this enum variant within the `NetworkBehaviour` hierarchy. Thus the version pinning is required.
 libp2p-swarm-test = { version = "0.3.0", path = "swarm-test" }

examples/metrics/src/main.rs

@@ -40,13 +40,16 @@ mod http_service;
 async fn main() -> Result<(), Box<dyn Error>> {
     setup_tracing()?;
 
+    let mut metric_registry = Registry::default();
+
     let mut swarm = libp2p::SwarmBuilder::with_new_identity()
         .with_tokio()
         .with_tcp(
             tcp::Config::default(),
             noise::Config::new,
             yamux::Config::default,
         )?
+        .with_bandwidth_metrics(&mut metric_registry)
         .with_behaviour(|key| Behaviour::new(key.public()))?
         .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(Duration::from_secs(u64::MAX)))
         .build();
@@ -59,7 +62,6 @@ async fn main() -> Result<(), Box<dyn Error>> {
         tracing::info!(address=%addr, "Dialed address")
     }
 
-    let mut metric_registry = Registry::default();
     let metrics = Metrics::new(&mut metric_registry);
     tokio::spawn(http_service::metrics_server(metric_registry));
 

libp2p/CHANGELOG.md

@@ -18,6 +18,9 @@
 - Remove deprecated `development_transport`.
   Use `libp2p::SwarmBuilder` instead.
   See [PR 4732](https://github.com/libp2p/rust-libp2p/pull/4732).
+- Introduce `SwarmBuilder::with_bandwidth_metrics` exposing Prometheus bandwidth metrics per transport protocol stack and direction (in-/ outbound).
+  Deprecate `Transport::with_bandwidth_logging` and `SwarmBuilder::with_bandwidth_logging` in favor of the new `SwarmBuilder::with_bandwidth_metrics`.
+  See [PR 4727](https://github.com/libp2p/rust-libp2p/pull/4727).
 
 ## 0.52.4
 

libp2p/src/bandwidth.rs

@@ -18,6 +18,8 @@
 // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 // DEALINGS IN THE SOFTWARE.
 
+#![allow(deprecated)]
+
 use crate::core::muxing::{StreamMuxer, StreamMuxerEvent};
 
 use futures::{
@@ -101,6 +103,9 @@ where
 }
 
 /// Allows obtaining the average bandwidth of the streams.
+#[deprecated(
+    note = "Use `libp2p::SwarmBuilder::with_bandwidth_metrics` or `libp2p_metrics::BandwidthTransport` instead."
+)]
 pub struct BandwidthSinks {
     inbound: AtomicU64,
     outbound: AtomicU64,

libp2p/src/builder.rs

@@ -73,7 +73,7 @@ mod tests {
     use crate::SwarmBuilder;
     use libp2p_core::{muxing::StreamMuxerBox, transport::dummy::DummyTransport};
     use libp2p_identity::PeerId;
-    use libp2p_swarm::{NetworkBehaviour, Swarm};
+    use libp2p_swarm::NetworkBehaviour;
 
     #[test]
     #[cfg(all(
@@ -524,6 +524,7 @@ mod tests {
         feature = "dns",
         feature = "relay",
         feature = "websocket",
+        feature = "metrics",
     ))]
     async fn all() {
         #[derive(NetworkBehaviour)]
@@ -532,7 +533,7 @@ mod tests {
             relay: libp2p_relay::client::Behaviour,
         }
 
-        let (builder, _bandwidth_sinks) = SwarmBuilder::with_new_identity()
+        let _ = SwarmBuilder::with_new_identity()
             .with_tokio()
             .with_tcp(
                 Default::default(),
@@ -548,26 +549,23 @@ mod tests {
             .unwrap()
             .with_relay_client(libp2p_tls::Config::new, libp2p_yamux::Config::default)
             .unwrap()
-            .with_bandwidth_logging();
-        let _: Swarm<MyBehaviour> = builder
+            .with_bandwidth_metrics(&mut libp2p_metrics::Registry::default())
             .with_behaviour(|_key, relay| MyBehaviour { relay })
             .unwrap()
             .build();
     }
 
     #[test]
     #[cfg(all(feature = "tokio", feature = "tcp", feature = "tls", feature = "yamux"))]
-    fn tcp_bandwidth_logging() -> Result<(), Box<dyn std::error::Error>> {
-        let (builder, _logging) = SwarmBuilder::with_new_identity()
+    fn tcp_bandwidth_metrics() -> Result<(), Box<dyn std::error::Error>> {
+        let _ = SwarmBuilder::with_new_identity()
             .with_tokio()
             .with_tcp(
                 Default::default(),
                 libp2p_tls::Config::new,
                 libp2p_yamux::Config::default,
             )?
-            .with_bandwidth_logging();
-
-        builder
+            .with_bandwidth_metrics(&mut libp2p_metrics::Registry::default())
             .with_behaviour(|_| libp2p_swarm::dummy::Behaviour)
             .unwrap()
             .build();
@@ -577,13 +575,11 @@ mod tests {
 
     #[test]
     #[cfg(all(feature = "tokio", feature = "quic"))]
-    fn quic_bandwidth_logging() -> Result<(), Box<dyn std::error::Error>> {
-        let (builder, _logging) = SwarmBuilder::with_new_identity()
+    fn quic_bandwidth_metrics() -> Result<(), Box<dyn std::error::Error>> {
+        let _ = SwarmBuilder::with_new_identity()
             .with_tokio()
             .with_quic()
-            .with_bandwidth_logging();
-
-        builder
+            .with_bandwidth_metrics(&mut libp2p_metrics::Registry::default())
             .with_behaviour(|_| libp2p_swarm::dummy::Behaviour)
             .unwrap()
             .build();
@@ -593,13 +589,11 @@ mod tests {
 
     #[test]
     #[cfg(feature = "tokio")]
-    fn other_transport_bandwidth_logging() -> Result<(), Box<dyn std::error::Error>> {
-        let (builder, _logging) = SwarmBuilder::with_new_identity()
+    fn other_transport_bandwidth_metrics() -> Result<(), Box<dyn std::error::Error>> {
+        let _ = SwarmBuilder::with_new_identity()
             .with_tokio()
             .with_other_transport(|_| DummyTransport::<(PeerId, StreamMuxerBox)>::new())?
-            .with_bandwidth_logging();
-
-        builder
+            .with_bandwidth_metrics(&mut libp2p_metrics::Registry::default())
             .with_behaviour(|_| libp2p_swarm::dummy::Behaviour)
             .unwrap()
             .build();

libp2p/src/builder/phase.rs

@@ -1,6 +1,7 @@
 #![allow(unused_imports)]
 
 mod bandwidth_logging;
+mod bandwidth_metrics;
 mod behaviour;
 mod build;
 mod dns;
@@ -14,6 +15,7 @@ mod tcp;
 mod websocket;
 
 use bandwidth_logging::*;
+use bandwidth_metrics::*;
 use behaviour::*;
 use build::*;
 use dns::*;

libp2p/src/builder/phase/bandwidth_logging.rs

@@ -1,4 +1,5 @@
 use super::*;
+#[allow(deprecated)]
 use crate::bandwidth::BandwidthSinks;
 use crate::transport_ext::TransportExt;
 use crate::SwarmBuilder;
@@ -13,16 +14,18 @@ pub struct BandwidthLoggingPhase<T, R> {
 impl<T: AuthenticatedMultiplexedTransport, Provider, R>
     SwarmBuilder<Provider, BandwidthLoggingPhase<T, R>>
 {
+    #[allow(deprecated)]
+    #[deprecated(note = "Use `with_bandwidth_metrics` instead.")]
     pub fn with_bandwidth_logging(
         self,
     ) -> (
-        SwarmBuilder<Provider, BehaviourPhase<impl AuthenticatedMultiplexedTransport, R>>,
+        SwarmBuilder<Provider, BandwidthMetricsPhase<impl AuthenticatedMultiplexedTransport, R>>,
         Arc<BandwidthSinks>,
     ) {
         let (transport, sinks) = self.phase.transport.with_bandwidth_logging();
         (
             SwarmBuilder {
-                phase: BehaviourPhase {
+                phase: BandwidthMetricsPhase {
                     relay_behaviour: self.phase.relay_behaviour,
                     transport,
                 },
@@ -33,9 +36,9 @@ impl<T: AuthenticatedMultiplexedTransport, Provider, R>
         )
     }
 
-    pub fn without_bandwidth_logging(self) -> SwarmBuilder<Provider, BehaviourPhase<T, R>> {
+    pub fn without_bandwidth_logging(self) -> SwarmBuilder<Provider, BandwidthMetricsPhase<T, R>> {
         SwarmBuilder {
-            phase: BehaviourPhase {
+            phase: BandwidthMetricsPhase {
                 relay_behaviour: self.phase.relay_behaviour,
                 transport: self.phase.transport,
             },
@@ -46,6 +49,18 @@ impl<T: AuthenticatedMultiplexedTransport, Provider, R>
 }
 
 // Shortcuts
+#[cfg(feature = "metrics")]
+impl<Provider, T: AuthenticatedMultiplexedTransport, R>
+    SwarmBuilder<Provider, BandwidthLoggingPhase<T, R>>
+{
+    pub fn with_bandwidth_metrics(
+        self,
+        registry: &mut libp2p_metrics::Registry,
+    ) -> SwarmBuilder<Provider, BehaviourPhase<impl AuthenticatedMultiplexedTransport, R>> {
+        self.without_bandwidth_logging()
+            .with_bandwidth_metrics(registry)
+    }
+}
 #[cfg(feature = "relay")]
 impl<Provider, T: AuthenticatedMultiplexedTransport>
     SwarmBuilder<Provider, BandwidthLoggingPhase<T, libp2p_relay::client::Behaviour>>
@@ -54,17 +69,20 @@ impl<Provider, T: AuthenticatedMultiplexedTransport>
         self,
         constructor: impl FnOnce(&libp2p_identity::Keypair, libp2p_relay::client::Behaviour) -> R,
     ) -> Result<SwarmBuilder<Provider, SwarmPhase<T, B>>, R::Error> {
-        self.without_bandwidth_logging().with_behaviour(constructor)
+        self.without_bandwidth_logging()
+            .without_bandwidth_metrics()
+            .with_behaviour(constructor)
     }
 }
-
 impl<Provider, T: AuthenticatedMultiplexedTransport>
     SwarmBuilder<Provider, BandwidthLoggingPhase<T, NoRelayBehaviour>>
 {
     pub fn with_behaviour<B, R: TryIntoBehaviour<B>>(
         self,
         constructor: impl FnOnce(&libp2p_identity::Keypair) -> R,
     ) -> Result<SwarmBuilder<Provider, SwarmPhase<T, B>>, R::Error> {
-        self.without_bandwidth_logging().with_behaviour(constructor)
+        self.without_bandwidth_logging()
+            .without_bandwidth_metrics()
+            .with_behaviour(constructor)
     }
 }

libp2p/src/builder/phase/bandwidth_metrics.rs

@@ -0,0 +1,69 @@
+use super::*;
+#[allow(deprecated)]
+use crate::bandwidth::BandwidthSinks;
+use crate::transport_ext::TransportExt;
+use crate::SwarmBuilder;
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+pub struct BandwidthMetricsPhase<T, R> {
+    pub(crate) relay_behaviour: R,
+    pub(crate) transport: T,
+}
+
+#[cfg(feature = "metrics")]
+impl<T: AuthenticatedMultiplexedTransport, Provider, R>
+    SwarmBuilder<Provider, BandwidthMetricsPhase<T, R>>
+{
+    pub fn with_bandwidth_metrics(
+        self,
+        registry: &mut libp2p_metrics::Registry,
+    ) -> SwarmBuilder<Provider, BehaviourPhase<impl AuthenticatedMultiplexedTransport, R>> {
+        SwarmBuilder {
+            phase: BehaviourPhase {
+                relay_behaviour: self.phase.relay_behaviour,
+                transport: libp2p_metrics::BandwidthTransport::new(self.phase.transport, registry)
+                    .map(|(peer_id, conn), _| (peer_id, StreamMuxerBox::new(conn))),
+            },
+            keypair: self.keypair,
+            phantom: PhantomData,
+        }
+    }
+}
+
+impl<T, Provider, R> SwarmBuilder<Provider, BandwidthMetricsPhase<T, R>> {
+    pub fn without_bandwidth_metrics(self) -> SwarmBuilder<Provider, BehaviourPhase<T, R>> {
+        SwarmBuilder {
+            phase: BehaviourPhase {
+                relay_behaviour: self.phase.relay_behaviour,
+                transport: self.phase.transport,
+            },
+            keypair: self.keypair,
+            phantom: PhantomData,
+        }
+    }
+}
+
+// Shortcuts
+#[cfg(feature = "relay")]
+impl<Provider, T: AuthenticatedMultiplexedTransport>
+    SwarmBuilder<Provider, BandwidthMetricsPhase<T, libp2p_relay::client::Behaviour>>
+{
+    pub fn with_behaviour<B, R: TryIntoBehaviour<B>>(
+        self,
+        constructor: impl FnOnce(&libp2p_identity::Keypair, libp2p_relay::client::Behaviour) -> R,
+    ) -> Result<SwarmBuilder<Provider, SwarmPhase<T, B>>, R::Error> {
+        self.without_bandwidth_metrics().with_behaviour(constructor)
+    }
+}
+
+impl<Provider, T: AuthenticatedMultiplexedTransport>
+    SwarmBuilder<Provider, BandwidthMetricsPhase<T, NoRelayBehaviour>>
+{
+    pub fn with_behaviour<B, R: TryIntoBehaviour<B>>(
+        self,
+        constructor: impl FnOnce(&libp2p_identity::Keypair) -> R,
+    ) -> Result<SwarmBuilder<Provider, SwarmPhase<T, B>>, R::Error> {
+        self.without_bandwidth_metrics().with_behaviour(constructor)
+    }
+}