Allow adding a prefix to the informant (#6174)

* Initial commit

Forked at: 1d93746b64
Parent branch: origin/master

* Add a Service Configuration's field + adapt informant + provide means to CLI

* CLEANUP

Forked at: 1d93746b64
Parent branch: origin/master

* fix tests

* fixed bad path to object

* Change OutputFormat enum to struct

* Add informant_prefix to builder and service

* Revert "Change OutputFormat enum to struct"

This reverts commit cd86c583c92668426c35cc174401155bf2880c1f.

* Revert "fix tests"

This reverts commit a3c306ebe94720f350c5bc74b9c5fcde2565d340.

* Revert "Add a Service Configuration's field + adapt informant + provide means to CLI"

This reverts commit 9c2e7267423305705916c30d605893524113c8e3.

* Implementation using the ServiceBuilder

* reduce line length

* fix line width again

* WIP

Forked at: 1d93746b64
Parent branch: origin/master

* WIP

Forked at: 1d93746b64
Parent branch: origin/master

* WIP

Forked at: 1d93746b64
Parent branch: origin/master

* use struct instead of enum

* WIP

Forked at: 1d93746b64
Parent branch: origin/master

* Update client/service/src/lib.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* improve doc

* Update client/service/src/builder.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/service/src/builder.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* change code

* Update client/informant/src/lib.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* enable_color

* reorg log

* remove macro

* Removed builder for informant prefix

* fix doc

* Update client/informant/src/lib.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/informant/src/lib.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/informant/src/lib.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/informant/src/lib.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/service/src/builder.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/service/src/builder.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

* Update client/service/src/builder.rs

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

Co-authored-by: Bastian Köcher <bkchr@users.noreply.github.com>

substrate/Cargo.lock

@@ -6315,9 +6315,10 @@ dependencies = [
  "parity-util-mem",
  "sc-client-api",
  "sc-network",
- "sc-service",
  "sp-blockchain",
  "sp-runtime",
+ "sp-transaction-pool",
+ "sp-utils",
  "wasm-timer",
 ]
 
@@ -6610,6 +6611,7 @@ dependencies = [
  "sc-client-db",
  "sc-executor",
  "sc-finality-grandpa",
+ "sc-informant",
  "sc-keystore",
  "sc-network",
  "sc-offchain",

substrate/client/cli/src/runner.rs

@@ -266,9 +266,6 @@ impl<C: SubstrateCli> Runner<C> {
 	{
 		let service = service_builder(self.config)?;
 
-		let informant_future = sc_informant::build(&service, sc_informant::OutputFormat::Coloured);
-		let _informant_handle = self.tokio_runtime.spawn(informant_future);
-
 		// we eagerly drop the service so that the internal exit future is fired,
 		// but we need to keep holding a reference to the global telemetry guard
 		// and drop the runtime first.

substrate/client/informant/Cargo.toml

@@ -19,6 +19,7 @@ parity-util-mem = { version = "0.6.1", default-features = false, features = ["pr
 wasm-timer = "0.2"
 sc-client-api = { version = "2.0.0-rc3", path = "../api" }
 sc-network = { version = "0.8.0-rc3", path = "../network" }
-sc-service = { version = "0.8.0-rc3", default-features = false, path = "../service" }
 sp-blockchain = { version = "2.0.0-rc3", path = "../../primitives/blockchain" }
 sp-runtime = { version = "2.0.0-rc3", path = "../../primitives/runtime" }
+sp-utils = { version = "2.0.0-rc2", path = "../../primitives/utils" }
+sp-transaction-pool = { version = "2.0.0-rc2", path = "../../primitives/transaction-pool" }

substrate/client/informant/src/display.rs

@@ -14,15 +14,17 @@
 // You should have received a copy of the GNU General Public License
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
-use ansi_term::Colour;
-use sc_client_api::ClientInfo;
-use log::info;
-use sc_network::SyncState;
-use sp_runtime::traits::{Block as BlockT, CheckedDiv, NumberFor, Zero, Saturating};
-use sc_service::NetworkStatus;
-use std::{convert::{TryFrom, TryInto}, fmt};
-use wasm_timer::Instant;
 use crate::OutputFormat;
+use ansi_term::Colour;
+use log::info;
+use sc_client_api::ClientInfo;
+use sc_network::{NetworkStatus, SyncState};
+use sp_runtime::traits::{Block as BlockT, CheckedDiv, NumberFor, Saturating, Zero};
+use std::{
+	convert::{TryFrom, TryInto},
+	fmt,
+};
+use wasm_timer::Instant;
 
 /// State of the informant display system.
 ///
@@ -67,16 +69,22 @@ impl<B: BlockT> InformantDisplay<B> {
 		self.last_update = Instant::now();
 		self.last_number = Some(best_number);
 
-		let (status, target) = match (net_status.sync_state, net_status.best_seen_block) {
-			(SyncState::Idle, _) => ("💤 Idle".into(), "".into()),
-			(SyncState::Downloading, None) => (format!("⚙️  Preparing{}", speed), "".into()),
-			(SyncState::Downloading, Some(n)) => (format!("⚙️  Syncing{}", speed), format!(", target=#{}", n)),
+		let (level, status, target) = match (net_status.sync_state, net_status.best_seen_block) {
+			(SyncState::Idle, _) => ("💤", "Idle".into(), "".into()),
+			(SyncState::Downloading, None) => ("⚙️ ", format!("Preparing{}", speed), "".into()),
+			(SyncState::Downloading, Some(n)) => (
+				"⚙️ ",
+				format!("Syncing{}", speed),
+				format!(", target=#{}", n),
+			),
 		};
 
-		if self.format == OutputFormat::Coloured {
+		if self.format.enable_color {
 			info!(
 				target: "substrate",
-				"{}{} ({} peers), best: #{} ({}), finalized #{} ({}), {} {}",
+				"{} {}{}{} ({} peers), best: #{} ({}), finalized #{} ({}), {} {}",
+				level,
+				self.format.prefix,
 				Colour::White.bold().paint(&status),
 				target,
 				Colour::White.bold().paint(format!("{}", num_connected_peers)),
@@ -86,11 +94,13 @@ impl<B: BlockT> InformantDisplay<B> {
 				info.chain.finalized_hash,
 				Colour::Green.paint(format!("⬇ {}", TransferRateFormat(net_status.average_download_per_sec))),
 				Colour::Red.paint(format!("⬆ {}", TransferRateFormat(net_status.average_upload_per_sec))),
-			);
+			)
 		} else {
 			info!(
 				target: "substrate",
-				"{}{} ({} peers), best: #{} ({}), finalized #{} ({}), ⬇ {} ⬆ {}",
+				"{} {}{}{} ({} peers), best: #{} ({}), finalized #{} ({}), ⬇ {} ⬆ {}",
+				level,
+				self.format.prefix,
 				status,
 				target,
 				num_connected_peers,
@@ -100,7 +110,7 @@ impl<B: BlockT> InformantDisplay<B> {
 				info.chain.finalized_hash,
 				TransferRateFormat(net_status.average_download_per_sec),
 				TransferRateFormat(net_status.average_upload_per_sec),
-			);
+			)
 		}
 	}
 }

substrate/client/informant/src/lib.rs

@@ -19,33 +19,66 @@
 //! Console informant. Prints sync progress and block events. Runs on the calling thread.
 
 use ansi_term::Colour;
-use sc_client_api::{BlockchainEvents, UsageProvider};
 use futures::prelude::*;
-use log::{info, warn, trace};
-use sp_runtime::traits::Header;
-use sc_service::AbstractService;
+use log::{info, trace, warn};
+use parity_util_mem::MallocSizeOf;
+use sc_client_api::{BlockchainEvents, UsageProvider};
+use sc_network::{network_state::NetworkState, NetworkStatus};
+use sp_blockchain::HeaderMetadata;
+use sp_runtime::traits::{Block as BlockT, Header};
+use sp_transaction_pool::TransactionPool;
+use sp_utils::mpsc::TracingUnboundedReceiver;
+use std::fmt::Display;
+use std::sync::Arc;
 use std::time::Duration;
 
 mod display;
 
 /// The format to print telemetry output in.
-#[derive(PartialEq)]
-pub enum OutputFormat {
-	Coloured,
-	Plain,
+#[derive(Clone)]
+pub struct OutputFormat {
+	/// Enable color output in logs.
+	pub enable_color: bool,
+	/// Add a prefix before every log line
+	pub prefix: String,
 }
 
-/// Creates an informant in the form of a `Future` that must be polled regularly.
-pub fn build(service: &impl AbstractService, format: OutputFormat) -> impl futures::Future<Output = ()> {
-	let client = service.client();
-	let pool = service.transaction_pool();
+/// Marker trait for a type that implements `TransactionPool` and `MallocSizeOf` on `not(target_os = "unknown")`.
+#[cfg(target_os = "unknown")]
+pub trait TransactionPoolAndMaybeMallogSizeOf: TransactionPool {}
 
-	let mut display = display::InformantDisplay::new(format);
+/// Marker trait for a type that implements `TransactionPool` and `MallocSizeOf` on `not(target_os = "unknown")`.
+#[cfg(not(target_os = "unknown"))]
+pub trait TransactionPoolAndMaybeMallogSizeOf: TransactionPool + MallocSizeOf {}
 
-	let display_notifications = service
-		.network_status(Duration::from_millis(5000))
+#[cfg(target_os = "unknown")]
+impl<T: TransactionPool> TransactionPoolAndMaybeMallogSizeOf for T {}
+
+#[cfg(not(target_os = "unknown"))]
+impl<T: TransactionPool + MallocSizeOf> TransactionPoolAndMaybeMallogSizeOf for T {}
+
+/// Builds the informant and returns a `Future` that drives the informant.
+pub fn build<B: BlockT, C>(
+	client: Arc<C>,
+	network_status_stream_builder: impl FnOnce(
+		Duration,
+	) -> TracingUnboundedReceiver<(
+		NetworkStatus<B>,
+		NetworkState,
+	)>,
+	pool: Arc<impl TransactionPoolAndMaybeMallogSizeOf>,
+	format: OutputFormat,
+) -> impl futures::Future<Output = ()>
+where
+	C: UsageProvider<B> + HeaderMetadata<B> + BlockchainEvents<B>,
+	<C as HeaderMetadata<B>>::Error: Display,
+{
+	let mut display = display::InformantDisplay::new(format.clone());
+
+	let client_1 = client.clone();
+	let display_notifications = network_status_stream_builder(Duration::from_millis(5000))
 		.for_each(move |(net_status, _)| {
-			let info = client.usage_info();
+			let info = client_1.usage_info();
 			if let Some(ref usage) = info.usage {
 				trace!(target: "usage", "Usage statistics: {}", usage);
 			} else {
@@ -64,7 +97,6 @@ pub fn build(service: &impl AbstractService, format: OutputFormat) -> impl futur
 			future::ready(())
 		});
 
-	let client = service.client();
 	let mut last_best = {
 		let info = client.usage_info();
 		Some((info.chain.best_number, info.chain.best_hash))
@@ -82,7 +114,8 @@ pub fn build(service: &impl AbstractService, format: OutputFormat) -> impl futur
 
 				match maybe_ancestor {
 					Ok(ref ancestor) if ancestor.hash != *last_hash => info!(
-						"♻️  Reorg on #{},{} to #{},{}, common ancestor #{},{}",
+						"♻️  {}Reorg on #{},{} to #{},{}, common ancestor #{},{}",
+						format.prefix,
 						Colour::Red.bold().paint(format!("{}", last_num)), last_hash,
 						Colour::Green.bold().paint(format!("{}", n.header.number())), n.hash,
 						Colour::White.bold().paint(format!("{}", ancestor.number)), ancestor.hash,
@@ -97,7 +130,13 @@ pub fn build(service: &impl AbstractService, format: OutputFormat) -> impl futur
 			last_best = Some((n.header.number().clone(), n.hash.clone()));
 		}
 
-		info!(target: "substrate", "✨ Imported #{} ({})", Colour::White.bold().paint(format!("{}", n.header.number())), n.hash);
+		info!(
+			target: "substrate",
+			"✨ {}Imported #{} ({})",
+			format.prefix,
+			Colour::White.bold().paint(format!("{}", n.header.number())), 
+			n.hash,
+		);
 		future::ready(())
 	});
 

substrate/client/network/src/lib.rs

@@ -269,6 +269,7 @@ pub use libp2p::{Multiaddr, PeerId};
 pub use libp2p::multiaddr;
 
 pub use sc_peerset::ReputationChange;
+use sp_runtime::traits::{Block as BlockT, NumberFor};
 
 /// The maximum allowed number of established connections per peer.
 ///
@@ -293,3 +294,22 @@ pub trait NetworkStateInfo {
 	/// Returns the local Peer ID.
 	fn local_peer_id(&self) -> PeerId;
 }
+
+/// Overview status of the network.
+#[derive(Clone)]
+pub struct NetworkStatus<B: BlockT> {
+	/// Current global sync state.
+	pub sync_state: SyncState,
+	/// Target sync block number.
+	pub best_seen_block: Option<NumberFor<B>>,
+	/// Number of peers participating in syncing.
+	pub num_sync_peers: u32,
+	/// Total number of connected peers
+	pub num_connected_peers: usize,
+	/// Total number of active peers.
+	pub num_active_peers: usize,
+	/// Downloaded bytes per second averaged over the past few seconds.
+	pub average_download_per_sec: u64,
+	/// Uploaded bytes per second averaged over the past few seconds.
+	pub average_upload_per_sec: u64,
+}

substrate/client/service/Cargo.toml

@@ -66,7 +66,7 @@ sc-rpc-server = { version = "2.0.0-rc3", path = "../rpc-servers" }
 sc-rpc = { version = "2.0.0-rc3", path = "../rpc" }
 sc-block-builder = { version = "0.8.0-rc3", path = "../block-builder" }
 sp-block-builder = { version = "2.0.0-rc3", path = "../../primitives/block-builder" }
-
+sc-informant = { version = "0.8.0-rc2", path = "../informant" }
 sc-telemetry = { version = "2.0.0-rc3", path = "../telemetry" }
 sc-offchain = { version = "2.0.0-rc3", path = "../offchain" }
 parity-multiaddr = { package = "parity-multiaddr", version = "0.7.3" }

substrate/client/service/src/builder.rs

@@ -101,6 +101,7 @@ pub struct ServiceBuilder<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp,
 	remote_backend: Option<Arc<dyn RemoteBlockchain<TBl>>>,
 	marker: PhantomData<(TBl, TRtApi)>,
 	block_announce_validator_builder: Option<Box<dyn FnOnce(Arc<TCl>) -> Box<dyn BlockAnnounceValidator<TBl> + Send> + Send>>,
+	informant_prefix: String,
 }
 
 /// A utility trait for building an RPC extension given a `DenyUnsafe` instance.
@@ -364,6 +365,7 @@ impl ServiceBuilder<(), (), (), (), (), (), (), (), (), (), ()> {
 			rpc_extensions_builder: Box::new(|_| ()),
 			remote_backend: None,
 			block_announce_validator_builder: None,
+			informant_prefix: Default::default(),
 			marker: PhantomData,
 		})
 	}
@@ -447,6 +449,7 @@ impl ServiceBuilder<(), (), (), (), (), (), (), (), (), (), ()> {
 			rpc_extensions_builder: Box::new(|_| ()),
 			remote_backend: Some(remote_blockchain),
 			block_announce_validator_builder: None,
+			informant_prefix: Default::default(),
 			marker: PhantomData,
 		})
 	}
@@ -541,6 +544,7 @@ impl<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend>
 			rpc_extensions_builder: self.rpc_extensions_builder,
 			remote_backend: self.remote_backend,
 			block_announce_validator_builder: self.block_announce_validator_builder,
+			informant_prefix: self.informant_prefix,
 			marker: self.marker,
 		})
 	}
@@ -586,6 +590,7 @@ impl<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend>
 			rpc_extensions_builder: self.rpc_extensions_builder,
 			remote_backend: self.remote_backend,
 			block_announce_validator_builder: self.block_announce_validator_builder,
+			informant_prefix: self.informant_prefix,
 			marker: self.marker,
 		})
 	}
