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!(
@@ -162,6 +163,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(),
@@ -176,6 +180,7 @@ pub fn new_full(mut config: Configuration) -> Result<TaskManager, ServiceError>
 			network_status_sinks,
 			system_rpc_tx,
 			config,
+			telemetry_span: Some(telemetry_span.clone()),
 		},
 	)?;
 
@@ -312,6 +317,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,
@@ -325,6 +333,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/Cargo.toml

@@ -124,6 +124,8 @@ nix = "0.19"
 serde_json = "1.0"
 regex = "1"
 platforms = "1.1"
+async-std = { version = "1.6.5", features = ["attributes"] }
+soketto = "0.4.2"
 
 [build-dependencies]
 structopt = { version = "0.3.8", optional = true }

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>;
@@ -226,6 +226,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,
@@ -240,6 +243,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()),
 		},
 	)?;
 
@@ -433,6 +437,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),
@@ -444,6 +451,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((

bin/node/cli/tests/telemetry.rs

@@ -0,0 +1,102 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program 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.
+
+// This program 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 this program. If not, see <https://www.gnu.org/licenses/>.
+
+use assert_cmd::cargo::cargo_bin;
+use nix::sys::signal::{kill, Signal::SIGINT};
+use nix::unistd::Pid;
+use std::convert::TryInto;
+use std::process;
+
+pub mod common;
+pub mod websocket_server;
+
+#[async_std::test]
+async fn telemetry_works() {
+	let config = websocket_server::Config {
+		capacity: 1,
+		max_frame_size: 1048 * 1024,
+		send_buffer_len: 32,
+		bind_address: "127.0.0.1:0".parse().unwrap(),
+	};
+	let mut server = websocket_server::WsServer::new(config).await.unwrap();
+
+	let addr = server.local_addr().unwrap();
+
+	let server_task = async_std::task::spawn(async move {
+		loop {
+			use websocket_server::Event;
+			match server.next_event().await {
+				// New connection on the listener.
+				Event::ConnectionOpen { address } => {
+					println!("New connection from {:?}", address);
+					server.accept();
+				}
+
+				// Received a message from a connection.
+				Event::BinaryFrame { message, .. } => {
+					let json: serde_json::Value = serde_json::from_slice(&message).unwrap();
+					let object = json
+						.as_object()
+						.unwrap()
+						.get("payload")
+						.unwrap()
+						.as_object()
+						.unwrap();
+					if matches!(object.get("best"), Some(serde_json::Value::String(_))) {
+						break;
+					}
+				}
+
+				Event::TextFrame { .. } => panic!("Got a TextFrame over the socket, this is a bug"),
+
+				// Connection has been closed.
+				Event::ConnectionError { .. } => {}
+			}
+		}
+	});
+
+	let mut substrate = process::Command::new(cargo_bin("substrate"));
+
+	let mut substrate = substrate
+		.args(&["--dev", "--tmp", "--telemetry-url"])
+		.arg(format!("ws://{} 10", addr))
+		.stdout(process::Stdio::piped())
+		.stderr(process::Stdio::piped())
+		.stdin(process::Stdio::null())
+		.spawn()
+		.unwrap();
+
+	server_task.await;
+
+	assert!(
+		substrate.try_wait().unwrap().is_none(),
+		"the process should still be running"
+	);
+
+	// Stop the process
+	kill(Pid::from_raw(substrate.id().try_into().unwrap()), SIGINT).unwrap();
+	assert!(common::wait_for(&mut substrate, 40)
+		.map(|x| x.success())
+		.unwrap_or_default());
+
+	let output = substrate.wait_with_output().unwrap();
+
+	println!("{}", String::from_utf8(output.stdout).unwrap());
+	eprintln!("{}", String::from_utf8(output.stderr).unwrap());
+	assert!(output.status.success());
+}