Feature/telemetry (paritytech#28)

* Add telemetry thread

* Run telemetry

* Format code

Cargo.toml

@@ -14,6 +14,7 @@ substrate-client-db = { git = "https://github.com/chainx-org/substrate" }
 substrate-keyring = { git = "https://github.com/chainx-org/substrate" }
 substrate-state-db = { git = "https://github.com/chainx-org/substrate" }
 substrate-state-machine = { git = "https://github.com/chainx-org/substrate" }
+substrate-telemetry = { git = "https://github.com/chainx-org/substrate" }
 substrate-codec = { git = "https://github.com/chainx-org/substrate", default_features = false }
 substrate-bft = { git = "https://github.com/chainx-org/substrate", default_features = false }
 substrate-rpc-servers = { git = "https://github.com/chainx-org/substrate" }
@@ -38,8 +39,11 @@ rhododendron = "0.3"
 hex-literal = "0.1"
 exit-future = "0.1"
 futures = "0.1.17"
+ansi_term = "0.10"
+sysinfo = "0.5.7"
 tokio = "0.1.7"
 clap = "2.30.0"
+slog = "^2"
 log = "0.3"
 
 [workspace]

src/cli.rs

@@ -28,6 +28,20 @@ pub fn build_cli() -> App<'static, 'static> {
             .help("Specify p2p protocol TCP port")
             .takes_value(true),
             )
+        .arg(
+            Arg::with_name("telemetry-url")
+            .long("telemetry-url")
+            .value_name("TELEMETRY_URL")
+            .help("The URL of the telemetry server. Implies --telemetry")
+            .takes_value(true),
+            )
+        .arg(
+            Arg::with_name("telemetry")
+            .long("telemetry")
+            .value_name("TELEMETRY")
+            .help("Should connect telemetry")
+            .takes_value(false),
+            )
         .arg(
             Arg::with_name("bootnodes")
             .long("bootnodes")
@@ -73,7 +87,7 @@ pub fn parse_address(
     default: &str,
     port_param: &str,
     matches: &ArgMatches,
-    ) -> Result<SocketAddr, String> {
+) -> Result<SocketAddr, String> {
     let mut address: SocketAddr = default.parse().ok().ok_or_else(|| {
         format!("Invalid address specified for --{}.", port_param)
     })?;

src/client.rs

@@ -23,12 +23,10 @@ pub fn build_client(db_path: &str, chainspec: ChainSpec) -> Arc<TClient> {
                 pruning: state_db::PruningMode::default(),
             },
             FINALIZATION_WINDOW,
-            ).unwrap(),
-            );
+        ).unwrap(),
+    );
 
-    let executor = substrate_client::LocalCallExecutor::new(
-        backend.clone(),
-        NativeExecutor::new());
+    let executor = substrate_client::LocalCallExecutor::new(backend.clone(), NativeExecutor::new());
 
     let genesis_config = super::genesis_config::testnet_genesis(chainspec);
 
@@ -39,6 +37,6 @@ pub fn build_client(db_path: &str, chainspec: ChainSpec) -> Arc<TClient> {
             executor,
             genesis_config,
             ExecutionStrategy::NativeWhenPossible,
-            ).unwrap(),
-            )
+        ).unwrap(),
+    )
 }

src/genesis_config.rs

@@ -1,27 +1,36 @@
 // Copyright 2018 chainpool
 
