peerset/metrics: Add notification layer metrics for peer states

prdoc/pr_11574.prdoc

@@ -0,0 +1,14 @@
+title: 'peerset/metrics: Add notification layer metrics for peer states'
+doc:
+- audience: Node Dev
+  description: |-
+    This PR adds two new metrics to enhance our monitoring insights into the notification protocol.
+
+    The data exposed:
+    - connected peers grouped by in / out and kind (reserve / non-reserve)
+    - disconnected and backed off peers
+
+    Because the slot counts are tightly coupled with the full peerset state machine, I've picked an eventually consistent model to easily bump the metrics at 1s tick intervals.
+crates:
+- name: sc-network
+  bump: patch

substrate/client/network/src/litep2p/shim/notification/config.rs

@@ -96,6 +96,7 @@ impl NotificationProtocolConfig {
 			set_config.reserved_nodes.iter().map(|address| address.peer_id).collect(),
 			Arc::clone(&connected_peers),
 			peerstore_handle,
+			metrics.clone(),
 		);
 
 		// create `litep2p` notification protocol configuration for the protocol

substrate/client/network/src/litep2p/shim/notification/peerset.rs

@@ -40,7 +40,10 @@
 
 use crate::{
 	peer_store::{PeerStoreProvider, ProtocolHandle},
-	service::traits::{self, ValidationResult},
+	service::{
+		metrics::NotificationMetrics,
+		traits::{self, ValidationResult},
+	},
 	ProtocolName, ReputationChange as Reputation,
 };
 
@@ -368,6 +371,9 @@ pub struct Peerset {
 
 	/// Next time when [`Peerset`] should perform slot allocation.
 	next_slot_allocation: Delay,
+
+	/// Notification metrics.
+	metrics: NotificationMetrics,
 }
 
 macro_rules! decrement_or_warn {
@@ -412,6 +418,7 @@ impl Peerset {
 		reserved_peers: HashSet<PeerId>,
 		connected_peers: Arc<AtomicUsize>,
 		peerstore_handle: Arc<dyn PeerStoreProvider>,
+		metrics: NotificationMetrics,
 	) -> (Self, TracingUnboundedSender<PeersetCommand>) {
 		let (cmd_tx, cmd_rx) = tracing_unbounded("mpsc-peerset-protocol", 100_000);
 		let peers = reserved_peers
@@ -447,6 +454,7 @@ impl Peerset {
 				connected_peers,
 				pending_backoffs: FuturesUnordered::new(),
 				next_slot_allocation: Delay::new(SLOT_ALLOCATION_FREQUENCY),
+				metrics,
 			},
 			cmd_tx,
 		)
@@ -931,6 +939,44 @@ impl Peerset {
 		}
 	}
 
+	/// Report connected peer counts to metrics.
+	fn update_slot_metrics(&self) {
+		let (mut in_reserved, mut in_non_reserved) = (0usize, 0usize);
+		let (mut out_reserved, mut out_non_reserved) = (0usize, 0usize);
+		let (mut num_disconnected, mut num_backoff) = (0usize, 0usize);
+
+		for state in self.peers.values() {
+			match state {
+				PeerState::Connected { direction: Direction::Inbound(Reserved::Yes) } => {
+					in_reserved += 1
+				},
+				PeerState::Connected { direction: Direction::Inbound(Reserved::No) } => {
+					in_non_reserved += 1
+				},
+				PeerState::Connected { direction: Direction::Outbound(Reserved::Yes) } => {
+					out_reserved += 1
+				},
+				PeerState::Connected { direction: Direction::Outbound(Reserved::No) } => {
+					out_non_reserved += 1
+				},
+				PeerState::Disconnected => num_disconnected += 1,
+				PeerState::Backoff => num_backoff += 1,
+
+				_ => {},
+			}
+		}
+
+		self.metrics.set_peerset_num_connected(
+			&self.protocol,
+			in_reserved,
+			in_non_reserved,
+			out_reserved,
+			out_non_reserved,
+			num_disconnected,
+			num_backoff,
+		);
+	}
+
 	/// Connect to all reserved peers.
 	///
 	/// Under the following conditions:
@@ -1515,10 +1561,14 @@ impl Stream for Peerset {
 				}
 			}
 
-			// start timer for the next allocation and if there were peers which the `Peerset`
+			// Start timer for the next allocation and if there were peers which the `Peerset`
 			// wasn't connected but should be, send command to litep2p to start opening substreams.
 			self.next_slot_allocation = Delay::new(SLOT_ALLOCATION_FREQUENCY);
 
+			// Update metrics on every tick of slot allocation. This ensures metrics are
+			// eventually consistent at 1s intervals.
+			self.update_slot_metrics();
+
 			if !connect_to.is_empty() {
 				log::trace!(
 					target: LOG_TARGET,

substrate/client/network/src/litep2p/shim/notification/tests/fuzz.rs

@@ -24,7 +24,10 @@ use crate::{
 		peerstore::Peerstore,
 		shim::notification::peerset::{OpenResult, Peerset, PeersetCommand},
 	},
-	service::traits::{Direction, PeerStore, ValidationResult},
+	service::{
+		metrics::NotificationMetrics,
+		traits::{Direction, PeerStore, ValidationResult},
+	},
 	ProtocolName,
 };
 
@@ -94,6 +97,7 @@ async fn test_once() {
 			.collect(),
 		Default::default(),
 		Arc::clone(&peer_store_handle),
+		NotificationMetrics::new(None),
 	);
 
 	tokio::spawn(peerstore.run());

substrate/client/network/src/litep2p/shim/notification/tests/peerset.rs

@@ -23,7 +23,10 @@ use crate::{
 			Direction, OpenResult, PeerState, Peerset, PeersetCommand, Reserved,
 		},
 	},
