Add instrumentation, with performance profiling capability (#4132)

* Implement instrumentation for performance profiling

* Add profiling to runtime functions declared in `decl_module` inc.
`on_initialize` and `on_finalize`

* Remove changes

* add docs, tidy up

* fix versions

* fix copyright date

* switch to hashmap and instant

* update example

* update example

* implement receiver for logger and make default

* fix comment

* use `if_std!` macro

* remove whitespace

* fix whitespace

* fix nits

substrate/client/cli/Cargo.toml

@@ -34,6 +34,7 @@ keyring = { package = "substrate-keyring", path = "../../primitives/keyring" }
 names = "0.11.0"
 structopt = "0.3.3"
 rpassword = "4.0.1"
+substrate-tracing = { package = "substrate-tracing", path = "../tracing" }
 
 [dev-dependencies]
 tempdir = "0.3.7"

substrate/client/cli/src/lib.rs

@@ -205,7 +205,6 @@ where
 	);
 
 	panic_handler::set(version.support_url, &full_version);
-
 	let matches = CoreParams::<CC, RP>::clap()
 		.name(version.executable_name)
 		.author(version.author)
@@ -216,7 +215,6 @@ where
 		.setting(AppSettings::SubcommandsNegateReqs)
 		.get_matches_from(args);
 	let cli_args = CoreParams::<CC, RP>::from_clap(&matches);
-
 	init_logger(cli_args.get_log_filter().as_ref().map(|v| v.as_ref()).unwrap_or(""));
 	fdlimit::raise_fd_limit();
 
@@ -846,6 +844,9 @@ where
 		config.telemetry_endpoints = Some(TelemetryEndpoints::new(cli.telemetry_endpoints));
 	}
 
+	config.tracing_targets = cli.tracing_targets.into();
+	config.tracing_receiver = cli.tracing_receiver.into();
+
 	// Imply forced authoring on --dev
 	config.force_authoring = cli.shared_params.dev || cli.force_authoring;
 
@@ -901,6 +902,8 @@ fn init_logger(pattern: &str) {
 	builder.filter(Some("ws"), log::LevelFilter::Off);
 	builder.filter(Some("hyper"), log::LevelFilter::Warn);
 	builder.filter(Some("cranelift_wasm"), log::LevelFilter::Warn);
+	// Always log the special target `substrate_tracing`, overrides global level
+	builder.filter(Some("substrate_tracing"), log::LevelFilter::Info);
 	// Enable info for others.
 	builder.filter(None, log::LevelFilter::Info);
 

substrate/client/cli/src/params.rs

@@ -259,6 +259,24 @@ pub struct TransactionPoolParams {
 	pub pool_kbytes: usize,
 }
 
+arg_enum! {
+	#[allow(missing_docs)]
+	#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+	pub enum TracingReceiver {
+		Log,
+		Telemetry,
+	}
+}
+
+impl Into<substrate_tracing::TracingReceiver> for TracingReceiver {
+	fn into(self) -> substrate_tracing::TracingReceiver {
+		match self {
+			TracingReceiver::Log => substrate_tracing::TracingReceiver::Log,
+			TracingReceiver::Telemetry => substrate_tracing::TracingReceiver::Telemetry,
+		}
+	}
+}
+
 /// Execution strategies parameters.
 #[derive(Debug, StructOpt, Clone)]
 pub struct ExecutionStrategies {
@@ -491,6 +509,20 @@ pub struct RunCmd {
 	#[structopt(long = "force-authoring")]
 	pub force_authoring: bool,
 
+	/// Comma separated list of targets for tracing
+	#[structopt(long = "tracing-targets", value_name = "TARGETS")]
+	pub tracing_targets: Option<String>,
+
+	/// Receiver to process tracing messages
+	#[structopt(
+		long = "tracing-receiver",
+		value_name = "RECEIVER",
+		possible_values = &TracingReceiver::variants(),
+		case_insensitive = true,
+		default_value = "Log"
+	)]
+	pub tracing_receiver: TracingReceiver,
+
 	/// Specify custom keystore path.
 	#[structopt(long = "keystore-path", value_name = "PATH", parse(from_os_str))]
 	pub keystore_path: Option<PathBuf>,

substrate/client/service/Cargo.toml

@@ -50,6 +50,8 @@ rpc = { package = "substrate-rpc", path = "../rpc" }
 tel = { package = "substrate-telemetry", path = "../telemetry" }
 offchain = { package = "substrate-offchain", path = "../offchain" }
 parity-multiaddr = { package = "parity-multiaddr", version = "0.5.0" }