@@ -624,6 +629,7 @@ impl<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend>
 			rpc_extensions_builder: self.rpc_extensions_builder,
 			remote_backend: self.remote_backend,
 			block_announce_validator_builder: self.block_announce_validator_builder,
+			informant_prefix: self.informant_prefix,
 			marker: self.marker,
 		})
 	}
@@ -690,6 +696,7 @@ impl<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend>
 			rpc_extensions_builder: self.rpc_extensions_builder,
 			remote_backend: self.remote_backend,
 			block_announce_validator_builder: self.block_announce_validator_builder,
+			informant_prefix: self.informant_prefix,
 			marker: self.marker,
 		})
 	}
@@ -746,6 +753,7 @@ impl<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend>
 			rpc_extensions_builder: self.rpc_extensions_builder,
 			remote_backend: self.remote_backend,
 			block_announce_validator_builder: self.block_announce_validator_builder,
+			informant_prefix: self.informant_prefix,
 			marker: self.marker,
 		})
 	}
@@ -783,6 +791,7 @@ impl<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend>
 			rpc_extensions_builder: Box::new(rpc_extensions_builder),
 			remote_backend: self.remote_backend,
 			block_announce_validator_builder: self.block_announce_validator_builder,
+			informant_prefix: self.informant_prefix,
 			marker: self.marker,
 		})
 	}
