apollo_network: added libp2p metrics

Author: sirandreww-starkware

Cargo.lock

@@ -261,6 +261,7 @@ phf = "0.11"
 pretty_assertions = "1.4.0"
 primitive-types = "0.12.1"
 proc-macro2 = "1.0"
+prometheus-client = "0.23.1"
 prometheus-parse = "0.2.4"
 prost = "0.12.1"
 prost-build = "0.12.1"

Cargo.toml

@@ -98,6 +98,7 @@ impl ConsensusManager {
             num_blacklisted_peers: CONSENSUS_NUM_BLACKLISTED_PEERS,
             broadcast_metrics_by_topic: Some(broadcast_metrics_by_topic),
             sqmr_metrics: None,
+            libp2p_metrics_prefix: None,
         });
         let mut network_manager =
             NetworkManager::new(self.config.network_config.clone(), None, network_manager_metrics);

crates/apollo_consensus_manager/src/consensus_manager.rs

@@ -50,6 +50,7 @@ pub fn create_p2p_propagator_and_runner(
         num_blacklisted_peers: MEMPOOL_P2P_NUM_BLACKLISTED_PEERS,
         broadcast_metrics_by_topic: Some(broadcast_metrics_by_topic),
         sqmr_metrics: None,
+        libp2p_metrics_prefix: None,
     });
     let mut network_manager = NetworkManager::new(
         mempool_p2p_config.network_config,

crates/apollo_mempool_p2p/src/lib.rs

@@ -25,6 +25,7 @@ libp2p = { workspace = true, features = [
   "identify",
   "kad",
   "macros",
+  "metrics",
   "noise",
   "quic",
   "serde",
@@ -34,6 +35,8 @@ libp2p = { workspace = true, features = [
 ] }
 metrics.workspace = true
 metrics-exporter-prometheus.workspace = true
+prometheus-client.workspace = true
+prometheus-parse.workspace = true
 replace_with.workspace = true
 serde = { workspace = true, features = ["derive"] }
 starknet_api.workspace = true

crates/apollo_network/Cargo.toml

@@ -0,0 +1,54 @@
+use libp2p::metrics::Registry;
+use metrics::{counter, describe_counter, describe_gauge, gauge};
+use prometheus_client::encoding::text::encode;
+use prometheus_parse::Value;
+
+/// libp2p uses `prometheus-client` for metrics, which updates the metrics to a `Registry`.
+/// We use `metrics-exporter-prometheus` so we need to update these
+/// metrics when the registry is updated.
+pub fn connect_libp2p_registry_to_metrics_exporter_prometheus(prefix: String, registry: Registry) {
+    tokio::spawn(async move {
+        // This will update the libp2p metrics in the registry.
+        loop {
+            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+
+            let mut sink = String::new();
+            encode(&mut sink, &registry).unwrap();
+
+            let metrics =
+                prometheus_parse::Scrape::parse(sink.lines().map(|s| Ok(s.to_owned()))).unwrap();
+
+            for sample in metrics.samples.iter() {
+                let metric_name = format!("{}{}", prefix, sample.metric);
+
+                match &sample.value {
+                    Value::Counter(value) => {
+                        // Register and update counter metric
+                        describe_counter!(metric_name.clone(), "LibP2P counter metric");
+                        #[allow(clippy::as_conversions)]
+                        counter!(metric_name).absolute(*value as u64);
+                    }
+                    Value::Gauge(value) => {
+                        // Register and update gauge metric
+                        describe_gauge!(metric_name.clone(), "LibP2P gauge metric");
+                        gauge!(metric_name).set(*value);
+                    }
+                    Value::Histogram(_) => {
+                        // Skip histograms for now as they require more complex handling
+
+                        continue;
+                    }
+                    Value::Summary(_) => {
+                        // Skip summaries for now as they require more complex handling
+                        continue;
+                    }
+                    Value::Untyped(value) => {
+                        // Treat untyped metrics as gauges
+                        describe_gauge!(metric_name.clone(), "LibP2P untyped metric");
+                        gauge!(metric_name).set(*value);
+                    }
+                }
+            }
+        }
+    });
+}

crates/apollo_network/src/network_manager/libp2p_metrics.rs

@@ -36,6 +36,10 @@ pub struct NetworkMetrics {
     pub num_blacklisted_peers: MetricGauge,
     pub broadcast_metrics_by_topic: Option<HashMap<TopicHash, BroadcastNetworkMetrics>>,
     pub sqmr_metrics: Option<SqmrNetworkMetrics>,
+    /// Set the desired prefix for the libp2p metrics.
+    /// If `None`, the metrics will not be registered.
+    /// If `Some(prefix)`, the metrics will be registered with the prefix.
+    pub libp2p_metrics_prefix: Option<String>,
 }
 
 impl NetworkMetrics {
@@ -52,5 +56,30 @@ impl NetworkMetrics {
         if let Some(sqmr_metrics) = self.sqmr_metrics.as_ref() {
             sqmr_metrics.register();
         }
+        // if let Some(libp2p_metrics) = self.libp2p_metrics.as_ref() {
+        //     libp2p_metrics.register();
+        // }
     }
 }
+
+// pub struct LibP2PMetrics {
+//     pub num_connected_peers: MetricGauge,
+//     pub num_blacklisted_peers: MetricGauge,
+// }
+
+// impl LibP2PMetrics {
+//     pub fn register(&self) {
+//         self.num_connected_peers.register();
+//         self.num_blacklisted_peers.register();
+//     }
+// }
+
+// /// Must implement debug for conversion
+// impl Debug for LibP2PMetrics {
+//     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+//         f.debug_struct("LibP2PMetrics")
+//             .field("num_connected_peers", &self.num_connected_peers.get_scope())
+//             .field("num_blacklisted_peers", &self.num_blacklisted_peers.get_scope())
+//             .finish()
+//     }
+// }

crates/apollo_network/src/network_manager/metrics.rs

@@ -1,3 +1,4 @@
+mod libp2p_metrics;
 pub mod metrics;
 mod swarm_trait;
 #[cfg(test)]
@@ -20,6 +21,7 @@ use futures::stream::{FuturesUnordered, Map, Stream};
 use futures::{pin_mut, FutureExt, Sink, SinkExt, StreamExt};
 use libp2p::gossipsub::{SubscriptionError, TopicHash};
 use libp2p::identity::Keypair;
+use libp2p::metrics::{Metrics, Recorder, Registry};
 use libp2p::swarm::SwarmEvent;
 use libp2p::{noise, yamux, Multiaddr, PeerId, StreamProtocol, Swarm, SwarmBuilder};
 use metrics::NetworkMetrics;
@@ -29,6 +31,7 @@ use self::swarm_trait::SwarmTrait;
 use crate::gossipsub_impl::Topic;
 use crate::misconduct_score::MisconductScore;
 use crate::mixed_behaviour::{self, BridgedBehaviour};
+use crate::network_manager::libp2p_metrics::connect_libp2p_registry_to_metrics_exporter_prometheus;
 use crate::sqmr::behaviour::SessionError;
 use crate::sqmr::{self, InboundSessionId, OutboundSessionId, SessionId};
 use crate::utils::{is_localhost, make_multiaddr, StreamMap};
@@ -61,6 +64,7 @@ pub struct GenericNetworkManager<SwarmT: SwarmTrait> {
     continue_propagation_sender: Sender<BroadcastedMessageMetadata>,
     continue_propagation_receiver: Receiver<BroadcastedMessageMetadata>,
     metrics: Option<NetworkMetrics>,
+    libp2p_metrics: Option<Metrics>,
 }
 
 impl<SwarmT: SwarmTrait> GenericNetworkManager<SwarmT> {
@@ -95,12 +99,13 @@ impl<SwarmT: SwarmTrait> GenericNetworkManager<SwarmT> {
 
     // TODO(shahak): remove the advertised_multiaddr arg once we manage external addresses
     // in a behaviour.
-    pub(crate) fn generic_new(
+    pub(crate) fn generic_new_with_libp2p_metrics(
         mut swarm: SwarmT,
         advertised_multiaddr: Option<Multiaddr>,
         metrics: Option<NetworkMetrics>,
         broadcasted_message_metadata_buffer_size: usize,
         reported_peer_ids_buffer_size: usize,
+        libp2p_metrics: Option<Metrics>,
     ) -> Self {
         let reported_peer_receivers = FuturesUnordered::new();
         reported_peer_receivers.push(futures::future::pending().boxed());
@@ -128,9 +133,28 @@ impl<SwarmT: SwarmTrait> GenericNetworkManager<SwarmT> {
             continue_propagation_sender,
             continue_propagation_receiver,
             metrics,
+            libp2p_metrics,
         }
     }
 
+    #[cfg(test)]
+    pub(crate) fn generic_new(
+        swarm: SwarmT,
+        advertised_multiaddr: Option<Multiaddr>,
+        metrics: Option<NetworkMetrics>,
+        broadcasted_message_metadata_buffer_size: usize,
+        reported_peer_ids_buffer_size: usize,
+    ) -> Self {
+        Self::generic_new_with_libp2p_metrics(
+            swarm,
+            advertised_multiaddr,
+            metrics,
+            broadcasted_message_metadata_buffer_size,
+            reported_peer_ids_buffer_size,
+            None,
+        )
+    }
+
     // TODO(Shahak): Support multiple protocols where they're all different versions of the same
     // protocol
     pub fn register_sqmr_protocol_server<Query, Response>(
@@ -267,6 +291,9 @@ impl<SwarmT: SwarmTrait> GenericNetworkManager<SwarmT> {
         &mut self,
         event: SwarmEvent<mixed_behaviour::Event>,
     ) -> Result<(), NetworkError> {
+        if let Some(metrics) = self.libp2p_metrics.as_ref() {
+            metrics.record(&event);
+        }
         match event {
             SwarmEvent::ConnectionEstablished { peer_id, num_established, .. } => {
                 debug!("Connected to peer id: {peer_id:?}");
@@ -686,7 +713,7 @@ impl NetworkManager {
     pub fn new(
         config: NetworkConfig,
         node_version: Option<String>,
-        metrics: Option<NetworkMetrics>,
+        mut metrics: Option<NetworkMetrics>,
     ) -> Self {
         let NetworkConfig {
             port,
@@ -711,13 +738,15 @@ impl NetworkManager {
             }
             None => Keypair::generate_ed25519(),
         };
+        let mut registry = Registry::default();
         let mut swarm = SwarmBuilder::with_existing_identity(key_pair)
             .with_tokio()
             // TODO(AndrewL): .with_quic()
             .with_tcp(Default::default(), noise::Config::new, yamux::Config::default)
             .expect("Error building TCP transport")
             .with_dns()
             .expect("Error building DNS transport")
+            .with_bandwidth_metrics(&mut registry)
             .with_behaviour(|key| {
                 mixed_behaviour::MixedBehaviour::new(
                     key.clone(),
@@ -733,6 +762,14 @@ impl NetworkManager {
             .with_swarm_config(|cfg| cfg.with_idle_connection_timeout(idle_connection_timeout))
             .build();
 
+        let libp2p_metrics = Metrics::new(&mut registry);
+
+        if let Some(metrics) = metrics.as_mut() {
+            if let Some(prefix) = metrics.libp2p_metrics_prefix.as_ref() {
+                connect_libp2p_registry_to_metrics_exporter_prometheus(prefix.clone(), registry);
+            }
+        }
+
         swarm
             .listen_on(listen_address.clone())
             .unwrap_or_else(|_| panic!("Error while binding to {listen_address}"));
@@ -742,12 +779,13 @@ impl NetworkManager {
                 .with_p2p(*swarm.local_peer_id())
                 .expect("advertised_multiaddr has a peer id different than the local peer id")
         });
-        Self::generic_new(
+        Self::generic_new_with_libp2p_metrics(
             swarm,
             advertised_multiaddr,
             metrics,
             broadcasted_message_metadata_buffer_size,
             reported_peer_ids_buffer_size,
+            Some(libp2p_metrics),
         )
     }
 

crates/apollo_network/src/network_manager/mod.rs

@@ -201,6 +201,7 @@ impl StateSyncRunner {
                     num_active_inbound_sessions: P2P_SYNC_NUM_ACTIVE_INBOUND_SESSIONS,
                     num_active_outbound_sessions: P2P_SYNC_NUM_ACTIVE_OUTBOUND_SESSIONS,
                 }),
+                libp2p_metrics_prefix: None,
             });
             NetworkManager::new(
                 network_config.clone(),