fluencelabs/kademlia-exporter
1
0
Fork 0
mirror of https://github.com/fluencelabs/kademlia-exporter synced 2025-04-24 22:02:13 +00:00
Code Issues Projects Releases Wiki Activity

libp2p 0.19 + fluence fork

Browse Source
This commit is contained in:
folex 2020-05-20 17:38:19 +03:00
parent 57301a8bcf
commit 2e06f5a325
4 changed files with 647 additions and 727 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1180
Cargo.lock generated
View File

File diff suppressed because it is too large Load Diff

12
Cargo.toml
View File

@ -9,11 +9,12 @@ edition = "2018"
[dependencies]
async-std = "1.0"
cidr = "0.1"
csv = "1"
csv = "1.1.3"
env_logger = "0.7.1"
futures = "0.3.1"
libp2p = { git = "https://github.com/mxinden/rust-libp2p", branch = "disjoint-paths-2" }
libp2p-kad = { git = "https://github.com/mxinden/rust-libp2p", branch = "disjoint-paths-2" }
futures = "0.3.5"
libp2p = { git = "https://github.com/fluencelabs/rust-libp2p", branch = "trust_graph_libp2p_git_fix" }
libp2p-kad = { git = "https://github.com/fluencelabs/rust-libp2p", branch = "trust_graph_libp2p_git_fix" }
trust-graph = { git = "https://github.com/fluencelabs/fluence", branch = "trust_graph_libp2p_git_fix" }
log = "0.4.1"
prometheus = "0.7"
void = "1.0.2"
@ -23,8 +24,11 @@ ctrlc = "3"
structopt = "0.3"
futures-timer = "3"
maxminddb = "*"
serde = "1.0.110"
# [patch."https://github.com/mxinden/rust-libp2p"]
# libp2p = { path = "/home/mxinden/code/github.com/libp2p/rust-libp2p" }
# libp2p-kad = { path = "/home/mxinden/code/github.com/libp2p/rust-libp2p/protocols/kad" }
#[patch.crates-io]
#serde = { version = "1.0.110" }

133
src/exporter.rs
View File