@@ -828,9 +837,43 @@ impl<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend>
 			rpc_extensions_builder: self.rpc_extensions_builder,
 			remote_backend: self.remote_backend,
 			block_announce_validator_builder: Some(Box::new(block_announce_validator_builder)),
+			informant_prefix: self.informant_prefix,
 			marker: self.marker,
 		})
 	}
+
+	/// Defines the informant's prefix for the logs. An empty string by default.
+	///
+	/// By default substrate will show logs without a prefix. Example:
+	///
+	/// ```text
+	/// 2020-05-28 15:11:06 ✨ Imported #2 (0xc21c…2ca8)
+	/// 2020-05-28 15:11:07 💤 Idle (0 peers), best: #2 (0xc21c…2ca8), finalized #0 (0x7299…e6df), ⬇ 0 ⬆ 0
+	/// ```
+	///
+	/// But you can define a prefix by using this function. Example:
+	///
+	/// ```rust,ignore
+	/// service.with_informant_prefix("[Prefix] ".to_string());
+	/// ```
+	///
+	/// This will output:
+	///
+	/// ```text
+	/// 2020-05-28 15:11:06 ✨ [Prefix] Imported #2 (0xc21c…2ca8)
+	/// 2020-05-28 15:11:07 💤 [Prefix] Idle (0 peers), best: #2 (0xc21c…2ca8), finalized #0 (0x7299…e6df), ⬇ 0 ⬆ 0
+	/// ```
+	pub fn with_informant_prefix(
+		self,
+		informant_prefix: String,
+	) -> Result<ServiceBuilder<TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp,
+		TExPool, TRpc, Backend>, Error>
+	where TSc: Clone, TFchr: Clone {
+		Ok(ServiceBuilder {
+			informant_prefix: informant_prefix,
+			..self
+		})
+	}
 }
 
 /// Implemented on `ServiceBuilder`. Allows running block commands, such as import/export/validate
