diff --git a/prdoc/pr_11574.prdoc b/prdoc/pr_11574.prdoc new file mode 100644 index 000000000000..cdf87f6e2b89 --- /dev/null +++ b/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: minor diff --git a/substrate/client/network/src/litep2p/shim/notification/config.rs b/substrate/client/network/src/litep2p/shim/notification/config.rs index 70e136da4ed9..9c1dda7e34c4 100644 --- a/substrate/client/network/src/litep2p/shim/notification/config.rs +++ b/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 diff --git a/substrate/client/network/src/litep2p/shim/notification/peerset.rs b/substrate/client/network/src/litep2p/shim/notification/peerset.rs index 8fdf65fb2ad6..a1d7e91d101d 100644 --- a/substrate/client/network/src/litep2p/shim/notification/peerset.rs +++ b/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, connected_peers: Arc, peerstore_handle: Arc, + metrics: NotificationMetrics, ) -> (Self, TracingUnboundedSender) { 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, diff --git a/substrate/client/network/src/litep2p/shim/notification/tests/fuzz.rs b/substrate/client/network/src/litep2p/shim/notification/tests/fuzz.rs index c183dae6fa2b..1dc5e4a53d5b 100644 --- a/substrate/client/network/src/litep2p/shim/notification/tests/fuzz.rs +++ b/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()); diff --git a/substrate/client/network/src/litep2p/shim/notification/tests/peerset.rs b/substrate/client/network/src/litep2p/shim/notification/tests/peerset.rs index b56c82f950a5..ec828677e580 100644 --- a/substrate/client/network/src/litep2p/shim/notification/tests/peerset.rs +++ b/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); diff --git a/substrate/client/network/src/service/metrics.rs b/substrate/client/network/src/service/metrics.rs index a34b7418fbad..aea3b587eb34 100644 --- a/substrate/client/network/src/service/metrics.rs +++ b/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, + + /// Number of disconnected and backed off peers. + pub peerset_num_state: GaugeVec, } 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, + )?, }) } }