Telemetry (Slog + WS) (#217)

* Merge remote-tracking branch 'origin/master' into gav-xts-dont-panic

* Update wasm.

* consensus, session and staking all panic-safe.

* Democracy doesn't panic in apply.

* Fix tests.

* Extra helper macro, council depanicked.

* Fix one test.

* Fix up all council tests. No panics!

* Council voting depanicked.

* Dispatch returns result.

* session & staking tests updated

* Fix democracy tests.

* Fix council tests.

* Fix up polkadot parachains in runtime

* Fix borked merge

* More Slicable support

Support general `Option` and array types.

* Basic storage types.

* Existential deposit for contract creation

* Basic implemnetation along with removals

* Fix tests.

* externalities builder fix.

* Tests.

* Fix up the runtime.

* Fix tests.

* Add generic `Address` type.

* Initial function integration of Address into Extrinsic.

* Fix build

* All tests compile.

* Fix (some) tests.

* Fix signing.

* Push error.

* transfer can accept Address

* Make Address generic over AccountIndex

* Fix test

* Make Council use Address for dispatch.

* Fix build

* Bend over backwards to support braindead derive.

* Repot some files.

* Fix tests.

* Fix grumbles

* Remove Default bound

* Fix build for new nightly.

* Make `apply_extrinsic` never panic, return useful Result.

* More merge hell

* Doesn't build, but might do soon

* Serde woes

* get substrate-runtime-staking compiling

* Polkadot builds again!

* Fix all build.

* Fix tests & binaries.

* Reserve some extra initial byte values of address for future format changes

* Make semantic of `ReservedBalance` clear.

* Fix panic handler.

* Integrate other balance transformations into the new model

Fix up staking tests.

* Fix runtime tests.

* Fix panic build.

* Tests for demonstrating interaction between balance types.

* Repot some runtime code

* Fix checkedblock in non-std builds

* Get rid of `DoLookup` phantom.

* Attempt to make transaction_pool work with lookups.

* Remove vscode settings

* New attempt at making transaction pool work.

* It builds again!

* --all builds

* Fix tests.

* New build.

* Test account nonce reset.

* polkadot transaction pool tests/framework.

* Initial draft (working).

* Address grumbles.

* Revert bad `map_or`

* Rebuild binaries, workaround.

* Avoid checking in vscode

* reconnecting, shared, slog

* CLI options for name and telemetry url

* ensure telemetry url imples enabled

* Avoid casting to usize early.

* Provide on-connect event for session message

* Better port

* heartbeat and some renaming

* transaction pool stuff

* minor renaming.

* report telemetry

* cleanups.

* Fix for previous cleanup

* Initial draft (working).

* Avoid checking in vscode

* reconnecting, shared, slog

* CLI options for name and telemetry url

* ensure telemetry url imples enabled

* Provide on-connect event for session message

* Better port

* heartbeat and some renaming

* transaction pool stuff

* minor renaming.

* report telemetry

* cleanups.

* Fix for previous cleanup

* Address grumble

* Final grumble and repot telemetry to substrate

* Fix comment

* Please build, travis...

polkadot/cli/Cargo.toml

@@ -12,6 +12,7 @@ log = "0.3"
 atty = "0.2"
 regex = "0.2"
 time = "0.1"
+slog = "^2"
 ansi_term = "0.10"
 lazy_static = "1.0"
 hex-literal = "0.1"
@@ -28,6 +29,7 @@ substrate-state-machine = { path = "../../substrate/state-machine" }
 substrate-rpc = { path = "../../substrate/rpc" }
 substrate-rpc-servers = { path = "../../substrate/rpc-servers" }
 substrate-network = { path = "../../substrate/network" }
+substrate-telemetry = { path = "../../substrate/telemetry" }
 polkadot-primitives = { path = "../primitives" }
 polkadot-service = { path = "../service" }
 polkadot-transaction-pool = { path = "../transaction-pool" }

polkadot/cli/src/cli.yml

@@ -67,4 +67,19 @@ args:
       value_name: CHAIN_SPEC
       help: Specify the chain specification (one of dev, local or poc-2)
       takes_value: true
+  - name:
+      long: name
+      value_name: NAME
+      help: The human-readable name for this node, as reported to the telemetry server, if enabled
+      takes_value: true
+  - telemetry:
+      short: t
+      long: telemetry
+      help: Should connect to the Polkadot telemetry server (off by default)
+      takes_value: false
+  - telemetry-url:
+      long: telemetry-url
+      value_name: TELEMETRY_URL
+      help: The URL of the telemetry server. Implies --telemetry.
+      takes_value: true
 subcommands:

polkadot/cli/src/informant.rs

@@ -39,18 +39,22 @@ pub fn start<B, E>(service: &Service<B, E>, handle: reactor::Handle)
 
 	let network = service.network();
 	let client = service.client();
+	let txpool = service.transaction_pool();
 
 	let display_notifications = interval.map_err(|e| debug!("Timer error: {:?}", e)).for_each(move |_| {
 		let sync_status = network.status();
 
 		if let Ok(best_block) = client.best_block_header() {
 			let hash = best_block.hash();
+			let num_peers = sync_status.num_peers;
 			let status = match (sync_status.sync.state, sync_status.sync.best_seen_block) {
 				(SyncState::Idle, _) => "Idle".into(),
 				(SyncState::Downloading, None) => "Syncing".into(),
 				(SyncState::Downloading, Some(n)) => format!("Syncing, target=#{}", n),
 			};
-			info!(target: "polkadot", "{} ({} peers), best: #{} ({})", status, sync_status.num_peers, best_block.number, hash)
+			let txpool_status = txpool.light_status();
+			info!(target: "polkadot", "{} ({} peers), best: #{} ({})", status, sync_status.num_peers, best_block.number, hash);
+			telemetry!("system.interval"; "status" => status, "peers" => num_peers, "height" => best_block.number, "best" => ?hash, "txcount" => txpool_status.transaction_count);
 		} else {
 			warn!("Error getting best block information");
 		}
@@ -60,10 +64,18 @@ pub fn start<B, E>(service: &Service<B, E>, handle: reactor::Handle)
 	let client = service.client();
 	let display_block_import = client.import_notification_stream().for_each(|n| {
 		info!(target: "polkadot", "Imported #{} ({})", n.header.number, n.hash);
+		telemetry!("block.import"; "height" => n.header.number, "best" => ?n.hash);
 		Ok(())
 	});
 
+	let txpool = service.transaction_pool();
+	let display_txpool_import = txpool.import_notification_stream().for_each(move |_| {
+		let status = txpool.light_status();
+		telemetry!("txpool.import"; "mem_usage" => status.mem_usage, "count" => status.transaction_count, "sender" => status.senders);
+		Ok(())
+	});
 	handle.spawn(display_notifications);
 	handle.spawn(display_block_import);
+	handle.spawn(display_txpool_import);
 }
 

polkadot/cli/src/lib.rs

@@ -39,6 +39,10 @@ extern crate substrate_rpc;
 extern crate substrate_rpc_servers as rpc;
 extern crate polkadot_primitives;
 extern crate polkadot_service as service;
+#[macro_use]
+extern crate slog;	// needed until we can reexport `slog_info` from `substrate_telemetry`
+#[macro_use]
+extern crate substrate_telemetry;
 extern crate polkadot_transaction_pool as txpool;
 
 #[macro_use]
@@ -57,11 +61,14 @@ use std::io;
 use std::net::SocketAddr;
 use std::path::{Path, PathBuf};
 use polkadot_primitives::Block;
+use substrate_telemetry::{init_telemetry, TelemetryConfig};
 
 use futures::sync::mpsc;
 use futures::{Sink, Future, Stream};
 use tokio_core::reactor;
-use service::ChainSpec;
+use service::{OptionChainSpec, ChainSpec};
+
+const DEFAULT_TELEMETRY_URL: &str = "wss://telemetry.polkadot.io:443";
 
 struct Configuration(service::Configuration);
 
@@ -113,8 +120,36 @@ pub fn run<I, T>(args: I) -> error::Result<()> where
 	init_logger(log_pattern);
 	fdlimit::raise_fd_limit();
 
+	info!("Parity ·:· Polkadot");
+	info!("  version {}", crate_version!());
+	info!("  by Parity Technologies, 2017, 2018");
+
 	let mut config = service::Configuration::default();
 
+	if let Some(name) = matches.value_of("name") {
+		config.name = name.into();
+		info!("Node name: {}", config.name);
+	}
+
+	let _guard = if matches.is_present("telemetry") || matches.value_of("telemetry-url").is_some() {
+		let name = config.name.clone();
+		let chain = config.chain_spec.clone();
+		Some(init_telemetry(TelemetryConfig {
+			url: matches.value_of("telemetry-url").unwrap_or(DEFAULT_TELEMETRY_URL).into(),
+			on_connect: Box::new(move || {
+				telemetry!("system.connected";
+					"name" => name.clone(),
+					"implementation" => "parity-polkadot",
+					"version" => crate_version!(),
+					"config" => "",
+					"chain" => <&'static str>::from(chain)
+				);
+			}),
+		}))
+	} else {
+		None
+	};
+
 	let base_path = matches.value_of("base-path")
 		.map(|x| Path::new(x).to_owned())
 		.unwrap_or_else(default_base_path);
@@ -129,28 +164,24 @@ pub fn run<I, T>(args: I) -> error::Result<()> where
 
 	let mut role = service::Role::FULL;
 	if matches.is_present("collator") {
-		info!("Starting collator.");
+		info!("Starting collator");
 		role = service::Role::COLLATOR;
 	} else if matches.is_present("validator") {
-		info!("Starting validator.");
+		info!("Starting validator");
 		role = service::Role::VALIDATOR;
 	} else if matches.is_present("light") {
-		info!("Starting light.");
+		info!("Starting (light)");
 		role = service::Role::LIGHT;
+	} else {
+		info!("Starting (heavy)");
 	}
 
 	match matches.value_of("chain") {
-		Some("dev") => config.chain_spec = ChainSpec::Development,
-		Some("local") => config.chain_spec = ChainSpec::LocalTestnet,
-		Some("poc-2") => config.chain_spec = ChainSpec::PoC2Testnet,
 		None => (),
-		Some(unknown) => panic!("Invalid chain name: {}", unknown),
+		Some(n) => config.chain_spec = OptionChainSpec::from(n).inner()
+			.unwrap_or_else(|| panic!("Invalid chain name: {}", n)),
 	}
-	info!("Chain specification: {}", match config.chain_spec {
+	info!("Chain specification: {}", config.chain_spec);
-		ChainSpec::Development => "Development",
-		ChainSpec::LocalTestnet => "Local Testnet",
-		ChainSpec::PoC2Testnet => "PoC-2 Testnet",
-	});
 
 	config.roles = role;
 	{

polkadot/runtime/src/lib.rs

@@ -41,7 +41,6 @@ extern crate hex_literal;
 #[cfg(test)]
 extern crate substrate_serializer;
 
-#[cfg_attr(feature = "std", macro_use)]
 extern crate substrate_primitives;
 
 #[macro_use]

polkadot/service/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "polkadot-service"
-version = "0.1.0"
+version = "0.2.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 
 [dependencies]
@@ -9,7 +9,10 @@ futures = "0.1.17"
 parking_lot = "0.4"
 tokio-timer = "0.1.2"
 error-chain = "0.11"
+lazy_static = "1.0"
 log = "0.3"
+slog = "^2"
+clap = "2.27"
 tokio-core = "0.1.12"
 exit-future = "0.1"
 ed25519 = { path = "../../substrate/ed25519" }
@@ -28,3 +31,4 @@ substrate-client-db = { path = "../../substrate/client/db" }
 substrate-codec = { path = "../../substrate/codec" }
 substrate-executor = { path = "../../substrate/executor" }
 substrate-state-machine = { path = "../../substrate/state-machine" }
+substrate-telemetry = { path = "../../substrate/telemetry" }

polkadot/service/src/config.rs

@@ -22,7 +22,7 @@ pub use network::NetworkConfiguration;
 
 /// The chain specification (this should eventually be replaced by a more general JSON-based chain
 /// specification).
-#[derive(Clone, Copy)]
+#[derive(Clone, Copy, Debug)]
 pub enum ChainSpec {
 	/// Whatever the current runtime is, with just Alice as an auth.
 	Development,
@@ -32,6 +32,47 @@ pub enum ChainSpec {
 	PoC2Testnet,
 }
 
+/// Synonym for Option<ChainSpec> because we cannot `impl From<..> for Option<ChainSpec>`
+pub struct OptionChainSpec(Option<ChainSpec>);
+
+impl OptionChainSpec {
+	/// Return the inner part.
+	pub fn inner(self) -> Option<ChainSpec> {
+		self.0
+	}
+}
+
+impl<'a> From<&'a str> for OptionChainSpec {
+	fn from(s: &'a str) -> Self {
+		OptionChainSpec(Some(match s {
+			"dev" => ChainSpec::Development,
+			"local" => ChainSpec::LocalTestnet,
+			"poc-2" => ChainSpec::PoC2Testnet,
+			_ => return OptionChainSpec(None),
+		}))
+	}
+}
+
+impl From<ChainSpec> for &'static str {
+	fn from(s: ChainSpec) -> &'static str {
+		match s {
+			ChainSpec::Development => "dev",
+			ChainSpec::LocalTestnet => "local",
+			ChainSpec::PoC2Testnet => "poc-2",
+		}
+	}
+}
+
+impl ::std::fmt::Display for ChainSpec {
+	fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+		write!(f, "{}", match *self {
+			ChainSpec::Development => "Development",
+			ChainSpec::LocalTestnet => "Local Testnet",
+			ChainSpec::PoC2Testnet => "PoC-2 Testnet",
+		})
+	}
+}
+
 /// Service configuration.
 #[derive(Clone)]
 pub struct Configuration {
@@ -49,6 +90,10 @@ pub struct Configuration {
 	pub keys: Vec<String>,
 	/// Chain specification.
 	pub chain_spec: ChainSpec,
+	/// Telemetry server URL, optional - only `Some` if telemetry reporting is enabled
+	pub telemetry: Option<String>,
+	/// Node name.
+	pub name: String,
 }
 
 impl Default for Configuration {
@@ -61,6 +106,8 @@ impl Default for Configuration {
 			database_path: Default::default(),
 			keys: Default::default(),
 			chain_spec: ChainSpec::Development,
+			telemetry: Default::default(),
+			name: "Anonymous".into(),
 		}
 	}
 }

polkadot/service/src/lib.rs

@@ -19,8 +19,8 @@
 
 extern crate futures;
 extern crate ed25519;
+extern crate clap;
 extern crate exit_future;
-extern crate parking_lot;
 extern crate tokio_timer;
 extern crate polkadot_primitives;
 extern crate polkadot_runtime;
@@ -40,9 +40,13 @@ extern crate tokio_core;
 extern crate substrate_client as client;
 extern crate substrate_client_db as client_db;
 
+#[macro_use]
+extern crate substrate_telemetry;
 #[macro_use]
 extern crate error_chain;
 #[macro_use]
+extern crate slog;	// needed until we can reexport `slog_info` from `substrate_telemetry`
+#[macro_use]
 extern crate log;
 #[macro_use]
 extern crate hex_literal;
@@ -71,7 +75,7 @@ use network::ManageNetwork;
 use exit_future::Signal;
 
 pub use self::error::{ErrorKind, Error};
-pub use config::{Configuration, Role, ChainSpec};
+pub use config::{Configuration, Role, OptionChainSpec, ChainSpec};
 
 type CodeExecutor = NativeExecutor<LocalDispatch>;
 
@@ -410,7 +414,8 @@ impl<B, E> Service<B, E>
 		let (client, on_demand) = client_creator(db_settings, executor, genesis_builder)?;
 		let api = api_creator(client.clone());
 		let best_header = client.best_block_header()?;
-		info!("Starting Polkadot. Best block is #{}", best_header.number);
+		info!("Best block is #{}", best_header.number);
+		telemetry!("node.start"; "height" => best_header.number, "best" => ?best_header.hash());
 		let transaction_pool = Arc::new(TransactionPool::new(config.transaction_pool));
 		let transaction_pool_adapter = Arc::new(TransactionPoolAdapter {
 			pool: transaction_pool.clone(),
@@ -430,6 +435,7 @@ impl<B, E> Service<B, E>
 		let barrier = ::std::sync::Arc::new(Barrier::new(2));
 		on_demand.map(|on_demand| on_demand.set_service_link(Arc::downgrade(&network)));
 
+
 		let thread = {
 			let client = client.clone();
 			let network = network.clone();
@@ -445,11 +451,11 @@ impl<B, E> Service<B, E>
 				// block notifications
 				let network1 = network.clone();
 				let txpool1 = txpool.clone();
+
 				let events = client.import_notification_stream()
 					.for_each(move |notification| {
 						network1.on_block_imported(notification.hash, &notification.header);
 						prune_imported(&*api, &*txpool1, notification.hash);
-
 						Ok(())
 					});
 				core.handle().spawn(events);