@@ -947,6 +990,7 @@ ServiceBuilder<
 			rpc_extensions_builder,
 			remote_backend,
 			block_announce_validator_builder,
+			informant_prefix,
 		} = self;
 
 		sp_session::generate_initial_session_keys(
@@ -1342,6 +1386,20 @@ ServiceBuilder<
 			}
 		}
 
+		// Spawn informant task
+		let network_status_sinks_1 = network_status_sinks.clone();
+		let informant_future = sc_informant::build(
+			client.clone(),
+			move |interval| {
+				let (sink, stream) = tracing_unbounded("mpsc_network_status");
+				network_status_sinks_1.lock().push(interval, sink);
+				stream
+			},
+			transaction_pool.clone(),
+			sc_informant::OutputFormat { enable_color: true, prefix: informant_prefix },
+		);
+		spawn_handle.spawn("informant", informant_future);
+
 		Ok(Service {
 			client,
 			task_manager,

substrate/client/service/src/lib.rs

@@ -52,11 +52,11 @@ use futures::{
 	sink::SinkExt,
 	task::{Spawn, FutureObj, SpawnError},
 };
-use sc_network::{NetworkService, network_state::NetworkState, PeerId};
+use sc_network::{NetworkService, NetworkStatus, network_state::NetworkState, PeerId};
 use log::{log, warn, debug, error, Level};
 use codec::{Encode, Decode};
 use sp_runtime::generic::BlockId;
-use sp_runtime::traits::{NumberFor, Block as BlockT};
+use sp_runtime::traits::Block as BlockT;
 use parity_util_mem::MallocSizeOf;
 use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver,  TracingUnboundedSender};
 
@@ -487,25 +487,6 @@ fn build_network_future<
 	})
 }
 
