fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-07-31 17:01:58 +00:00
Code Issues Projects Releases Wiki Activity

misc/metrics: Track # connected nodes supporting specific protocol (#2734)

Browse Source 
* misc/metrics: Explicitly delegate event recording to each recorder

This allows delegating a single event to multiple `Recorder`s. That enables e.g. the
`identify::Metrics` `Recorder` to act both on `IdentifyEvent` and `SwarmEvent`. The latter enables
it to garbage collect per peer data on disconnects.

* protocols/dcutr: Expose PROTOCOL_NAME

* protocols/identify: Expose PROTOCOL_NAME and PUSH_PROTOCOL_NAME

* protocols/ping: Expose PROTOCOL_NAME

* protocols/relay: Expose HOP_PROTOCOL_NAME and STOP_PROTOCOL_NAME

* misc/metrics: Track # connected nodes supporting specific protocol

An example metric exposed with this patch:

```
libp2p_identify_protocols{protocol="/ipfs/ping/1.0.0"} 10
```

This implies that 10 of the currently connected nodes support the ping protocol.
This commit is contained in:
Max Inden
2022-07-15 09:16:03 +02:00
committed by GitHub
parent 7c8a97739f
commit d4f8ec2d48
21 changed files with 259 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
misc/metrics/src/dcutr.rs
View File

