Rework telemetry to replace the use of tracing with an object we pass around

bin/node-template/node/src/service.rs

@@ -11,7 +11,7 @@ pub use sc_executor::NativeExecutor;
 use sp_consensus_aura::sr25519::{AuthorityPair as AuraPair};
 use sc_finality_grandpa::SharedVoterState;
 use sc_keystore::LocalKeystore;
-use sc_telemetry::TelemetrySpan;
+use sc_telemetry::{Telemetry, TelemetryWorker};
 
 // Our native executor instance.
 native_executor_instance!(
@@ -37,6 +37,7 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 			AuraPair
 		>,
 		sc_finality_grandpa::LinkHalf<Block, FullClient, FullSelectChain>,
+		Option<Telemetry>,
 	)
 >, ServiceError> {
 	if config.keystore_remote.is_some() {
@@ -45,10 +46,20 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 	}
 	let inherent_data_providers = sp_inherents::InherentDataProviders::new();
 
+	let telemetry_worker = TelemetryWorker::new(16, None)?;
+	let telemetry = config.telemetry_endpoints.clone()
+		.filter(|x| !x.is_empty())
+		.map(|endpoints| telemetry_worker.handle().new_telemetry(endpoints));
+
 	let (client, backend, keystore_container, task_manager) =
-		sc_service::new_full_parts::<Block, RuntimeApi, Executor>(&config)?;
+		sc_service::new_full_parts::<Block, RuntimeApi, Executor>(
+			&config,
+			telemetry.as_ref().map(|x| x.handle()),
+		)?;
 	let client = Arc::new(client);
 
+	task_manager.spawn_handle().spawn("telemetry", telemetry_worker.run());
+
 	let select_chain = sc_consensus::LongestChain::new(backend.clone());
 
 	let transaction_pool = sc_transaction_pool::BasicPool::new_full(
@@ -60,7 +71,10 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 	);
 
 	let (grandpa_block_import, grandpa_link) = sc_finality_grandpa::block_import(
-		client.clone(), &(client.clone() as Arc<_>), select_chain.clone(),
+		client.clone(),
+		&(client.clone() as Arc<_>),
+		select_chain.clone(),
+		telemetry.as_ref().map(|x| x.handle()),
 	)?;
 
 	let aura_block_import = sc_consensus_aura::AuraBlockImport::<_, _, _, AuraPair>::new(
@@ -76,6 +90,7 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 		&task_manager.spawn_essential_handle(),
 		config.prometheus_registry(),
 		sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),
+		telemetry.as_ref().map(|x| x.handle()),
 	)?;
 
 	Ok(sc_service::PartialComponents {
@@ -87,7 +102,7 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 		select_chain,
 		transaction_pool,
 		inherent_data_providers,
-		other: (aura_block_import, grandpa_link),
+		other: (aura_block_import, grandpa_link, telemetry),
 	})
 }
 
@@ -109,7 +124,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		select_chain,
 		transaction_pool,
 		inherent_data_providers,
-		other: (block_import, grandpa_link),
+		other: (block_import, grandpa_link, mut telemetry),
 	} = new_partial(&config)?;
 
 	if let Some(url) = &config.keystore_remote {
@@ -163,10 +178,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		})
 	};
 
-	let telemetry_span = TelemetrySpan::new();
-	let _telemetry_span_entered = telemetry_span.enter();
-
-	let (_rpc_handlers, telemetry_connection_notifier) = sc_service::spawn_tasks(
+	let _rpc_handlers = sc_service::spawn_tasks(
 		sc_service::SpawnTasksParams {
 			network: network.clone(),
 			client: client.clone(),
@@ -180,7 +192,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 			network_status_sinks,
 			system_rpc_tx,
 			config,
-			telemetry_span: Some(telemetry_span.clone()),
+			telemetry: telemetry.as_mut(),
 		},
 	)?;
 
@@ -190,6 +202,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 			client.clone(),
 			transaction_pool,
 			prometheus_registry.as_ref(),
+			telemetry.as_ref().map(|x| x.handle()),
 		);
 
 		let can_author_with =
@@ -207,6 +220,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 			backoff_authoring_blocks,
 			keystore_container.sync_keystore(),
 			can_author_with,
+			telemetry.as_ref().map(|x| x.handle()),
 		)?;
 
 		// the AURA authoring task is considered essential, i.e. if it
@@ -230,6 +244,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		observer_enabled: false,
 		keystore,
 		is_authority: role.is_authority(),
+		telemetry: telemetry.as_ref().map(|x| x.handle()),
 	};
 
 	if enable_grandpa {
@@ -243,10 +258,10 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 			config: grandpa_config,
 			link: grandpa_link,
 			network,
-			telemetry_on_connect: telemetry_connection_notifier.map(|x| x.on_connect_stream()),
 			voting_rule: sc_finality_grandpa::VotingRulesBuilder::default().build(),
 			prometheus_registry,
 			shared_voter_state: SharedVoterState::empty(),
+			telemetry: telemetry.as_ref().map(|x| x.handle()),
 		};
 
 		// the GRANDPA voter task is considered infallible, i.e.