+substrate-tracing = { package = "substrate-tracing", path = "../tracing" }
+tracing = "0.1.10"
 
 [dev-dependencies]
 substrate-test-runtime-client = { path = "../../test/utils/runtime/client" }

substrate/client/service/src/builder.rs

@@ -35,7 +35,7 @@ use futures03::{
 	StreamExt as _, TryStreamExt as _,
 };
 use keystore::{Store as Keystore};
-use log::{info, warn};
+use log::{info, warn, error};
 use network::{FinalityProofProvider, OnDemand, NetworkService, NetworkStateInfo, DhtEvent};
 use network::{config::BoxFinalityProofRequestBuilder, specialization::NetworkSpecialization};
 use parking_lot::{Mutex, RwLock};
@@ -1130,6 +1130,17 @@ ServiceBuilder<
 			telemetry
 		});
 
+		// Instrumentation
+		if let Some(tracing_targets) = config.tracing_targets.as_ref() {
+			let subscriber = substrate_tracing::ProfilingSubscriber::new(
+				config.tracing_receiver, tracing_targets
+			);
+			match tracing::subscriber::set_global_default(subscriber) {
+				Ok(_) => (),
+				Err(e) => error!(target: "tracing", "Unable to set global default subscriber {}", e),
+			}
+		}
+
 		Ok(Service {
 			client,
 			network,

substrate/client/service/src/config.rs

@@ -98,6 +98,10 @@ pub struct Configuration<C, G, E = NoExtension> {
 	///
 	/// Should only be set when `node` is running development mode.
 	pub dev_key_seed: Option<String>,
+	/// Tracing targets
+	pub tracing_targets: Option<String>,
+	/// Tracing receiver
+	pub tracing_receiver: substrate_tracing::TracingReceiver,
 }
 
 /// Configuration of the database of the client.
@@ -156,6 +160,8 @@ impl<C, G, E> Configuration<C, G, E> where
 			disable_grandpa: false,
 			keystore_password: None,
 			dev_key_seed: None,
+			tracing_targets: Default::default(),
+			tracing_receiver: Default::default(),
 		};
 		configuration.network.boot_nodes = configuration.chain_spec.boot_nodes().to_vec();
 

substrate/client/service/test/src/lib.rs

@@ -198,6 +198,8 @@ fn node_config<G, E: Clone> (
 		force_authoring: false,
 		disable_grandpa: false,
 		dev_key_seed: key_seed,
+		tracing_targets: None,
+		tracing_receiver: Default::default(),
 	}
 }
 

substrate/client/tracing/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "substrate-tracing"
+version = "2.0.0"
+license = "GPL-3.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2018"
+
+[dependencies]
+log = { version = "0.4.8" }
+parking_lot = "0.9.0"
+tracing-core = "0.1.7"
+
+substrate-telemetry = { path = "../telemetry" }
+
+[dev-dependencies]
+tracing = "0.1.10"

substrate/client/tracing/src/lib.rs