@@ -77,10 +77,9 @@ impl From<&libp2p_dcutr::behaviour::Event> for EventType {
}
}
impl super::Recorder<libp2p_dcutr::behaviour::Event> for super::Metrics {
impl super::Recorder<libp2p_dcutr::behaviour::Event> for Metrics {
fn record(&self, event: &libp2p_dcutr::behaviour::Event) {
self.dcutr
.events
self.events
.get_or_create(&EventLabels {
event: event.into(),
})

4
misc/metrics/src/gossipsub.rs
View File

@@ -40,10 +40,10 @@ impl Metrics {
}
}
impl super::Recorder<libp2p_gossipsub::GossipsubEvent> for super::Metrics {
impl super::Recorder<libp2p_gossipsub::GossipsubEvent> for Metrics {
fn record(&self, event: &libp2p_gossipsub::GossipsubEvent) {
if let libp2p_gossipsub::GossipsubEvent::Message { .. } = event {
self.gossipsub.messages.inc();
self.messages.inc();
}
}
}

145
misc/metrics/src/identify.rs
View File

@@ -18,12 +18,18 @@
// FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
use libp2p_core::PeerId;
use prometheus_client::encoding::text::{EncodeMetric, Encoder};
use prometheus_client::metrics::counter::Counter;
use prometheus_client::metrics::histogram::{exponential_buckets, Histogram};
use prometheus_client::metrics::MetricType;
use prometheus_client::registry::Registry;
use std::collections::HashMap;
use std::iter;
use std::sync::{Arc, Mutex};
pub struct Metrics {
protocols: Protocols,
error: Counter,
pushed: Counter,
received: Counter,
@@ -36,6 +42,15 @@ impl Metrics {
pub fn new(registry: &mut Registry) -> Self {
let sub_registry = registry.sub_registry_with_prefix("identify");
let protocols = Protocols::default();
sub_registry.register(
"protocols",
"Number of connected nodes supporting a specific protocol, with \
\"unrecognized\" for each peer supporting one or more unrecognized \
protocols",
Box::new(protocols.clone()),
);
let error = Counter::default();
sub_registry.register(
"errors",
@@ -86,6 +101,7 @@ impl Metrics {
);
Self {
protocols,
error,
pushed,
received,
@@ -96,27 +112,136 @@ impl Metrics {
}
}
impl super::Recorder<libp2p_identify::IdentifyEvent> for super::Metrics {
impl super::Recorder<libp2p_identify::IdentifyEvent> for Metrics {
fn record(&self, event: &libp2p_identify::IdentifyEvent) {
match event {
libp2p_identify::IdentifyEvent::Error { .. } => {
self.identify.error.inc();
self.error.inc();
}
libp2p_identify::IdentifyEvent::Pushed { .. } => {
self.identify.pushed.inc();
self.pushed.inc();
}
libp2p_identify::IdentifyEvent::Received { info, .. } => {
self.identify.received.inc();
self.identify
.received_info_protocols
libp2p_identify::IdentifyEvent::Received { peer_id, info, .. } => {
{
let mut protocols: Vec<String> = info
.protocols
.iter()
.filter(|p| {
let allowed_protocols: &[&[u8]] = &[
#[cfg(feature = "dcutr")]
libp2p_dcutr::PROTOCOL_NAME,
// #[cfg(feature = "gossipsub")]
// #[cfg(not(target_os = "unknown"))]
// TODO: Add Gossipsub protocol name
libp2p_identify::PROTOCOL_NAME,
libp2p_identify::PUSH_PROTOCOL_NAME,
#[cfg(feature = "kad")]
libp2p_kad::protocol::DEFAULT_PROTO_NAME,
#[cfg(feature = "ping")]
libp2p_ping::PROTOCOL_NAME,
#[cfg(feature = "relay")]
libp2p_relay::v2::STOP_PROTOCOL_NAME,
#[cfg(feature = "relay")]
libp2p_relay::v2::HOP_PROTOCOL_NAME,
];
allowed_protocols.contains(&p.as_bytes())
})
.cloned()
.collect();
// Signal via an additional label value that one or more
// protocols of the remote peer have not been recognized.
if protocols.len() < info.protocols.len() {
protocols.push("unrecognized".to_string());
}
protocols.sort_unstable();
protocols.dedup();
self.protocols.add(*peer_id, protocols);
}
self.received.inc();
self.received_info_protocols
.observe(info.protocols.len() as f64);
self.identify
.received_info_listen_addrs
self.received_info_listen_addrs
.observe(info.listen_addrs.len() as f64);
}
libp2p_identify::IdentifyEvent::Sent { .. } => {
self.identify.sent.inc();
self.sent.inc();
}
}
}
}
impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleErr>> for Metrics {
fn record(&self, event: &libp2p_swarm::SwarmEvent<TBvEv, THandleErr>) {
if let libp2p_swarm::SwarmEvent::ConnectionClosed {
peer_id,
num_established,
..
} = event
{
if *num_established == 0 {
self.protocols.remove(*peer_id)
}
}
}
}
#[derive(Default, Clone)]
struct Protocols {
peers: Arc<Mutex<HashMap<PeerId, Vec<String>>>>,
}
impl Protocols {
fn add(&self, peer: PeerId, protocols: Vec<String>) {
self.peers
.lock()
.expect("Lock not to be poisoned")
.insert(peer, protocols);
}
fn remove(&self, peer: PeerId) {
self.peers
.lock()
.expect("Lock not to be poisoned")
.remove(&peer);
}
}
impl EncodeMetric for Protocols {
fn encode(&self, mut encoder: Encoder) -> Result<(), std::io::Error> {
let count_by_protocol = self
.peers
.lock()
.expect("Lock not to be poisoned")
.iter()
.fold(
HashMap::<String, u64>::default(),
|mut acc, (_, protocols)| {
for protocol in protocols {
let count = acc.entry(protocol.to_string()).or_default();
*count = *count + 1;
}
acc
},
);
for (protocol, count) in count_by_protocol {
encoder
.with_label_set(&("protocol", protocol))
.no_suffix()?
.no_bucket()?
.encode_value(count)?
.no_exemplar()?;
}
Ok(())
}
fn metric_type(&self) -> MetricType {
MetricType::Gauge
}
}

40
misc/metrics/src/kad.rs
View File

@@ -159,25 +159,21 @@ impl Metrics {
}
}
impl super::Recorder<libp2p_kad::KademliaEvent> for super::Metrics {
impl super::Recorder<libp2p_kad::KademliaEvent> for Metrics {
fn record(&self, event: &libp2p_kad::KademliaEvent) {
match event {
libp2p_kad::KademliaEvent::OutboundQueryCompleted { result, stats, .. } => {
self.kad
.query_result_num_requests
self.query_result_num_requests
.get_or_create(&result.into())
.observe(stats.num_requests().into());
self.kad
.query_result_num_success
self.query_result_num_success
.get_or_create(&result.into())
.observe(stats.num_successes().into());
self.kad
.query_result_num_failure
self.query_result_num_failure
.get_or_create(&result.into())
.observe(stats.num_failures().into());
if let Some(duration) = stats.duration() {
self.kad
.query_result_duration
self.query_result_duration
.get_or_create(&result.into())
.observe(duration.as_secs_f64());
}
@@ -185,36 +181,30 @@ impl super::Recorder<libp2p_kad::KademliaEvent> for super::Metrics {
match result {
libp2p_kad::QueryResult::GetRecord(result) => match result {
Ok(ok) => self
.kad
.query_result_get_record_ok
.observe(ok.records.len() as f64),
Err(error) => {
self.kad
.query_result_get_record_error
self.query_result_get_record_error
.get_or_create(&error.into())
.inc();
}
},
libp2p_kad::QueryResult::GetClosestPeers(result) => match result {
Ok(ok) => self
.kad
.query_result_get_closest_peers_ok
.observe(ok.peers.len() as f64),
Err(error) => {
self.kad
.query_result_get_closest_peers_error
self.query_result_get_closest_peers_error
.get_or_create(&error.into())
.inc();
}
},
libp2p_kad::QueryResult::GetProviders(result) => match result {
Ok(ok) => self
.kad
.query_result_get_providers_ok
.observe(ok.providers.len() as f64),
Err(error) => {
self.kad
.query_result_get_providers_error
self.query_result_get_providers_error
.get_or_create(&error.into())
.inc();
}
@@ -230,16 +220,14 @@ impl super::Recorder<libp2p_kad::KademliaEvent> for super::Metrics {
} => {
let bucket = low.ilog2().unwrap_or(0);
if *is_new_peer {
self.kad
.routing_updated
self.routing_updated
.get_or_create(&RoutingUpdated {
action: RoutingAction::Added,
bucket,
})
.inc();
} else {
self.kad
.routing_updated
self.routing_updated
.get_or_create(&RoutingUpdated {
action: RoutingAction::Updated,
bucket,
@@ -248,8 +236,7 @@ impl super::Recorder<libp2p_kad::KademliaEvent> for super::Metrics {
}
if old_peer.is_some() {
self.kad
.routing_updated
self.routing_updated
.get_or_create(&RoutingUpdated {
action: RoutingAction::Evicted,
bucket,
@@ -259,10 +246,7 @@ impl super::Recorder<libp2p_kad::KademliaEvent> for super::Metrics {
}
libp2p_kad::KademliaEvent::InboundRequest { request } => {
self.kad
.inbound_requests
.get_or_create(&request.into())
.inc();
self.inbound_requests.get_or_create(&request.into()).inc();
}
_ => {}
}

52
misc/metrics/src/lib.rs
View File

@@ -95,3 +95,55 @@ pub trait Recorder<Event> {
/// Record the given event.
fn record(&self, event: &Event);
}
#[cfg(feature = "dcutr")]
impl Recorder<libp2p_dcutr::behaviour::Event> for Metrics {
fn record(&self, event: &libp2p_dcutr::behaviour::Event) {
self.dcutr.record(event)
}
}
#[cfg(feature = "gossipsub")]
#[cfg(not(target_os = "unknown"))]
impl Recorder<libp2p_gossipsub::GossipsubEvent> for Metrics {
fn record(&self, event: &libp2p_gossipsub::GossipsubEvent) {
self.gossipsub.record(event)
}
}
#[cfg(feature = "identify")]
impl Recorder<libp2p_identify::IdentifyEvent> for Metrics {
fn record(&self, event: &libp2p_identify::IdentifyEvent) {
self.identify.record(event)
}
}
#[cfg(feature = "kad")]
impl Recorder<libp2p_kad::KademliaEvent> for Metrics {
fn record(&self, event: &libp2p_kad::KademliaEvent) {
self.kad.record(event)
}
}
#[cfg(feature = "ping")]
impl Recorder<libp2p_ping::PingEvent> for Metrics {
fn record(&self, event: &libp2p_ping::PingEvent) {
self.ping.record(event)
}
}
#[cfg(feature = "relay")]
impl Recorder<libp2p_relay::v2::relay::Event> for Metrics {
fn record(&self, event: &libp2p_relay::v2::relay::Event) {
self.relay.record(event)
}
}
impl<TBvEv, THandleErr> Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleErr>> for Metrics {
fn record(&self, event: &libp2p_swarm::SwarmEvent<TBvEv, THandleErr>) {
self.swarm.record(event);
#[cfg(feature = "identify")]
self.identify.record(event)
}
}

8
misc/metrics/src/ping.rs
View File

@@ -92,17 +92,17 @@ impl Metrics {
}
}
impl super::Recorder<libp2p_ping::PingEvent> for super::Metrics {
impl super::Recorder<libp2p_ping::PingEvent> for Metrics {
fn record(&self, event: &libp2p_ping::PingEvent) {
match &event.result {
Ok(libp2p_ping::PingSuccess::Pong) => {
self.ping.pong_received.inc();
self.pong_received.inc();
}
Ok(libp2p_ping::PingSuccess::Ping { rtt }) => {
self.ping.rtt.observe(rtt.as_secs_f64());
self.rtt.observe(rtt.as_secs_f64());
}
Err(failure) => {
self.ping.failure.get_or_create(&failure.into()).inc();
self.failure.get_or_create(&failure.into()).inc();
}
}
}

5
misc/metrics/src/relay.rs
View File

@@ -102,10 +102,9 @@ impl From<&libp2p_relay::v2::relay::Event> for EventType {
}
}
impl super::Recorder<libp2p_relay::v2::relay::Event> for super::Metrics {
impl super::Recorder<libp2p_relay::v2::relay::Event> for Metrics {
fn record(&self, event: &libp2p_relay::v2::relay::Event) {
self.relay
.events
self.events
.get_or_create(&EventLabels {
event: event.into(),
})

30
misc/metrics/src/swarm.rs
View File

@@ -138,34 +138,29 @@ impl Metrics {
}
}
impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleErr>>
for super::Metrics
{
impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleErr>> for Metrics {
fn record(&self, event: &libp2p_swarm::SwarmEvent<TBvEv, THandleErr>) {
match event {
libp2p_swarm::SwarmEvent::Behaviour(_) => {}
libp2p_swarm::SwarmEvent::ConnectionEstablished { endpoint, .. } => {
self.swarm
.connections_established
self.connections_established
.get_or_create(&ConnectionEstablishedLabels {
role: endpoint.into(),
})
.inc();
}
libp2p_swarm::SwarmEvent::ConnectionClosed { endpoint, .. } => {
self.swarm
.connections_closed
self.connections_closed
.get_or_create(&ConnectionClosedLabels {
role: endpoint.into(),
})
.inc();
}
libp2p_swarm::SwarmEvent::IncomingConnection { .. } => {
self.swarm.connections_incoming.inc();
self.connections_incoming.inc();
}
libp2p_swarm::SwarmEvent::IncomingConnectionError { error, .. } => {
self.swarm
.connections_incoming_error
self.connections_incoming_error
.get_or_create(&IncomingConnectionErrorLabels {
error: error.into(),
})
@@ -178,8 +173,7 @@ impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleE
};
let record = |error| {
self.swarm
.outgoing_connection_error
self.outgoing_connection_error
.get_or_create(&OutgoingConnectionErrorLabels { peer, error })
.inc();
};
@@ -228,22 +222,22 @@ impl<TBvEv, THandleErr> super::Recorder<libp2p_swarm::SwarmEvent<TBvEv, THandleE
};
}
libp2p_swarm::SwarmEvent::BannedPeer { .. } => {
self.swarm.connected_to_banned_peer.inc();
self.connected_to_banned_peer.inc();
}
libp2p_swarm::SwarmEvent::NewListenAddr { .. } => {
self.swarm.new_listen_addr.inc();
self.new_listen_addr.inc();
}
libp2p_swarm::SwarmEvent::ExpiredListenAddr { .. } => {
self.swarm.expired_listen_addr.inc();
self.expired_listen_addr.inc();
}
libp2p_swarm::SwarmEvent::ListenerClosed { .. } => {
self.swarm.listener_closed.inc();
self.listener_closed.inc();
}
libp2p_swarm::SwarmEvent::ListenerError { .. } => {
self.swarm.listener_error.inc();
self.listener_error.inc();
}
libp2p_swarm::SwarmEvent::Dialing(_) => {
self.swarm.dial_attempt.inc();
self.dial_attempt.inc();
}
}
}