@@ -263,8 +278,18 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 
 /// Builds a new service for a light client.
 pub fn new_light(mut config: Configuration) -> Result<TaskManager, ServiceError> {
+	let telemetry_worker = TelemetryWorker::new(16, None)?;
+	let mut telemetry = config.telemetry_endpoints.clone()
+		.filter(|x| !x.is_empty())
+		.map(|endpoints| telemetry_worker.handle().new_telemetry(endpoints));
+
 	let (client, backend, keystore_container, mut task_manager, on_demand) =
-		sc_service::new_light_parts::<Block, RuntimeApi, Executor>(&config)?;
+		sc_service::new_light_parts::<Block, RuntimeApi, Executor>(
+			&config,
+			telemetry.as_ref().map(|x| x.handle()),
+		)?;
+
+	task_manager.spawn_handle().spawn("telemetry", telemetry_worker.run());
 
 	config.network.extra_sets.push(sc_finality_grandpa::grandpa_peers_set_config());
 
@@ -282,6 +307,7 @@ pub fn new_light(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		client.clone(),
 		&(client.clone() as Arc<_>),
 		select_chain.clone(),
+		telemetry.as_ref().map(|x| x.handle()),
 	)?;
 
 	let aura_block_import = sc_consensus_aura::AuraBlockImport::<_, _, _, AuraPair>::new(
@@ -298,6 +324,7 @@ pub fn new_light(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		&task_manager.spawn_essential_handle(),
 		config.prometheus_registry(),
 		sp_consensus::NeverCanAuthor,
+		telemetry.as_ref().map(|x| x.handle()),
 	)?;
 
 	let (network, network_status_sinks, system_rpc_tx, network_starter) =
@@ -317,9 +344,6 @@ pub fn new_light(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		);
 	}
 
-	let telemetry_span = TelemetrySpan::new();
-	let _telemetry_span_entered = telemetry_span.enter();
-
 	sc_service::spawn_tasks(sc_service::SpawnTasksParams {
 		remote_blockchain: Some(backend.remote_blockchain()),
 		transaction_pool,
@@ -333,7 +357,7 @@ pub fn new_light(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		network,
 		network_status_sinks,
 		system_rpc_tx,
-		telemetry_span: Some(telemetry_span.clone()),
+		telemetry: telemetry.as_mut(),
 	})?;
 
 	network_starter.start_network();

bin/node/bench/src/construct.rs

@@ -151,6 +151,7 @@ impl core::Benchmark for ConstructionBenchmark {
 			context.client.clone(),
 			self.transactions.clone().into(),
 			None,
+			None,
 		);
 		let inherent_data_providers = sp_inherents::InherentDataProviders::new();
 		inherent_data_providers

bin/node/cli/Cargo.toml

@@ -105,6 +105,7 @@ try-runtime-cli = { version = "0.9.0", optional = true, path = "../../../utils/f
 wasm-bindgen = { version = "0.2.57", optional = true }
 wasm-bindgen-futures = { version = "0.4.18", optional = true }
 browser-utils = { package = "substrate-browser-utils", path = "../../../utils/browser", optional = true, version = "0.9.0"}
+libp2p-wasm-ext = { version = "0.27", features = ["websocket"], optional = true }
 
 [target.'cfg(target_arch="x86_64")'.dependencies]
 node-executor = { version = "2.0.0", path = "../executor", features = [ "wasmtime" ] }
@@ -148,6 +149,7 @@ browser = [
 	"browser-utils",
 	"wasm-bindgen",
 	"wasm-bindgen-futures",
+	"libp2p-wasm-ext",
 ]
 cli = [
 	"node-executor/wasmi-errno",

bin/node/cli/src/browser.rs

@@ -21,7 +21,7 @@ use log::info;
 use wasm_bindgen::prelude::*;
 use browser_utils::{
 	Client,
-	browser_configuration, init_logging_and_telemetry, set_console_error_panic_hook,
+	browser_configuration, init_logging, set_console_error_panic_hook,
 };
 
 /// Starts the client.
@@ -37,18 +37,14 @@ async fn start_inner(
 	log_directives: String,
 ) -> Result<Client, Box<dyn std::error::Error>> {
 	set_console_error_panic_hook();
-	let telemetry_worker = init_logging_and_telemetry(&log_directives)?;
+	init_logging(&log_directives)?;
 	let chain_spec = match chain_spec {
 		Some(chain_spec) => ChainSpec::from_json_bytes(chain_spec.as_bytes().to_vec())
 			.map_err(|e| format!("{:?}", e))?,
 		None => crate::chain_spec::development_config(),
 	};
 
-	let telemetry_handle = telemetry_worker.handle();
-	let config = browser_configuration(
-		chain_spec,
-		Some(telemetry_handle),
-	).await?;
+	let config = browser_configuration(chain_spec).await?;
 
 	info!("Substrate browser node");
 	info!("✌️  version {}", config.impl_version);
@@ -60,10 +56,8 @@ async fn start_inner(
 	// Create the service. This is the most heavy initialization step.
 	let (task_manager, rpc_handlers) =
 		crate::service::new_light_base(config)
-			.map(|(components, rpc_handlers, _, _, _, _)| (components, rpc_handlers))
+			.map(|(components, rpc_handlers, _, _, _)| (components, rpc_handlers))
 			.map_err(|e| format!("{:?}", e))?;
 
-	task_manager.spawn_handle().spawn("telemetry", telemetry_worker.run());
-
 	Ok(browser_utils::start_client(task_manager, rpc_handlers))
 }

bin/node/cli/src/chain_spec.rs

@@ -441,7 +441,7 @@ pub(crate) mod tests {
 				Ok(sc_service_test::TestNetComponents::new(task_manager, client, network, transaction_pool))
 			},
 			|config| {
-				let (keep_alive, _, _, client, network, transaction_pool) = new_light_base(config)?;
+				let (keep_alive, _, client, network, transaction_pool) = new_light_base(config)?;
 				Ok(sc_service_test::TestNetComponents::new(keep_alive, client, network, transaction_pool))
 			}
 		);