[cmd] PeerWatcher: A tools for find and monitor peer on peer to peer …

…network. (starcoinorg#2409)

Cargo.toml

@@ -89,7 +89,8 @@ members = [
     "dataformat-generator",
     "cmd/indexer",
     "cmd/resource-exporter",
-    "cmd/merkle-generator"
+    "cmd/merkle-generator",
+    "cmd/peer-watcher"
 ]
 
 default-members = [
@@ -181,7 +182,8 @@ default-members = [
     "dataformat-generator",
     "cmd/indexer",
     "cmd/resource-exporter",
-    "cmd/merkle-generator"
+    "cmd/merkle-generator",
+    "cmd/peer-watcher"
 ]
 
 [profile.dev]

cmd/peer-watcher/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "starcoin-peer-watcher"
+version = "1.0.0-beta.4"
+authors = ["Starcoin Core Dev <[email protected]>"]
+license = "Apache-2.0"
+publish = false
+edition = "2018"
+
+[[bin]]
+name = "starcoin_peer_watcher"
+path = "src/main.rs"
+
+[dependencies]
+anyhow = "1.0.40"
+structopt = "0.3.21"
+async-std = "1.9"
+futures = "0.3.12"
+starcoin-crypto = { path = "../../commons/crypto"}
+starcoin-config = { path = "../../config"}
+starcoin-genesis = { path = "../../genesis"}
+starcoin-types = { path = "../../types"}
+starcoin-logger = { path = "../../commons/logger" }
+network-p2p-types = {path = "../../network-p2p/types"}
+network-p2p = {path = "../../network-p2p"}
+starcoin-network = {path = "../../network"}
+starcoin-storage = {path = "../../storage"}
+
+

cmd/peer-watcher/README.md

@@ -0,0 +1,11 @@
+## PeerWatcher
+
+A tools for find and monitor peer on peer to peer network.
+
+### Usage
+
+
+
+```bash
+$ .target/release/starcoin_peer_watcher -n barnard -d /tmp/starcoin --max-outgoing-peers 1024 --max-incoming-peers 1024
+```

cmd/peer-watcher/src/lib.rs

@@ -0,0 +1,26 @@
+// Copyright (c) The Starcoin Core Contributors
+// SPDX-License-Identifier: Apache-2.0
+
+use anyhow::Result;
+use network_p2p::NetworkWorker;
+use starcoin_config::{ChainNetwork, NetworkConfig};
+use starcoin_network::{build_network_worker, NotificationMessage};
+use starcoin_storage::storage::StorageInstance;
+use starcoin_storage::Storage;
+use starcoin_types::peer_info::PeerInfo;
+use std::sync::Arc;
+
+pub fn build_lighting_network(
+    net: &ChainNetwork,
+    network_config: &NetworkConfig,
+) -> Result<(PeerInfo, NetworkWorker)> {
+    let genesis = starcoin_genesis::Genesis::load(net)?;
+    let storage = Arc::new(Storage::new(StorageInstance::new_cache_instance())?);
+    let chain_info = genesis.execute_genesis_block(net, storage)?;
+    build_network_worker(
+        network_config,
+        chain_info,
+        NotificationMessage::protocols(),
+        None,
+    )
+}

cmd/peer-watcher/src/main.rs

@@ -0,0 +1,47 @@
+// Copyright (c) The Starcoin Core Contributors
+// SPDX-License-Identifier: Apache-2.0
+
+use futures::StreamExt;
+use network_p2p::Event;
+use starcoin_config::{NodeConfig, StarcoinOpt};
+use starcoin_peer_watcher::build_lighting_network;
+use starcoin_types::peer_info::PeerInfo;
+use structopt::StructOpt;
+
+/// A lighting node, connect to peer to peer network, and monitor peers.
+fn main() {
+    let _logger = starcoin_logger::init();
+    let opt: StarcoinOpt = StarcoinOpt::from_args();
+    let config = NodeConfig::load_with_opt(&opt).unwrap();
+    let (peer_info, worker) = build_lighting_network(config.net(), &config.network).unwrap();
+    println!("Self peer_info: {:?}", peer_info);
+    let service = worker.service().clone();
+    async_std::task::spawn(worker);
+    let stream = service.event_stream("peer_watcher");
+    futures::executor::block_on(async move {
+        stream
+            .filter_map(|event| async move {
+                match event {
+                    Event::NotificationStreamOpened {
+                        remote,
+                        protocol: _,
+                        info,
+                        notif_protocols,
+                        rpc_protocols,
+                    } => Some(PeerInfo::new(
+                        remote.into(),
+                        *info,
+                        notif_protocols,
+                        rpc_protocols,
+                    )),
+                    _ => None,
+                }
+            })
+            .for_each(|peer| async move {
+                //TODO save peer info to database or post to a webhook
+                // get peer's more info from network state, such as ip address, version etc.
+                println!("Find peer: {:?}", peer)
+            })
+            .await;
+    });
+}

network/api/src/tests.rs

@@ -65,19 +65,26 @@ fn test_peer_selector() {
 #[test]
 fn test_better_peer() {
     let mut peers = Vec::new();
-    let random_peer = PeerInfo::random();
     for _ in 0..20 {
         peers.push(PeerInfo::random());
     }
 
-    let peer_selector = PeerSelector::new(peers, PeerStrategy::default());
-    let better_selector = peer_selector.betters(random_peer.total_difficulty(), 10);
-    assert!(better_selector.is_some());
-
-    let better_selector = better_selector.unwrap();
-    assert!(!better_selector.contains(&random_peer));
+    let first_peer = peers.get(0).cloned().expect("first peer must exist.");
 
-    better_selector.iter().for_each(|better_peer| {
-        assert!(better_peer.total_difficulty() >= random_peer.total_difficulty());
-    });
+    let peer_selector = PeerSelector::new(peers, PeerStrategy::default());
+    let better_selector = peer_selector.betters(first_peer.total_difficulty(), 10);
+    if let Some(better_selector) = better_selector {
+        assert!(!better_selector.contains(&first_peer));
+        better_selector.iter().for_each(|better_peer| {
+            assert!(better_peer.total_difficulty() >= first_peer.total_difficulty());
+        });
+    } else {
+        peer_selector
+            .bests(0.into())
+            .unwrap()
+            .iter()
+            .for_each(|peer| {
+                assert!(peer.total_difficulty() <= first_peer.total_difficulty());
+            })
+    }
 }

network/src/service.rs

@@ -54,7 +54,7 @@ impl NetworkActorService {
         H: PeerMessageHandler + 'static,
     {
         let (self_info, worker) = build_network_worker(
-            config.as_ref(),
+            &config.network,
             chain_info,
             NotificationMessage::protocols(),
             rpc,

network/src/worker.rs

@@ -13,7 +13,7 @@ use network_p2p::{
 };
 use network_p2p_types::{is_memory_addr, ProtocolRequest};
 use prometheus::default_registry;
-use starcoin_config::NodeConfig;
+use starcoin_config::NetworkConfig;
 use starcoin_network_rpc::NetworkRpcService;
 use starcoin_service_registry::ServiceRef;
 use starcoin_types::peer_info::RpcInfo;
@@ -26,14 +26,14 @@ const REQUEST_BUFFER_SIZE: usize = 128;
 pub const RPC_PROTOCOL_PREFIX: &str = "/starcoin/rpc/";
 
 pub fn build_network_worker(
-    node_config: &NodeConfig,
+    network_config: &NetworkConfig,
     chain_info: ChainInfo,
     protocols: Vec<Cow<'static, str>>,
     rpc_service: Option<(RpcInfo, ServiceRef<NetworkRpcService>)>,
 ) -> Result<(PeerInfo, NetworkWorker)> {
-    let node_name = node_config.node_name();
-    let discover_local = node_config.network.discover_local();
-    let transport_config = if is_memory_addr(&node_config.network.listen()) {
+    let node_name = network_config.node_name();
+    let discover_local = network_config.discover_local();
+    let transport_config = if is_memory_addr(&network_config.listen()) {
         TransportConfig::MemoryOnly
     } else {
         TransportConfig::Normal {
@@ -74,11 +74,11 @@ pub fn build_network_worker(
         None => vec![],
     };
     let allow_non_globals_in_dht = discover_local;
-    let boot_nodes = node_config.network.seeds();
+    let boot_nodes = network_config.seeds();
 
     info!("Final bootstrap seeds: {:?}", boot_nodes);
     let self_info = PeerInfo::new(
-        node_config.network.self_peer_id(),
+        network_config.self_peer_id(),
         chain_info.clone(),
         protocols.to_vec(),
         rpc_protocols
@@ -87,17 +87,17 @@ pub fn build_network_worker(
             .collect(),
     );
     let config = NetworkConfiguration {
-        listen_addresses: vec![node_config.network.listen()],
+        listen_addresses: vec![network_config.listen()],
         boot_nodes,
         node_key: {
             let secret = identity::ed25519::SecretKey::from_bytes(
-                &mut node_config.network.network_keypair().0.to_bytes(),
+                &mut network_config.network_keypair().0.to_bytes(),
             )
             .expect("decode network node key should success.");
             NodeKeyConfig::Ed25519(Secret::Input(secret))
         },
-        in_peers: node_config.network.max_incoming_peers(),
-        out_peers: node_config.network.max_outgoing_peers(),
+        in_peers: network_config.max_incoming_peers(),
+        out_peers: network_config.max_outgoing_peers(),
         notifications_protocols: protocols,
         request_response_protocols: rpc_protocols,
         transport: transport_config,

network/tests/network_service_test.rs

@@ -62,7 +62,8 @@ fn build_test_network_services(num: usize) -> Vec<NetworkComponent> {
         let mut protocols = NotificationMessage::protocols();
         protocols.push(TEST_NOTIF_PROTOCOL_NAME.into());
         let (_peer_info, worker) =
-            build_network_worker(&node_config, chain_info.clone(), protocols, None).unwrap();
+            build_network_worker(&node_config.network, chain_info.clone(), protocols, None)
+                .unwrap();
         let network_service = worker.service().clone();
         async_std::task::spawn(worker);
         result.push({