@@ -0,0 +1,198 @@
+// Copyright 2019 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Substrate is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Instrumentation implementation for substrate.
+//!
+//! This crate is unstable and the API and usage may change.
+//!
+//! # Usage
+//!
+//! Monitor performance throughout the codebase via the creation of `Span`s.
+//! A span is set in the following way:
+//! ```
+//! let span = tracing::span!(tracing::Level::INFO, "my_span_name");
+//! let _enter = span.enter();
+//! ```
+//! To begin timing, a span must be entered. When the span is dropped, the execution time
+//! is recorded and details sent to the `Receiver` which defines how to process it.
+//!
+//! Currently we provide `Log` (default) and `Telemetry` variants for `Receiver`
+
+use std::collections::HashMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+
+use parking_lot::Mutex;
+use tracing_core::{event::Event, Level, metadata::Metadata, span::{Attributes, Id, Record}, subscriber::Subscriber};
+
+use substrate_telemetry::{telemetry, SUBSTRATE_INFO};
+
+/// Used to configure how to receive the metrics
+#[derive(Debug, Clone)]
+pub enum TracingReceiver {
+	/// Output to logger
+	Log,
+	/// Output to telemetry
+	Telemetry,
+}
+
+impl Default for TracingReceiver {
+	fn default() -> Self {
+		Self::Log
+	}
+}
+
+#[derive(Debug)]
+struct SpanDatum {
+	id: u64,
+	name: &'static str,
+	target: String,
+	level: Level,
+	line: u32,
+	start_time: Instant,
+	overall_time: Duration,
+}
+
+/// Responsible for assigning ids to new spans, which are not re-used.
+pub struct ProfilingSubscriber {
+	next_id: AtomicU64,
+	targets: Vec<(String, Level)>,
+	receiver: TracingReceiver,
+	span_data: Mutex<HashMap<u64, SpanDatum>>,
+}
+
+impl ProfilingSubscriber {
+	/// Takes a `Receiver` and a comma separated list of targets,
+	/// either with a level "paint=trace"
+	/// or without "paint".
+	pub fn new(receiver: TracingReceiver, targets: &String) -> Self {
+		let targets: Vec<_> = targets.split(',').map(|s| parse_target(s)).collect();
+		ProfilingSubscriber {
+			next_id: AtomicU64::new(1),
+			targets,
+			receiver,
+			span_data: Mutex::new(HashMap::new()),
+		}
+	}
+}
+
+// Default to TRACE if no level given or unable to parse Level
+// We do not support a global `Level` currently
+fn parse_target(s: &str) -> (String, Level) {
+	match s.find("=") {
+		Some(i) => {
+			let target = s[0..i].to_string();
+			if s.len() > i {
+				let level = s[i + 1..s.len()].parse::<Level>().unwrap_or(Level::TRACE);
+				(target, level)
+			} else {
+				(target, Level::TRACE)
+			}
+		}
+		None => (s.to_string(), Level::TRACE)
+	}
+}
+
+impl Subscriber for ProfilingSubscriber {
+	fn enabled(&self, metadata: &Metadata<'_>) -> bool {
+		for t in &self.targets {
+			if metadata.target().starts_with(t.0.as_str()) && metadata.level() <= &t.1 {
+				log::debug!("Enabled target: {}, level: {}", metadata.target(), metadata.level());
+				return true;
+			} else {
+				log::debug!("Disabled target: {}, level: {}", metadata.target(), metadata.level());
+			}
+		}
+		false
+	}
+
+	fn new_span(&self, attrs: &Attributes<'_>) -> Id {
+		let id = self.next_id.fetch_add(1, Ordering::Relaxed);
+		let span_datum = SpanDatum {
+			id: id,
+			name: attrs.metadata().name(),
+			target: attrs.metadata().target().to_string(),
+			level: attrs.metadata().level().clone(),
+			line: attrs.metadata().line().unwrap_or(0),
+			start_time: Instant::now(),
+			overall_time: Duration::from_nanos(0),
+		};
+		self.span_data.lock().insert(id, span_datum);
+		Id::from_u64(id)
+	}
+
+	fn record(&self, _span: &Id, _values: &Record<'_>) {}
+
+	fn record_follows_from(&self, _span: &Id, _follows: &Id) {}
+
+	fn event(&self, _event: &Event<'_>) {}
+
+	fn enter(&self, span: &Id) {
+		let mut span_data = self.span_data.lock();
+		let start_time = Instant::now();
+		if let Some(mut s) = span_data.get_mut(&span.into_u64()) {
+			s.start_time = start_time;
+		} else {
+			log::warn!("Tried to enter span {:?} that has already been closed!", span);
+		}
+	}
+
+	fn exit(&self, span: &Id) {
+		let mut span_data = self.span_data.lock();
+		let end_time = Instant::now();
+		if let Some(mut s) = span_data.get_mut(&span.into_u64()) {
+			s.overall_time = end_time - s.start_time + s.overall_time;
+		}
+	}
+
+	fn try_close(&self, span: Id) -> bool {
+		let mut span_data = self.span_data.lock();
+		if let Some(data) = span_data.remove(&span.into_u64()) {
+			self.send_span(data);
+		};
+		true
+	}
+}
+
+impl ProfilingSubscriber {
+	fn send_span(&self, span_datum: SpanDatum) {
+		match self.receiver {
+			TracingReceiver::Log => print_log(span_datum),
+			TracingReceiver::Telemetry => send_telemetry(span_datum),
+		}
+	}
+}
+
+fn print_log(span_datum: SpanDatum) {
+	let message = format!(
+		"Tracing: {} {}: {}, line: {}, time: {} ns",
+		span_datum.level,
+		span_datum.target,
+		span_datum.name,
+		span_datum.line,
+		span_datum.overall_time.as_nanos()
+	);
+	log::info!(target: "substrate_tracing", "{}", message);
+}
+
+fn send_telemetry(span_datum: SpanDatum) {
+	telemetry!(SUBSTRATE_INFO; "tracing.profiling";
+		"name" => span_datum.name,
+		"target" => span_datum.target,
+		"line" => span_datum.line,
+		"time" => span_datum.overall_time.as_nanos(),
+	);
+}