-/// Overview status of the network.
-#[derive(Clone)]
-pub struct NetworkStatus<B: BlockT> {
-	/// Current global sync state.
-	pub sync_state: sc_network::SyncState,
-	/// Target sync block number.
-	pub best_seen_block: Option<NumberFor<B>>,
-	/// Number of peers participating in syncing.
-	pub num_sync_peers: u32,
-	/// Total number of connected peers
-	pub num_connected_peers: usize,
-	/// Total number of active peers.
-	pub num_active_peers: usize,
-	/// Downloaded bytes per second averaged over the past few seconds.
-	pub average_download_per_sec: u64,
-	/// Uploaded bytes per second averaged over the past few seconds.
-	pub average_upload_per_sec: u64,
-}
-
 #[cfg(not(target_os = "unknown"))]
 // Wrapper for HTTP and WS servers that makes sure they are properly shut down.
 mod waiting {

substrate/utils/browser/src/lib.rs

@@ -116,11 +116,6 @@ struct RpcMessage {
 
 /// Create a Client object that connects to a service.
 pub fn start_client(mut service: impl AbstractService) -> Client {
-	// Spawn informant
-	wasm_bindgen_futures::spawn_local(
-		sc_informant::build(&service, sc_informant::OutputFormat::Plain).map(drop)
-	);
-
 	// We dispatch a background task responsible for processing the service.
 	//
 	// The main action performed by the code below consists in polling the service with