-	service::traits::{self, ValidationResult},
+	service::{
+		metrics::NotificationMetrics,
+		traits::{self, ValidationResult},
+	},
 	ProtocolName,
 };
 
@@ -65,6 +68,7 @@ async fn inbound_substream_for_outbound_peer() {
 		Default::default(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -114,6 +118,7 @@ async fn canceled_peer_gets_banned() {
 		peers.clone(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -168,6 +173,7 @@ async fn peer_added_and_removed_from_peerset() {
 		Default::default(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -290,6 +296,7 @@ async fn set_reserved_peers() {
 		reserved.clone(),
 		Default::default(),
 		Arc::new(peerstore_handle_test()),
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -374,6 +381,7 @@ async fn set_reserved_peers_one_peer_already_in_the_set() {
 		reserved.clone(),
 		Default::default(),
 		Arc::new(peerstore_handle_test()),
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -478,6 +486,7 @@ async fn add_reserved_peers_one_peer_already_in_the_set() {
 		reserved.iter().cloned().collect(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -570,6 +579,7 @@ async fn opening_peer_gets_canceled_and_disconnected() {
 		Default::default(),
 		Arc::clone(&num_connected),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0);
 	assert_eq!(peerset.num_out(), 0);
@@ -649,6 +659,7 @@ async fn open_failure_for_canceled_peer() {
 		Default::default(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -713,6 +724,7 @@ async fn peer_disconnected_when_being_validated_then_rejected() {
 		Default::default(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -745,6 +757,7 @@ async fn removed_reserved_peer_kept_due_to_free_slots() {
 		peers.clone(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -815,6 +828,7 @@ async fn set_reserved_peers_but_available_slots() {
 		Default::default(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -919,6 +933,7 @@ async fn set_reserved_peers_move_previously_reserved() {
 		known_peers.clone(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -1058,6 +1073,7 @@ async fn set_reserved_peers_cannot_move_previously_reserved() {
 		known_peers.clone(),
 		Default::default(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);
@@ -1157,6 +1173,7 @@ async fn reserved_only_rejects_non_reserved_peers() {
 		reserved_peers.clone(),
 		connected_peers.clone(),
 		peerstore_handle,
+		NotificationMetrics::new(None),
 	);
 	assert_eq!(peerset.num_in(), 0usize);
 	assert_eq!(peerset.num_out(), 0usize);

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

@@ -365,6 +365,46 @@ impl NotificationMetrics {
 				.observe(size as f64);
 		}
 	}
+
+	/// Update the number of connected peers per direction and reservation status.
+	pub fn set_peerset_num_connected(
+		&self,
+		protocol: &ProtocolName,
+		in_reserved: usize,
+		in_non_reserved: usize,
+		out_reserved: usize,
+		out_non_reserved: usize,
+		num_disconnected: usize,
+		num_backoff: usize,
+	) {
+		if let Some(metrics) = &self.metrics {
+			metrics
+				.peerset_num_connected
+				.with_label_values(&["in", "reserved", protocol])
+				.set(in_reserved as u64);
+			metrics
+				.peerset_num_connected
+				.with_label_values(&["in", "non-reserved", protocol])
+				.set(in_non_reserved as u64);
+			metrics
+				.peerset_num_connected
+				.with_label_values(&["out", "reserved", protocol])
+				.set(out_reserved as u64);
+			metrics
+				.peerset_num_connected
+				.with_label_values(&["out", "non-reserved", protocol])
+				.set(out_non_reserved as u64);
+
+			metrics
+				.peerset_num_state
+				.with_label_values(&["disconnected", protocol])
+				.set(num_disconnected as u64);
+			metrics
+				.peerset_num_state
+				.with_label_values(&["backoff", protocol])
+				.set(num_backoff as u64);
+		}
+	}
 }
 
 /// Notification metrics.
@@ -378,6 +418,12 @@ struct InnerNotificationMetrics {
 
 	/// In/outbound notification sizes.
 	pub notifications_sizes: HistogramVec,
+
+	/// Number of connected peers per direction, reservation status and protocol.
+	pub peerset_num_connected: GaugeVec<U64>,
+
+	/// Number of disconnected and backed off peers.
+	pub peerset_num_state: GaugeVec<U64>,
 }
 
 impl InnerNotificationMetrics {
@@ -417,6 +463,26 @@ impl InnerNotificationMetrics {
 				)?,
 				registry,
 			)?,
+			peerset_num_connected: prometheus::register(
+				GaugeVec::new(
+					Opts::new(
+						"substrate_sub_libp2p_peerset_num_connected",
+						"Number of connected peers per direction, reservation status and protocol",
+					),
+					&["direction", "kind", "protocol"],
+				)?,
+				registry,
+			)?,
+			peerset_num_state: prometheus::register(
+				GaugeVec::new(
+					Opts::new(
+						"substrate_sub_libp2p_peerset_num_state",
+						"Number of peers per state in the peerset manager",
+					),
+					&["state", "protocol"],
+				)?,
+				registry,
+			)?,
 		})
 	}
 }