-use chainx_runtime::{GenesisConfig, ConsensusConfig, CouncilConfig, DemocracyConfig,SessionConfig, StakingConfig, TimestampConfig,BalancesConfig};
+use chainx_runtime::{GenesisConfig, ConsensusConfig, CouncilConfig, DemocracyConfig,
+                     SessionConfig, StakingConfig, TimestampConfig, BalancesConfig};
 use super::cli::ChainSpec;
 use keyring::Keyring;
 use ed25519;
 
 
 pub fn testnet_genesis(chainspec: ChainSpec) -> GenesisConfig {
-    let auth1 = ed25519::Pair::from_seed(b"Alice                           ").public().into();
-    let auth2 = ed25519::Pair::from_seed(b"Bob                             ").public().into();
-    let auth3 = ed25519::Pair::from_seed(b"Gavin                           ").public().into();
-    let auth4 = ed25519::Pair::from_seed(b"Satoshi                         ").public().into();
+    let auth1 = ed25519::Pair::from_seed(b"Alice                           ")
+        .public()
+        .into();
+    let auth2 = ed25519::Pair::from_seed(b"Bob                             ")
+        .public()
+        .into();
+    let auth3 = ed25519::Pair::from_seed(b"Gavin                           ")
+        .public()
+        .into();
+    let auth4 = ed25519::Pair::from_seed(b"Satoshi                         ")
+        .public()
+        .into();
     let initial_authorities = match chainspec {
-        ChainSpec::Dev => vec![auth1,],
-        ChainSpec::Local => vec![auth1, auth2,],
+        ChainSpec::Dev => vec![auth1],
+        ChainSpec::Local => vec![auth1, auth2],
         ChainSpec::Multi => vec![auth1, auth2, auth3, auth4],
     };
     GenesisConfig {
         consensus: Some(ConsensusConfig {
             code: include_bytes!(
-                      "../runtime/wasm/target/wasm32-unknown-unknown/release/chainx_runtime.compact.wasm"
-                  ).to_vec(),
-                  authorities: initial_authorities.clone(),
+                "../runtime/wasm/target/wasm32-unknown-unknown/release/chainx_runtime.compact.wasm"
+            ).to_vec(),
+            authorities: initial_authorities.clone(),
         }),
         system: None,
         balances: Some(BalancesConfig {
@@ -32,13 +41,18 @@ pub fn testnet_genesis(chainspec: ChainSpec) -> GenesisConfig {
             creation_fee: 0,
             reclaim_rebate: 0,
             balances: vec![
-                (Keyring::Alice.to_raw_public().into(),10000),
-                (Keyring::Bob.to_raw_public().into(),10000),
-                (Keyring::Charlie.to_raw_public().into(),10000)],
+                (Keyring::Alice.to_raw_public().into(), 10000),
+                (Keyring::Bob.to_raw_public().into(), 10000),
+                (Keyring::Charlie.to_raw_public().into(), 10000),
+            ],
         }),
 
         session: Some(SessionConfig {
-            validators: initial_authorities.iter().cloned().map(Into::into).collect(),
+            validators: initial_authorities
+                .iter()
+                .cloned()
+                .map(Into::into)
+                .collect(),
             session_length: 720, // that's 1 hour per session.
         }),
         staking: Some(StakingConfig {

src/main.rs

@@ -8,6 +8,8 @@ extern crate substrate_network;
 extern crate substrate_network_libp2p;
 extern crate substrate_primitives;
 extern crate substrate_rpc_servers as rpc_server;
+#[macro_use]
+extern crate substrate_telemetry as tel;
 extern crate substrate_runtime_primitives;
 extern crate substrate_state_db as state_db;
 extern crate substrate_state_machine as state_machine;
@@ -31,15 +33,20 @@ extern crate hex_literal;
 extern crate jsonrpc_http_server;
 extern crate jsonrpc_ws_server;
 extern crate rhododendron;
+extern crate ansi_term;
+extern crate sysinfo;
 extern crate tokio;
 #[macro_use]
+extern crate slog;
+#[macro_use]
 extern crate log;
 
-mod cli;
-mod client;
 mod genesis_config;
+mod telemetry;
 mod network;
+mod client;
 mod rpc;
+mod cli;
 
 use substrate_client::BlockchainEvents;
 
@@ -74,7 +81,7 @@ fn main() {
     let port = match matches.value_of("port") {
         Some(port) => port
             .parse()
-            .map_err(|_| "Invalid p2p port value specified.")
+            .map_err(|_| "invalid p2p port value specified.")
             .unwrap(),
         None => 20222,
     };
@@ -147,7 +154,7 @@ fn main() {
             }
         };
 
-        let consensus_net = ConsensusNetwork::new(network, client.clone());
+        let consensus_net = ConsensusNetwork::new(network.clone(), client.clone());
         Some(consensus::Service::new(
             client.clone(),
             client.clone(),
@@ -162,6 +169,17 @@ fn main() {
 
     let (_rpc_http, _rpc_ws) = rpc::start(&client, &task_executor, &matches, &extrinsic_pool);
 
-    let _ = runtime.block_on(exit);
+    if matches.is_present("telemetry") {
+        let telemetry_url = match matches.value_of("telemetry_url") {
+            Some(url) => Some(url.to_owned()),
+            None => Some("http://aws.chainx.org:8888".to_owned()),
+        };
+        let _telemetry = telemetry::build_telemetry(telemetry_url, validator_mode);
+        telemetry::run_telemetry(network, client, extrinsic_pool, task_executor);
+        let _ = runtime.block_on(exit.clone());
+    } else {
+        let _ = runtime.block_on(exit);
+    }
+
     exit_send.fire();
 }

src/rpc.rs

@@ -2,8 +2,6 @@
 
 use jsonrpc_http_server::Server as HttpServer;
 use jsonrpc_ws_server::Server as WsServer;
-use chainx_rpc::chainext::ChainExt;
-use rpc_server::apis::chain::Chain;
 use chainx_pool::TransactionPool;
 use tokio::runtime::TaskExecutor;
 use chainx_api::ChainXApi;