@ -4,7 +4,7 @@ use futures::prelude::*;
use futures_timer::Delay;
use libp2p::{
identify::IdentifyEvent,
kad::{GetClosestPeersOk, KademliaEvent},
kad::{GetClosestPeersOk, KademliaEvent, QueryResult},
multiaddr::{Multiaddr, Protocol},
ping::{PingEvent, PingSuccess},
PeerId,
@ -74,8 +74,10 @@ impl Exporter {
})
.collect();
let nodes_to_probe_periodically =
dhts.into_iter().map(|(name, _, _)| (name, vec![])).collect();
let nodes_to_probe_periodically = dhts
.into_iter()
.map(|(name, _, _)| (name, vec![]))
.collect();
Ok(Exporter {
clients,
@ -163,68 +165,70 @@ impl Exporter {
}
},
client::Event::Kademlia(event) => match *event {
KademliaEvent::BootstrapResult(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "bootstrap"])
.inc();
}
KademliaEvent::GetClosestPeersResult(res) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "get_closest_peers"])
.inc();
KademliaEvent::QueryResult { result, .. } => match result {
QueryResult::Bootstrap(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "bootstrap"])
.inc();
}
QueryResult::GetClosestPeers(res) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "get_closest_peers"])
.inc();
// Record lookup latency.
let result_label = if res.is_ok() { "ok" } else { "error" };
let peer_id = PeerId::from_bytes(match res {
Ok(GetClosestPeersOk { key, .. }) => key,
Err(err) => err.into_key(),
})
.unwrap();
let duration =
Instant::now() - self.in_flight_lookups.remove(&peer_id).unwrap();
self.metrics
.random_node_lookup_duration
.with_label_values(&[&name, result_label])
.observe(duration.as_secs_f64());
}
KademliaEvent::GetProvidersResult(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "get_providers"])
.inc();
}
KademliaEvent::StartProvidingResult(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "start_providing"])
.inc();
}
KademliaEvent::RepublishProviderResult(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "republish_provider"])
.inc();
}
KademliaEvent::GetRecordResult(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "get_record"])
.inc();
}
KademliaEvent::PutRecordResult(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "put_record"])
.inc();
}
KademliaEvent::RepublishRecordResult(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "republish_record"])
.inc();
}
// Record lookup latency.
let result_label = if res.is_ok() { "ok" } else { "error" };
let peer_id = PeerId::from_bytes(match res {
Ok(GetClosestPeersOk { key, .. }) => key,
Err(err) => err.into_key(),
})
.unwrap();
let duration =
Instant::now() - self.in_flight_lookups.remove(&peer_id).unwrap();
self.metrics
.random_node_lookup_duration
.with_label_values(&[&name, result_label])
.observe(duration.as_secs_f64());
}
QueryResult::GetProviders(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "get_providers"])
.inc();
}
QueryResult::StartProviding(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "start_providing"])
.inc();
}
QueryResult::RepublishProvider(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "republish_provider"])
.inc();
}
QueryResult::GetRecord(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "get_record"])
.inc();
}
QueryResult::PutRecord(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "put_record"])
.inc();
}
QueryResult::RepublishRecord(_) => {
self.metrics
.event_counter
.with_label_values(&[&name, "kad", "republish_record"])
.inc();
}
},
// Note: Do not interpret Discovered event as a proof of a node
// being online.
KademliaEvent::Discovered { .. } => {
@ -320,7 +324,6 @@ impl Exporter {
None
}
}
impl Future for Exporter {
type Output = ();
fn poll(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {

49
src/exporter/client.rs
View File

@ -1,9 +1,9 @@
use futures::prelude::*;
use libp2p::identity::PublicKey;
use libp2p::{
core::{
self, connection::ConnectionLimit, either::EitherError, either::EitherOutput,
multiaddr::Protocol, muxing::StreamMuxerBox, transport::boxed::Boxed, transport::Transport,
upgrade, Multiaddr,
self, either::EitherError, either::EitherOutput, multiaddr::Protocol,
muxing::StreamMuxerBox, transport::boxed::Boxed, transport::Transport, upgrade, Multiaddr,
},
dns,
identify::{Identify, IdentifyEvent},
@ -12,7 +12,10 @@ use libp2p::{
mplex, noise,
ping::{Ping, PingConfig, PingEvent},
secio,
swarm::{NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters, SwarmBuilder},
swarm::{
DialError, NetworkBehaviourAction, NetworkBehaviourEventProcess, PollParameters,
SwarmBuilder,
},
tcp, yamux, InboundUpgradeExt, NetworkBehaviour, OutboundUpgradeExt, PeerId, Swarm,
};
use std::{
@ -23,6 +26,7 @@ use std::{
time::Duration,
usize,
};
use trust_graph::TrustGraph;
mod global_only;
@ -32,13 +36,16 @@ pub struct Client {
}
impl Client {
pub fn new(mut bootnode: Multiaddr, use_disjoint_paths: bool) -> Result<Client, Box<dyn Error>> {
pub fn new(
mut bootnode: Multiaddr,
use_disjoint_paths: bool,
) -> Result<Client, Box<dyn Error>> {
// Create a random key for ourselves.
let local_key = Keypair::generate_ed25519();
let local_peer_id = PeerId::from(local_key.public());
let behaviour = MyBehaviour::new(local_key.clone(), use_disjoint_paths)?;
let transport = build_transport(local_key);
let transport = build_transport(local_key.clone());
let mut swarm = SwarmBuilder::new(transport, behaviour, local_peer_id)
.incoming_connection_limit(10)
.outgoing_connection_limit(10)
@ -53,8 +60,17 @@ impl Client {
panic!("expected peer id");
};
swarm.kademlia.add_address(&bootnode_peer_id, bootnode);
swarm.kademlia.bootstrap();
let public = match local_key.public() {
PublicKey::Ed25519(pk) => pk,
_ => unreachable!("generate_ed25519 was used"),
};
swarm
.kademlia
.add_address(&bootnode_peer_id, bootnode, public);
if let Err(err) = swarm.kademlia.bootstrap() {
log::error!("Bootstrap failed: {}", err);
}
Ok(Client {
swarm,
@ -66,7 +82,7 @@ impl Client {
self.swarm.kademlia.get_closest_peers(peer_id);
}
pub fn dial(&mut self, peer_id: &PeerId) -> Result<bool, ConnectionLimit> {
pub fn dial(&mut self, peer_id: &PeerId) -> Result<(), DialError> {
Swarm::dial(&mut self.swarm, peer_id)
}
}
@ -125,9 +141,16 @@ impl MyBehaviour {
// kind: PermissionDenied, error: "len > max" })`
kademlia_config.set_max_packet_size(8000);
if use_disjoint_paths {
kademlia_config.use_disjoint_path_queries();
log::warn!("Disjoint paths aren't supported yet");
// kademlia_config.use_disjoint_path_queries();
}
let kademlia = Kademlia::with_config(local_peer_id, store, kademlia_config);
let trust = TrustGraph::default();
let kp = match &local_key {
Keypair::Ed25519(kp) => kp,
_ => unreachable!("only ed25519 is supported"),
};
let kademlia =
Kademlia::with_config(kp.clone(), local_peer_id, store, kademlia_config, trust);
let ping = Ping::new(PingConfig::new().with_keep_alive(true));
@ -184,7 +207,9 @@ fn build_transport(keypair: Keypair) -> Boxed<(PeerId, StreamMuxerBox), impl Err
let global_only_tcp = global_only::GlobalIpOnly::new(tcp);
let transport = dns::DnsConfig::new(global_only_tcp).unwrap();
let noise_keypair = noise::Keypair::new().into_authentic(&keypair).unwrap();
let noise_keypair = noise::Keypair::<noise::X25519>::new()
.into_authentic(&keypair)
.unwrap();
let noise_config = noise::NoiseConfig::ix(noise_keypair);
let secio_config = secio::SecioConfig::new(keypair).max_frame_len(1024 * 1024);