Fix telemetry span not entering properly attempt 3

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

@@ -11,6 +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;
 
 // Our native executor instance.
 native_executor_instance!(
@@ -161,6 +162,9 @@ 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(
 		sc_service::SpawnTasksParams {
 			network: network.clone(),
@@ -175,6 +179,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 			network_status_sinks,
 			system_rpc_tx,
 			config,
+			telemetry_span: Some(telemetry_span.clone()),
 		},
 	)?;
 
@@ -311,6 +316,9 @@ 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,
@@ -324,6 +332,7 @@ pub fn new_light(mut config: Configuration) -> Result<TaskManager, ServiceError>
 		network,
 		network_status_sinks,
 		system_rpc_tx,
+		telemetry_span: Some(telemetry_span.clone()),
 	})?;
 
 	network_starter.start_network();

bin/node/cli/src/service.rs

@@ -34,7 +34,7 @@ use sp_runtime::traits::Block as BlockT;
 use futures::prelude::*;
 use sc_client_api::{ExecutorProvider, RemoteBackend};
 use node_executor::Executor;
-use sc_telemetry::TelemetryConnectionNotifier;
+use sc_telemetry::{TelemetryConnectionNotifier, TelemetrySpan};
 
 type FullClient = sc_service::TFullClient<Block, RuntimeApi, Executor>;
 type FullBackend = sc_service::TFullBackend<Block>;
