net/metrics: Add metrics for inbound/outbound traffic (#10846)

This PR adds a new metric for inbound / outbound traffic for individual
request-response protocols.

- the PR is motivated by
#10765 which shows a
significant number of bytes as downloaded (4-5 MiB/s). This is
suspicious for a fully synced validator, 1-2 blocks to the tip of the
chain.
- It suggests a protocol is internally consuming too much bandwidth
leading to network inefficiencies, wasted CPU, and in the case of the
issue to OOM kills

cc @paritytech/sdk-node

---------

Signed-off-by: Alexandru Vasile <[email protected]>
Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: lexnv

prdoc/pr_10846.prdoc

@@ -0,0 +1,13 @@
+title: 'net/metrics: Add metrics for inbound/outbound traffic '
+doc:
+- audience: Node Dev
+  description: |-
+    This PR adds a new metric for inbound / outbound traffic for individual request-response protocols.
+
+    - the PR is motivated by https://github.com/paritytech/polkadot-sdk/issues/10765 which shows a significant number of bytes as downloaded (4-5 MiB/s). This is suspicious for a fully synced validator, 1-2 blocks to the tip of the chain.
+    - It suggests a protocol is internally consuming too much bandwidth leading to network inefficiencies, wasted CPU, and in the case of the issue to OOM kills
+
+    cc @paritytech/sdk-node
+crates:
+- name: sc-network
+  bump: patch

substrate/client/network/src/litep2p/shim/request_response/metrics.rs

@@ -75,4 +75,44 @@ impl RequestResponseMetrics {
 				.observe(duration.as_secs_f64());
 		}
 	}
+
+	/// Register inbound bytes (request payload received from peer) to Prometheus
+	pub fn register_inbound_request_bytes(&self, bytes: usize) {
+		if let Some(metrics) = &self.metrics {
+			metrics
+				.requests_response_bytes_total
+				.with_label_values(&["in", &self.protocol])
+				.inc_by(bytes as u64);
+		}
+	}
+
+	/// Register outbound bytes (response payload sent to peer) to Prometheus
+	pub fn register_outbound_response_bytes(&self, bytes: usize) {
+		if let Some(metrics) = &self.metrics {
+			metrics
+				.requests_response_bytes_total
+				.with_label_values(&["out", &self.protocol])
+				.inc_by(bytes as u64);
+		}
+	}
+
+	/// Register outbound bytes (request payload sent to peer) to Prometheus
+	pub fn register_outbound_request_bytes(&self, bytes: usize) {
+		if let Some(metrics) = &self.metrics {
+			metrics
+				.requests_response_bytes_total
+				.with_label_values(&["out", &self.protocol])
+				.inc_by(bytes as u64);
+		}
+	}
+
+	/// Register inbound bytes (response payload received from peer) to Prometheus
+	pub fn register_inbound_response_bytes(&self, bytes: usize) {
+		if let Some(metrics) = &self.metrics {
+			metrics
+				.requests_response_bytes_total
+				.with_label_values(&["in", &self.protocol])
+				.inc_by(bytes as u64);
+		}
+	}
 }

substrate/client/network/src/litep2p/shim/request_response/mod.rs

@@ -245,8 +245,10 @@ impl RequestResponseProtocol {
 			dial_options,
 		);
 
+		let request_len = request.len();
 		match self.handle.try_send_request(peer.into(), request, dial_options) {
 			Ok(request_id) => {
+				self.metrics.register_outbound_request_bytes(request_len);
 				self.pending_inbound_responses
 					.insert(request_id, PendingRequest::new(tx, Instant::now(), fallback_request));
 			},
@@ -280,6 +282,8 @@ impl RequestResponseProtocol {
 			request.len(),
 		);
 
+		self.metrics.register_inbound_request_bytes(request.len());
+
 		let Some(inbound_queue) = &self.inbound_queue else {
 			log::trace!(
 				target: LOG_TARGET,
@@ -350,6 +354,7 @@ impl RequestResponseProtocol {
 					response.len(),
 				);
 
+				self.metrics.register_inbound_response_bytes(response.len());
 				let _ = tx.send(Ok((response, self.protocol.clone())));
 				self.metrics.register_outbound_request_success(started.elapsed());
 			},
@@ -512,6 +517,8 @@ impl RequestResponseProtocol {
 					response.len(),
 				);
 
+				self.metrics.register_outbound_response_bytes(response.len());
+
 				match sent_feedback {
 					None => self.handle.send_response(request_id, response),
 					Some(feedback) =>

substrate/client/network/src/service/metrics.rs

@@ -85,6 +85,7 @@ pub struct Metrics {
 	pub requests_in_success_total: HistogramVec,
 	pub requests_out_failure_total: CounterVec<U64>,
 	pub requests_out_success_total: HistogramVec,
+	pub requests_response_bytes_total: CounterVec<U64>,
 }
 
 impl Metrics {
@@ -211,6 +212,13 @@ impl Metrics {
 				},
 				&["protocol"]
 			)?, registry)?,
+			requests_response_bytes_total: prometheus::register(CounterVec::new(
+				Opts::new(
+					"substrate_sub_libp2p_requests_response_bytes_total",
+					"Total bytes sent and received by request-response protocols"
+				),
+				&["direction", "protocol"]
+			)?, registry)?,
 		})
 	}
 }