* Revert "improve libp2p connected peer metrics (#4870)"
This reverts commit 0c3fef59b3
.
This commit is contained in:
parent
a229b52723
commit
50c423ad88
@ -68,7 +68,7 @@ use std::path::PathBuf;
|
|||||||
use std::pin::Pin;
|
use std::pin::Pin;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use sysinfo::{System, SystemExt};
|
use sysinfo::{System, SystemExt};
|
||||||
use system_health::{observe_nat, observe_system_health_bn};
|
use system_health::observe_system_health_bn;
|
||||||
use task_spawner::{Priority, TaskSpawner};
|
use task_spawner::{Priority, TaskSpawner};
|
||||||
use tokio::sync::{
|
use tokio::sync::{
|
||||||
mpsc::{Sender, UnboundedSender},
|
mpsc::{Sender, UnboundedSender},
|
||||||
@ -3965,7 +3965,13 @@ pub fn serve<T: BeaconChainTypes>(
|
|||||||
.and(warp::path::end())
|
.and(warp::path::end())
|
||||||
.then(|task_spawner: TaskSpawner<T::EthSpec>| {
|
.then(|task_spawner: TaskSpawner<T::EthSpec>| {
|
||||||
task_spawner.blocking_json_task(Priority::P1, move || {
|
task_spawner.blocking_json_task(Priority::P1, move || {
|
||||||
Ok(api_types::GenericResponse::from(observe_nat()))
|
Ok(api_types::GenericResponse::from(
|
||||||
|
lighthouse_network::metrics::NAT_OPEN
|
||||||
|
.as_ref()
|
||||||
|
.map(|v| v.get())
|
||||||
|
.unwrap_or(0)
|
||||||
|
!= 0,
|
||||||
|
))
|
||||||
})
|
})
|
||||||
});
|
});
|
||||||
|
|
||||||
|
@ -1003,13 +1003,11 @@ impl<TSpec: EthSpec> NetworkBehaviour for Discovery<TSpec> {
|
|||||||
}
|
}
|
||||||
discv5::Event::SocketUpdated(socket_addr) => {
|
discv5::Event::SocketUpdated(socket_addr) => {
|
||||||
info!(self.log, "Address updated"; "ip" => %socket_addr.ip(), "udp_port" => %socket_addr.port());
|
info!(self.log, "Address updated"; "ip" => %socket_addr.ip(), "udp_port" => %socket_addr.port());
|
||||||
// We have SOCKET_UPDATED messages. This occurs when discovery has a majority of
|
metrics::inc_counter(&metrics::ADDRESS_UPDATE_COUNT);
|
||||||
// users reporting an external port and our ENR gets updated.
|
metrics::check_nat();
|
||||||
// Which means we are able to do NAT traversal.
|
|
||||||
metrics::set_gauge_vec(&metrics::NAT_OPEN, &["discv5"], 1);
|
|
||||||
|
|
||||||
// Discv5 will have updated our local ENR. We save the updated version
|
// Discv5 will have updated our local ENR. We save the updated version
|
||||||
// to disk.
|
// to disk.
|
||||||
|
|
||||||
if (self.update_ports.tcp4 && socket_addr.is_ipv4())
|
if (self.update_ports.tcp4 && socket_addr.is_ipv4())
|
||||||
|| (self.update_ports.tcp6 && socket_addr.is_ipv6())
|
|| (self.update_ports.tcp6 && socket_addr.is_ipv6())
|
||||||
{
|
{
|
||||||
|
@ -1,17 +1,29 @@
|
|||||||
pub use lighthouse_metrics::*;
|
pub use lighthouse_metrics::*;
|
||||||
|
|
||||||
lazy_static! {
|
lazy_static! {
|
||||||
pub static ref NAT_OPEN: Result<IntGaugeVec> = try_create_int_gauge_vec(
|
pub static ref NAT_OPEN: Result<IntCounter> = try_create_int_counter(
|
||||||
"nat_open",
|
"nat_open",
|
||||||
"An estimate indicating if the local node is reachable from external nodes",
|
"An estimate indicating if the local node is exposed to the internet."
|
||||||
&["protocol"]
|
|
||||||
);
|
);
|
||||||
pub static ref ADDRESS_UPDATE_COUNT: Result<IntCounter> = try_create_int_counter(
|
pub static ref ADDRESS_UPDATE_COUNT: Result<IntCounter> = try_create_int_counter(
|
||||||
"libp2p_address_update_total",
|
"libp2p_address_update_total",
|
||||||
"Count of libp2p socked updated events (when our view of our IP address has changed)"
|
"Count of libp2p socked updated events (when our view of our IP address has changed)"
|
||||||
);
|
);
|
||||||
pub static ref PEERS_CONNECTED: Result<IntGaugeVec> =
|
pub static ref PEERS_CONNECTED: Result<IntGauge> = try_create_int_gauge(
|
||||||
try_create_int_gauge_vec("libp2p_peers", "Count of libp2p peers currently connected", &["direction", "transport"]);
|
"libp2p_peers",
|
||||||
|
"Count of libp2p peers currently connected"
|
||||||
|
);
|
||||||
|
|
||||||
|
pub static ref TCP_PEERS_CONNECTED: Result<IntGauge> = try_create_int_gauge(
|
||||||
|
"libp2p_tcp_peers",
|
||||||
|
"Count of libp2p peers currently connected via TCP"
|
||||||
|
);
|
||||||
|
|
||||||
|
pub static ref QUIC_PEERS_CONNECTED: Result<IntGauge> = try_create_int_gauge(
|
||||||
|
"libp2p_quic_peers",
|
||||||
|
"Count of libp2p peers currently connected via QUIC"
|
||||||
|
);
|
||||||
|
|
||||||
pub static ref PEER_CONNECT_EVENT_COUNT: Result<IntCounter> = try_create_int_counter(
|
pub static ref PEER_CONNECT_EVENT_COUNT: Result<IntCounter> = try_create_int_counter(
|
||||||
"libp2p_peer_connect_event_total",
|
"libp2p_peer_connect_event_total",
|
||||||
"Count of libp2p peer connect events (not the current number of connected peers)"
|
"Count of libp2p peer connect events (not the current number of connected peers)"
|
||||||
@ -20,10 +32,13 @@ lazy_static! {
|
|||||||
"libp2p_peer_disconnect_event_total",
|
"libp2p_peer_disconnect_event_total",
|
||||||
"Count of libp2p peer disconnect events"
|
"Count of libp2p peer disconnect events"
|
||||||
);
|
);
|
||||||
pub static ref DISCOVERY_BYTES: Result<IntGaugeVec> = try_create_int_gauge_vec(
|
pub static ref DISCOVERY_SENT_BYTES: Result<IntGauge> = try_create_int_gauge(
|
||||||
"discovery_bytes",
|
"discovery_sent_bytes",
|
||||||
"The number of bytes sent and received in discovery",
|
"The number of bytes sent in discovery"
|
||||||
&["direction"]
|
);
|
||||||
|
pub static ref DISCOVERY_RECV_BYTES: Result<IntGauge> = try_create_int_gauge(
|
||||||
|
"discovery_recv_bytes",
|
||||||
|
"The number of bytes received in discovery"
|
||||||
);
|
);
|
||||||
pub static ref DISCOVERY_QUEUE: Result<IntGauge> = try_create_int_gauge(
|
pub static ref DISCOVERY_QUEUE: Result<IntGauge> = try_create_int_gauge(
|
||||||
"discovery_queue_size",
|
"discovery_queue_size",
|
||||||
@ -120,6 +135,17 @@ lazy_static! {
|
|||||||
&["type"]
|
&["type"]
|
||||||
);
|
);
|
||||||
|
|
||||||
|
/*
|
||||||
|
* Inbound/Outbound peers
|
||||||
|
*/
|
||||||
|
/// The number of peers that dialed us.
|
||||||
|
pub static ref NETWORK_INBOUND_PEERS: Result<IntGauge> =
|
||||||
|
try_create_int_gauge("network_inbound_peers","The number of peers that are currently connected that have dialed us.");
|
||||||
|
|
||||||
|
/// The number of peers that we dialed us.
|
||||||
|
pub static ref NETWORK_OUTBOUND_PEERS: Result<IntGauge> =
|
||||||
|
try_create_int_gauge("network_outbound_peers","The number of peers that are currently connected that we dialed.");
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Peer Reporting
|
* Peer Reporting
|
||||||
*/
|
*/
|
||||||
@ -130,11 +156,31 @@ lazy_static! {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Checks if we consider the NAT open.
|
||||||
|
///
|
||||||
|
/// Conditions for an open NAT:
|
||||||
|
/// 1. We have 1 or more SOCKET_UPDATED messages. This occurs when discovery has a majority of
|
||||||
|
/// users reporting an external port and our ENR gets updated.
|
||||||
|
/// 2. We have 0 SOCKET_UPDATED messages (can be true if the port was correct on boot), then we
|
||||||
|
/// rely on whether we have any inbound messages. If we have no socket update messages, but
|
||||||
|
/// manage to get at least one inbound peer, we are exposed correctly.
|
||||||
|
pub fn check_nat() {
|
||||||
|
// NAT is already deemed open.
|
||||||
|
if NAT_OPEN.as_ref().map(|v| v.get()).unwrap_or(0) != 0 {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
if ADDRESS_UPDATE_COUNT.as_ref().map(|v| v.get()).unwrap_or(0) != 0
|
||||||
|
|| NETWORK_INBOUND_PEERS.as_ref().map(|v| v.get()).unwrap_or(0) != 0_i64
|
||||||
|
{
|
||||||
|
inc_counter(&NAT_OPEN);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub fn scrape_discovery_metrics() {
|
pub fn scrape_discovery_metrics() {
|
||||||
let metrics =
|
let metrics =
|
||||||
discv5::metrics::Metrics::from(discv5::Discv5::<discv5::DefaultProtocolId>::raw_metrics());
|
discv5::metrics::Metrics::from(discv5::Discv5::<discv5::DefaultProtocolId>::raw_metrics());
|
||||||
set_float_gauge(&DISCOVERY_REQS, metrics.unsolicited_requests_per_second);
|
set_float_gauge(&DISCOVERY_REQS, metrics.unsolicited_requests_per_second);
|
||||||
set_gauge(&DISCOVERY_SESSIONS, metrics.active_sessions as i64);
|
set_gauge(&DISCOVERY_SESSIONS, metrics.active_sessions as i64);
|
||||||
set_gauge_vec(&DISCOVERY_BYTES, &["inbound"], metrics.bytes_recv as i64);
|
set_gauge(&DISCOVERY_SENT_BYTES, metrics.bytes_sent as i64);
|
||||||
set_gauge_vec(&DISCOVERY_BYTES, &["outbound"], metrics.bytes_sent as i64);
|
set_gauge(&DISCOVERY_RECV_BYTES, metrics.bytes_recv as i64);
|
||||||
}
|
}
|
||||||
|
@ -726,14 +726,29 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let mut connected_peer_count = 0;
|
||||||
|
let mut inbound_connected_peers = 0;
|
||||||
|
let mut outbound_connected_peers = 0;
|
||||||
let mut clients_per_peer = HashMap::new();
|
let mut clients_per_peer = HashMap::new();
|
||||||
|
|
||||||
for (_peer, peer_info) in self.network_globals.peers.read().connected_peers() {
|
for (_peer, peer_info) in self.network_globals.peers.read().connected_peers() {
|
||||||
|
connected_peer_count += 1;
|
||||||
|
if let PeerConnectionStatus::Connected { n_in, .. } = peer_info.connection_status() {
|
||||||
|
if *n_in > 0 {
|
||||||
|
inbound_connected_peers += 1;
|
||||||
|
} else {
|
||||||
|
outbound_connected_peers += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
*clients_per_peer
|
*clients_per_peer
|
||||||
.entry(peer_info.client().kind.to_string())
|
.entry(peer_info.client().kind.to_string())
|
||||||
.or_default() += 1;
|
.or_default() += 1;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
metrics::set_gauge(&metrics::PEERS_CONNECTED, connected_peer_count);
|
||||||
|
metrics::set_gauge(&metrics::NETWORK_INBOUND_PEERS, inbound_connected_peers);
|
||||||
|
metrics::set_gauge(&metrics::NETWORK_OUTBOUND_PEERS, outbound_connected_peers);
|
||||||
|
|
||||||
for client_kind in ClientKind::iter() {
|
for client_kind in ClientKind::iter() {
|
||||||
let value = clients_per_peer.get(&client_kind.to_string()).unwrap_or(&0);
|
let value = clients_per_peer.get(&client_kind.to_string()).unwrap_or(&0);
|
||||||
metrics::set_gauge_vec(
|
metrics::set_gauge_vec(
|
||||||
@ -838,8 +853,11 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
|
|||||||
// start a ping and status timer for the peer
|
// start a ping and status timer for the peer
|
||||||
self.status_peers.insert(*peer_id);
|
self.status_peers.insert(*peer_id);
|
||||||
|
|
||||||
|
let connected_peers = self.network_globals.connected_peers() as i64;
|
||||||
|
|
||||||
// increment prometheus metrics
|
// increment prometheus metrics
|
||||||
metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT);
|
metrics::inc_counter(&metrics::PEER_CONNECT_EVENT_COUNT);
|
||||||
|
metrics::set_gauge(&metrics::PEERS_CONNECTED, connected_peers);
|
||||||
|
|
||||||
true
|
true
|
||||||
}
|
}
|
||||||
|
@ -154,8 +154,6 @@ impl<TSpec: EthSpec> NetworkBehaviour for PeerManager<TSpec> {
|
|||||||
self.on_dial_failure(peer_id);
|
self.on_dial_failure(peer_id);
|
||||||
}
|
}
|
||||||
FromSwarm::ExternalAddrConfirmed(_) => {
|
FromSwarm::ExternalAddrConfirmed(_) => {
|
||||||
// We have an external address confirmed, means we are able to do NAT traversal.
|
|
||||||
metrics::set_gauge_vec(&metrics::NAT_OPEN, &["libp2p"], 1);
|
|
||||||
// TODO: we likely want to check this against our assumed external tcp
|
// TODO: we likely want to check this against our assumed external tcp
|
||||||
// address
|
// address
|
||||||
}
|
}
|
||||||
@ -245,14 +243,14 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
|
|||||||
self.events.push(PeerManagerEvent::MetaData(peer_id));
|
self.events.push(PeerManagerEvent::MetaData(peer_id));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Check NAT if metrics are enabled
|
||||||
|
if self.network_globals.local_enr.read().udp4().is_some() {
|
||||||
|
metrics::check_nat();
|
||||||
|
}
|
||||||
|
|
||||||
// increment prometheus metrics
|
// increment prometheus metrics
|
||||||
if self.metrics_enabled {
|
if self.metrics_enabled {
|
||||||
let remote_addr = endpoint.get_remote_address();
|
let remote_addr = endpoint.get_remote_address();
|
||||||
let direction = if endpoint.is_dialer() {
|
|
||||||
"outbound"
|
|
||||||
} else {
|
|
||||||
"inbound"
|
|
||||||
};
|
|
||||||
match remote_addr.iter().find(|proto| {
|
match remote_addr.iter().find(|proto| {
|
||||||
matches!(
|
matches!(
|
||||||
proto,
|
proto,
|
||||||
@ -260,10 +258,10 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
|
|||||||
)
|
)
|
||||||
}) {
|
}) {
|
||||||
Some(multiaddr::Protocol::QuicV1) => {
|
Some(multiaddr::Protocol::QuicV1) => {
|
||||||
metrics::inc_gauge_vec(&metrics::PEERS_CONNECTED, &[direction, "quic"]);
|
metrics::inc_gauge(&metrics::QUIC_PEERS_CONNECTED);
|
||||||
}
|
}
|
||||||
Some(multiaddr::Protocol::Tcp(_)) => {
|
Some(multiaddr::Protocol::Tcp(_)) => {
|
||||||
metrics::inc_gauge_vec(&metrics::PEERS_CONNECTED, &[direction, "tcp"]);
|
metrics::inc_gauge(&metrics::TCP_PEERS_CONNECTED);
|
||||||
}
|
}
|
||||||
Some(_) => unreachable!(),
|
Some(_) => unreachable!(),
|
||||||
None => {
|
None => {
|
||||||
@ -341,12 +339,6 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
|
|||||||
let remote_addr = endpoint.get_remote_address();
|
let remote_addr = endpoint.get_remote_address();
|
||||||
// Update the prometheus metrics
|
// Update the prometheus metrics
|
||||||
if self.metrics_enabled {
|
if self.metrics_enabled {
|
||||||
let direction = if endpoint.is_dialer() {
|
|
||||||
"outbound"
|
|
||||||
} else {
|
|
||||||
"inbound"
|
|
||||||
};
|
|
||||||
|
|
||||||
match remote_addr.iter().find(|proto| {
|
match remote_addr.iter().find(|proto| {
|
||||||
matches!(
|
matches!(
|
||||||
proto,
|
proto,
|
||||||
@ -354,10 +346,10 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
|
|||||||
)
|
)
|
||||||
}) {
|
}) {
|
||||||
Some(multiaddr::Protocol::QuicV1) => {
|
Some(multiaddr::Protocol::QuicV1) => {
|
||||||
metrics::dec_gauge_vec(&metrics::PEERS_CONNECTED, &[direction, "quic"]);
|
metrics::dec_gauge(&metrics::QUIC_PEERS_CONNECTED);
|
||||||
}
|
}
|
||||||
Some(multiaddr::Protocol::Tcp(_)) => {
|
Some(multiaddr::Protocol::Tcp(_)) => {
|
||||||
metrics::dec_gauge_vec(&metrics::PEERS_CONNECTED, &[direction, "tcp"]);
|
metrics::dec_gauge(&metrics::TCP_PEERS_CONNECTED);
|
||||||
}
|
}
|
||||||
// If it's an unknown protocol we already logged when connection was established.
|
// If it's an unknown protocol we already logged when connection was established.
|
||||||
_ => {}
|
_ => {}
|
||||||
|
@ -244,16 +244,6 @@ pub fn inc_counter_vec(int_counter_vec: &Result<IntCounterVec>, name: &[&str]) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Sets the `int_counter_vec` with the given `name` to the `amount`,
|
|
||||||
/// should only be called with `ammount`s equal or above the current value
|
|
||||||
/// as per Prometheus spec, the `counter` type should only go up.
|
|
||||||
pub fn set_counter_vec_by(int_counter_vec: &Result<IntCounterVec>, name: &[&str], amount: u64) {
|
|
||||||
if let Some(counter) = get_int_counter(int_counter_vec, name) {
|
|
||||||
counter.reset();
|
|
||||||
counter.inc_by(amount);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pub fn inc_counter_vec_by(int_counter_vec: &Result<IntCounterVec>, name: &[&str], amount: u64) {
|
pub fn inc_counter_vec_by(int_counter_vec: &Result<IntCounterVec>, name: &[&str], amount: u64) {
|
||||||
if let Some(counter) = get_int_counter(int_counter_vec, name) {
|
if let Some(counter) = get_int_counter(int_counter_vec, name) {
|
||||||
counter.inc_by(amount);
|
counter.inc_by(amount);
|
||||||
|
@ -198,25 +198,6 @@ pub fn observe_system_health_vc(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Observes if NAT traversal is possible.
|
|
||||||
pub fn observe_nat() -> bool {
|
|
||||||
let discv5_nat = lighthouse_network::metrics::get_int_gauge(
|
|
||||||
&lighthouse_network::metrics::NAT_OPEN,
|
|
||||||
&["discv5"],
|
|
||||||
)
|
|
||||||
.map(|g| g.get() == 1)
|
|
||||||
.unwrap_or_default();
|
|
||||||
|
|
||||||
let libp2p_nat = lighthouse_network::metrics::get_int_gauge(
|
|
||||||
&lighthouse_network::metrics::NAT_OPEN,
|
|
||||||
&["libp2p"],
|
|
||||||
)
|
|
||||||
.map(|g| g.get() == 1)
|
|
||||||
.unwrap_or_default();
|
|
||||||
|
|
||||||
discv5_nat && libp2p_nat
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Observes the Beacon Node system health.
|
/// Observes the Beacon Node system health.
|
||||||
pub fn observe_system_health_bn<TSpec: EthSpec>(
|
pub fn observe_system_health_bn<TSpec: EthSpec>(
|
||||||
sysinfo: Arc<RwLock<System>>,
|
sysinfo: Arc<RwLock<System>>,
|
||||||
@ -242,7 +223,11 @@ pub fn observe_system_health_bn<TSpec: EthSpec>(
|
|||||||
.unwrap_or_else(|| (String::from("None"), 0, 0));
|
.unwrap_or_else(|| (String::from("None"), 0, 0));
|
||||||
|
|
||||||
// Determine if the NAT is open or not.
|
// Determine if the NAT is open or not.
|
||||||
let nat_open = observe_nat();
|
let nat_open = lighthouse_network::metrics::NAT_OPEN
|
||||||
|
.as_ref()
|
||||||
|
.map(|v| v.get())
|
||||||
|
.unwrap_or(0)
|
||||||
|
!= 0;
|
||||||
|
|
||||||
SystemHealthBN {
|
SystemHealthBN {
|
||||||
system_health,
|
system_health,
|
||||||
|
Loading…
Reference in New Issue
Block a user