@@ -225,6 +225,9 @@ pub fn new_full_base(
 	let enable_grandpa = !config.disable_grandpa;
 	let prometheus_registry = config.prometheus_registry().cloned();
 
+	let telemetry_span = TelemetrySpan::new();
+	let _telemetry_span_entered = telemetry_span.enter();
+
 	let (_rpc_handlers, telemetry_connection_notifier) = sc_service::spawn_tasks(
 		sc_service::SpawnTasksParams {
 			config,
@@ -239,6 +242,7 @@ pub fn new_full_base(
 			remote_blockchain: None,
 			network_status_sinks: network_status_sinks.clone(),
 			system_rpc_tx,
+			telemetry_span: Some(telemetry_span.clone()),
 		},
 	)?;
 
@@ -432,6 +436,9 @@ pub fn new_light_base(mut config: Configuration) -> Result<(
 
 	let rpc_extensions = node_rpc::create_light(light_deps);
 
+	let telemetry_span = TelemetrySpan::new();
+	let _telemetry_span_entered = telemetry_span.enter();
+
 	let (rpc_handlers, telemetry_connection_notifier) =
 		sc_service::spawn_tasks(sc_service::SpawnTasksParams {
 			on_demand: Some(on_demand),
@@ -443,6 +450,7 @@ pub fn new_light_base(mut config: Configuration) -> Result<(
 			config, backend, network_status_sinks, system_rpc_tx,
 			network: network.clone(),
 			task_manager: &mut task_manager,
+			telemetry_span: Some(telemetry_span.clone()),
 		})?;
 
 	Ok((

client/cli/src/config.rs

@@ -33,7 +33,7 @@ use sc_service::config::{
 	TaskExecutor, TelemetryEndpoints, TransactionPoolOptions, WasmExecutionMethod,
 };
 use sc_service::{ChainSpec, TracingReceiver, KeepBlocks, TransactionStorageMode};
-use sc_telemetry::{TelemetryHandle, TelemetrySpan};
+use sc_telemetry::TelemetryHandle;
 use sc_tracing::logging::LoggerBuilder;
 use std::net::SocketAddr;
 use std::path::PathBuf;
@@ -494,7 +494,6 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 			.transpose()?
 			// Don't initialise telemetry if `telemetry_endpoints` == Some([])
 			.filter(|x| !x.is_empty());
-		let telemetry_span = telemetry_endpoints.as_ref().map(|_| TelemetrySpan::new());
 
 		let unsafe_pruning = self
 			.import_params()
@@ -534,7 +533,6 @@ pub trait CliConfiguration<DCV: DefaultConfigurationValues = ()>: Sized {
 			rpc_cors: self.rpc_cors(is_dev)?,
 			prometheus_config: self.prometheus_config(DCV::prometheus_listen_port())?,
 			telemetry_endpoints,
-			telemetry_span,
 			telemetry_external_transport: self.telemetry_external_transport()?,
 			default_heap_pages: self.default_heap_pages()?,
 			offchain_worker: self.offchain_worker(&role)?,

client/service/Cargo.toml

@@ -92,3 +92,4 @@ grandpa-primitives = { version = "2.0.0", package = "sp-finality-grandpa", path
 tokio = { version = "0.2.25", default-features = false }
 async-std = { version = "1.6.5", default-features = false }
 tracing-subscriber = "0.2.15"
+tracing-log = "0.1.1"

client/service/src/builder.rs

@@ -55,6 +55,7 @@ use sc_telemetry::{
 	telemetry,
 	ConnectionMessage,
 	TelemetryConnectionNotifier,
+	TelemetrySpan,
 	SUBSTRATE_INFO,
 };
 use sp_transaction_pool::MaintainedTransactionPool;
@@ -307,7 +308,7 @@ pub fn new_full_parts<TBl, TRtApi, TExecDisp>(
 
 	let task_manager = {
 		let registry = config.prometheus_config.as_ref().map(|cfg| &cfg.registry);
-		TaskManager::new(config.task_executor.clone(), registry, config.telemetry_span.clone())?
+		TaskManager::new(config.task_executor.clone(), registry)?
 	};
 
 	let executor = NativeExecutor::<TExecDisp>::new(
@@ -376,7 +377,7 @@ pub fn new_light_parts<TBl, TRtApi, TExecDisp>(
 	let keystore_container = KeystoreContainer::new(&config.keystore)?;
 	let task_manager = {
 		let registry = config.prometheus_config.as_ref().map(|cfg| &cfg.registry);
-		TaskManager::new(config.task_executor.clone(), registry, config.telemetry_span.clone())?
+		TaskManager::new(config.task_executor.clone(), registry)?
 	};
 
 	let executor = NativeExecutor::<TExecDisp>::new(
@@ -490,6 +491,10 @@ pub struct SpawnTasksParams<'a, TBl: BlockT, TCl, TExPool, TRpc, Backend> {
 	pub network_status_sinks: NetworkStatusSinks<TBl>,
 	/// A Sender for RPC requests.
 	pub system_rpc_tx: TracingUnboundedSender<sc_rpc::system::Request<TBl>>,
+	/// Telemetry span.
+	///
+	/// This span needs to be entered **before** calling [`spawn_tasks()`].
+	pub telemetry_span: Option<TelemetrySpan>,
 }
 
 /// Build a shared offchain workers instance.
@@ -569,6 +574,7 @@ pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
 		network,
 		network_status_sinks,
 		system_rpc_tx,
+		telemetry_span,
 	} = params;
 
 	let chain_info = client.usage_info().chain;
@@ -581,6 +587,7 @@ pub fn spawn_tasks<TBl, TBackend, TExPool, TRpc, TCl>(
 
 	let telemetry_connection_notifier = init_telemetry(
 		&mut config,
+		telemetry_span,
 		network.clone(),
 		client.clone(),
 	);
@@ -681,10 +688,11 @@ async fn transaction_notifications<TBl, TExPool>(
 
 fn init_telemetry<TBl: BlockT, TCl: BlockBackend<TBl>>(
 	config: &mut Configuration,
+	telemetry_span: Option<TelemetrySpan>,
 	network: Arc<NetworkService<TBl, <TBl as BlockT>::Hash>>,
 	client: Arc<TCl>,
 ) -> Option<TelemetryConnectionNotifier> {
-	let telemetry_span = config.telemetry_span.clone()?;
+	let telemetry_span = telemetry_span?;
 	let endpoints = config.telemetry_endpoints.clone()?;
 	let genesis_hash = client.block_hash(Zero::zero()).ok().flatten().unwrap_or_default();
 	let connection_message = ConnectionMessage {

client/service/src/config.rs

@@ -101,10 +101,6 @@ pub struct Configuration {
 	/// This is a handle to a `TelemetryWorker` instance. It is used to initialize the telemetry for
 	/// a substrate node.
 	pub telemetry_handle: Option<sc_telemetry::TelemetryHandle>,
-	/// Telemetry span.
-	///
-	/// This span is entered for every background task spawned using the TaskManager.
-	pub telemetry_span: Option<sc_telemetry::TelemetrySpan>,
 	/// The default number of 64KB pages to allocate for Wasm execution
 	pub default_heap_pages: Option<u64>,
 	/// Should offchain workers be executed.

client/service/src/task_manager/mod.rs

@@ -24,7 +24,7 @@ use log::{debug, error};
 use futures::{
 	Future, FutureExt, StreamExt,
 	future::{select, Either, BoxFuture, join_all, try_join_all, pending},
-	sink::SinkExt, task::{Context, Poll},
+	sink::SinkExt,
 };
 use prometheus_endpoint::{
 	exponential_buckets, register,
@@ -34,51 +34,18 @@ use prometheus_endpoint::{
 use sp_utils::mpsc::{TracingUnboundedSender, TracingUnboundedReceiver, tracing_unbounded};
 use tracing_futures::Instrument;
 use crate::{config::{TaskExecutor, TaskType, JoinFuture}, Error};
-use sc_telemetry::TelemetrySpan;
 
 mod prometheus_future;
 #[cfg(test)]
 mod tests;
 
-/// A wrapper around a `[Option<TelemetrySpan>]` and a [`Future`].
-///
-/// The telemetry in Substrate uses a span to identify the telemetry context. The span "infrastructure"
-/// is provided by the tracing-crate. Now it is possible to have your own spans as well. To support
-/// this with the [`TaskManager`] we have this wrapper. This wrapper enters the telemetry span every
-/// time the future is polled and polls the inner future. So, the inner future can still have its
-/// own span attached and we get our telemetry span ;)
-struct WithTelemetrySpan<T> {
-	span: Option<TelemetrySpan>,
-	inner: T,
-}
-
-impl<T> WithTelemetrySpan<T> {
-	fn new(span: Option<TelemetrySpan>, inner: T) -> Self {
-		Self {
-			span,
-			inner,
-		}
-	}
-}
-
-impl<T: Future<Output = ()> + Unpin> Future for WithTelemetrySpan<T> {
-	type Output = ();
-
-	fn poll(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {
-		let span = self.span.clone();
-		let _enter = span.as_ref().map(|s| s.enter());
-		Pin::new(&mut self.inner).poll(ctx)
-	}
-}
-
 /// An handle for spawning tasks in the service.
 #[derive(Clone)]
 pub struct SpawnTaskHandle {
 	on_exit: exit_future::Exit,
 	executor: TaskExecutor,
 	metrics: Option<Metrics>,
 	task_notifier: TracingUnboundedSender<JoinFuture>,
-	telemetry_span: Option<TelemetrySpan>,
 }
 
 impl SpawnTaskHandle {
@@ -155,11 +122,7 @@ impl SpawnTaskHandle {
 			}
 		};
 
-		let future = future.in_current_span().boxed();
-		let join_handle = self.executor.spawn(
-			WithTelemetrySpan::new(self.telemetry_span.clone(), future).boxed(),
-			task_type,
-		);
+		let join_handle = self.executor.spawn(future.in_current_span().boxed(), task_type);
 
 		let mut task_notifier = self.task_notifier.clone();
 		self.executor.spawn(
@@ -266,8 +229,6 @@ pub struct TaskManager {
 	/// terminates and gracefully shutdown. Also ends the parent `future()` if a child's essential
 	/// task fails.
 	children: Vec<TaskManager>,
-	/// A `TelemetrySpan` used to enter the telemetry span when a task is spawned.
-	telemetry_span: Option<TelemetrySpan>,
 }
 
 impl TaskManager {
@@ -276,7 +237,6 @@ impl TaskManager {
 	pub(super) fn new(
 		executor: TaskExecutor,
 		prometheus_registry: Option<&Registry>,
-		telemetry_span: Option<TelemetrySpan>,
 	) -> Result<Self, PrometheusError> {
 		let (signal, on_exit) = exit_future::signal();
 
@@ -305,7 +265,6 @@ impl TaskManager {
 			task_notifier,
 			completion_future,
 			children: Vec::new(),
-			telemetry_span,
 		})
 	}
 
@@ -316,7 +275,6 @@ impl TaskManager {
 			executor: self.executor.clone(),
 			metrics: self.metrics.clone(),
 			task_notifier: self.task_notifier.clone(),
-			telemetry_span: self.telemetry_span.clone(),
 		}
 	}