chore(p2p_service): add metrics for number of blocks requested over p…

…2p req/res protocol

CHANGELOG.md

@@ -8,6 +8,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Added
 - [2122](https://github.com/FuelLabs/fuel-core/pull/2122): Changed the relayer URI address to be a vector and use a quorum provider. The `relayer` argument now supports multiple URLs to fetch information from different sources.
+- [2135](https://github.com/FuelLabs/fuel-core/pull/2135): Added metrics logging for number of blocks served over the p2p req/res protocol.
 
 ### Changed
 - [2113](https://github.com/FuelLabs/fuel-core/pull/2113): Modify the way the gas price service and shared algo is initialized to have some default value based on best guess instead of `None`, and initialize service before graphql.

crates/metrics/src/p2p_metrics.rs

@@ -1,16 +1,24 @@
 use crate::global_registry;
-use prometheus_client::metrics::counter::Counter;
+use prometheus_client::metrics::{
+    counter::Counter,
+    gauge::Gauge,
+};
 use std::sync::OnceLock;
 
 pub struct P2PMetrics {
     pub unique_peers: Counter,
+    pub blocks_requested: Gauge,
 }
 
 impl P2PMetrics {
     fn new() -> Self {
         let unique_peers = Counter::default();
+        let blocks_requested = Gauge::default();
 
-        let metrics = P2PMetrics { unique_peers };
+        let metrics = P2PMetrics {
+            unique_peers,
+            blocks_requested,
+        };
 
         let mut registry = global_registry().registry.lock();
         registry.register(
@@ -19,6 +27,12 @@ impl P2PMetrics {
             metrics.unique_peers.clone(),
         );
 
+        registry.register(
+            "Blocks_Requested",
+            "A Gauge which keeps track of how many blocks were requested and served over the p2p req/res protocol",
+            metrics.blocks_requested.clone()
+        );
+
         metrics
     }
 }
@@ -28,3 +42,11 @@ static P2P_METRICS: OnceLock<P2PMetrics> = OnceLock::new();
 pub fn p2p_metrics() -> &'static P2PMetrics {
     P2P_METRICS.get_or_init(P2PMetrics::new)
 }
+
+pub fn inc_unique_peers() {
+    p2p_metrics().unique_peers.inc();
+}
+
+pub fn set_blocks_requested(count: usize) {
+    p2p_metrics().blocks_requested.set(count as i64);
+}

crates/services/p2p/src/p2p_service.rs

@@ -35,7 +35,7 @@ use crate::{
     },
     TryPeerId,
 };
-use fuel_core_metrics::p2p_metrics::p2p_metrics;
+use fuel_core_metrics::p2p_metrics::inc_unique_peers;
 use fuel_core_types::{
     fuel_types::BlockHeight,
     services::p2p::peer_reputation::AppScore,
@@ -271,6 +271,15 @@ impl FuelP2PService {
         }
     }
 
+    pub fn log_metrics<T>(&self, cb: T)
+    where
+        T: FnOnce(),
+    {
+        if self.metrics {
+            cb();
+        }
+    }
+
     #[cfg(feature = "test-helpers")]
     pub fn multiaddrs(&self) -> Vec<Multiaddr> {
         let local_peer = self.local_peer_id;
@@ -644,9 +653,7 @@ impl FuelP2PService {
     fn handle_identify_event(&mut self, event: identify::Event) -> Option<FuelP2PEvent> {
         match event {
             identify::Event::Received { peer_id, info } => {
-                if self.metrics {
-                    p2p_metrics().unique_peers.inc();
-                }
+                self.log_metrics(inc_unique_peers);
 
                 let mut addresses = info.listen_addrs;
                 let agent_version = info.agent_version;

crates/services/p2p/src/service.rs

@@ -26,6 +26,7 @@ use crate::{
     },
 };
 use anyhow::anyhow;
+use fuel_core_metrics::p2p_metrics::set_blocks_requested;
 use fuel_core_services::{
     stream::BoxStream,
     RunnableService,
@@ -196,9 +197,20 @@ pub trait TaskP2PService: Send {
     ) -> anyhow::Result<()>;
 
     fn update_block_height(&mut self, height: BlockHeight) -> anyhow::Result<()>;
+
+    fn log_metrics<T>(&self, cb: T)
+    where
+        T: FnOnce();
 }
 
 impl TaskP2PService for FuelP2PService {
+    fn log_metrics<T>(&self, cb: T)
+    where
+        T: FnOnce(),
+    {
+        FuelP2PService::log_metrics(self, cb)
+    }
+
     fn get_all_peer_info(&self) -> Vec<(&PeerId, &PeerInfo)> {
         self.peer_manager().get_all_peers().collect()
     }
@@ -427,6 +439,13 @@ where
     V: AtomicView + 'static,
     V::LatestView: P2pDb,
 {
+    fn log_metrics<T>(&self, cb: T)
+    where
+        T: FnOnce(),
+    {
+        self.p2p_service.log_metrics(cb)
+    }
+
     fn process_request(
         &mut self,
         request_message: RequestMessage,
@@ -464,8 +483,11 @@ where
         // If there are other types of data we send over p2p req/res protocol, then this needs
         // to be generalized
         let max_len = self.max_headers_per_request;
+        let range_len = range.len();
+
+        self.log_metrics(|| set_blocks_requested(range_len));
 
-        if range.len() > max_len {
+        if range_len > max_len {
             tracing::error!(
                 requested_length = range.len(),
                 max_len,
@@ -1031,6 +1053,13 @@ pub mod tests {
     }
 
     impl TaskP2PService for FakeP2PService {
+        fn log_metrics<T>(&self, _: T)
+        where
+            T: FnOnce(),
+        {
+            unimplemented!()
+        }
+
         fn get_all_peer_info(&self) -> Vec<(&PeerId, &PeerInfo)> {
             self.peer_info.iter().map(|tup| (&tup.0, &tup.1)).collect()
         }