Adjust consensus telemetry

core/client/src/client.rs

@@ -956,6 +956,11 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
 			let header = self.header(&BlockId::Hash(finalized_hash))?
 				.expect("header already known to exist in DB because it is indicated in the tree route; qed");
 
+			telemetry!(SUBSTRATE_INFO; "notify.finalized";
+				"height" => format!("{}", header.number()),
+				"best" => ?finalized_hash,
+			);
+
 			let notification = FinalityNotification {
 				header,
 				hash: finalized_hash,

core/finality-grandpa/Cargo.toml

@@ -16,6 +16,7 @@ runtime_primitives = { package = "sr-primitives", path = "../sr-primitives" }
 consensus_common = { package = "substrate-consensus-common", path = "../consensus/common" }
 substrate-primitives = { path = "../primitives" }
 substrate-telemetry = { path = "../telemetry" }
+serde_json = "1.0"
 client = { package = "substrate-client", path = "../client" }
 inherents = { package = "substrate-inherents", path = "../../core/inherents" }
 network = { package = "substrate-network", path = "../network" }

core/finality-grandpa/src/aux_schema.rs

@@ -25,6 +25,7 @@ use fork_tree::ForkTree;
 use grandpa::round::State as RoundState;
 use runtime_primitives::traits::{Block as BlockT, NumberFor};
 use log::{info, warn};
+use substrate_telemetry::{telemetry, CONSENSUS_INFO};
 
 use crate::authorities::{AuthoritySet, SharedAuthoritySet, PendingChange, DelayKind};
 use crate::consensus_changes::{SharedConsensusChanges, ConsensusChanges};
@@ -365,6 +366,17 @@ pub(crate) fn update_authority_set<Block: BlockT, F, R>(
 	let encoded_set = set.encode();
 
 	if let Some(new_set) = new_set {
+		telemetry!(CONSENSUS_INFO; "afg.authority_set";
+			"hash" => ?new_set.canon_hash,
+			"number" => ?new_set.canon_number,
+			"authority_set_id" => ?new_set.set_id,
+			"authorities" => {
+				let authorities: Vec<String> =
+					new_set.authorities.iter().map(|(id, _)| format!("{}", id)).collect();
+				format!("{:?}", authorities)
+			}
+		);
+
 		// we also overwrite the "last completed round" entry with a blank slate
 		// because from the perspective of the finality gadget, the chain has
 		// reset.

core/finality-grandpa/src/communication/mod.rs

@@ -681,7 +681,7 @@ impl<Block: BlockT, N: Network<Block>> Sink for CommitsOut<Block, N> {
 		let (round, commit) = input;
 		let round = Round(round);
 
-		telemetry!(CONSENSUS_INFO; "afg.commit_issued";
+		telemetry!(CONSENSUS_DEBUG; "afg.commit_issued";
 			"target_number" => ?commit.target_number, "target_hash" => ?commit.target_hash,
 		);
 		let (precommits, auth_data) = commit.precommits.into_iter()

core/finality-grandpa/src/lib.rs

@@ -69,6 +69,7 @@ use inherents::InherentDataProviders;
 use runtime_primitives::generic::BlockId;
 use substrate_primitives::{ed25519, H256, Pair, Blake2Hasher};
 use substrate_telemetry::{telemetry, CONSENSUS_INFO, CONSENSUS_DEBUG, CONSENSUS_WARN};
+use serde_json;
 
 use srml_finality_tracker;
 
@@ -105,6 +106,7 @@ use environment::{CompletedRound, CompletedRounds, Environment, HasVoted, Shared
 use import::GrandpaBlockImport;
 use until_imported::UntilCommitBlocksImported;
 use communication::NetworkBridge;
+use service::TelemetryOnConnect;
 
 use ed25519::{Public as AuthorityId, Signature as AuthoritySignature};
 
@@ -433,14 +435,26 @@ fn register_finality_tracker_inherent_data_provider<B, E, Block: BlockT<Hash=H25
 	}
 }
 
+/// Parameters used to run Grandpa.
+pub struct GrandpaParams<'a, B, E, Block: BlockT<Hash=H256>, N, RA, X> {
+	/// Configuration for the GRANDPA service.
+	pub config: Config,
+	/// A link to the block import worker.
+	pub link: LinkHalf<B, E, Block, RA>,
+	/// The Network instance.
+	pub network: N,
+	/// The inherent data providers.
+	pub inherent_data_providers: InherentDataProviders,
+	/// Handle to a future that will resolve on exit.
+	pub on_exit: X,
+	/// If supplied, can be used to hook on telemetry connection established events.
+	pub telemetry_on_connect: Option<TelemetryOnConnect<'a>>,
+}
+
 /// Run a GRANDPA voter as a task. Provide configuration and a link to a
 /// block import worker that has already been instantiated with `block_import`.
-pub fn run_grandpa_voter<B, E, Block: BlockT<Hash=H256>, N, RA>(
-	config: Config,
-	link: LinkHalf<B, E, Block, RA>,
-	network: N,
-	inherent_data_providers: InherentDataProviders,
-	on_exit: impl Future<Item=(),Error=()> + Clone + Send + 'static,
+pub fn run_grandpa_voter<B, E, Block: BlockT<Hash=H256>, N, RA, X>(
+	grandpa_params: GrandpaParams<B, E, Block, N, RA, X>,
 ) -> ::client::error::Result<impl Future<Item=(),Error=()> + Send + 'static> where
 	Block::Hash: Ord,
 	B: Backend<Block, Blake2Hasher> + 'static,
@@ -451,7 +465,17 @@ pub fn run_grandpa_voter<B, E, Block: BlockT<Hash=H256>, N, RA>(
 	DigestFor<Block>: Encode,
 	DigestItemFor<Block>: DigestItem<AuthorityId=AuthorityId>,
 	RA: Send + Sync + 'static,
+	X: Future<Item=(),Error=()> + Clone + Send + 'static,
 {
+	let GrandpaParams {
+		config,
+		link,
+		network,
+		inherent_data_providers,
+		on_exit,
+		telemetry_on_connect,
+	} = grandpa_params;
+
 	use futures::future::{self, Loop as FutureLoop};
 
 	let (network, network_startup) = NetworkBridge::new(network, config.clone(), on_exit.clone());
@@ -465,6 +489,28 @@ pub fn run_grandpa_voter<B, E, Block: BlockT<Hash=H256>, N, RA>(
 
 	register_finality_tracker_inherent_data_provider(client.clone(), &inherent_data_providers)?;
 
+	if let Some(telemetry_on_connect) = telemetry_on_connect {
+		let authorities = authority_set.clone();
+		let events = telemetry_on_connect.telemetry_connection_sinks
+			.for_each(move |_| {
+				telemetry!(CONSENSUS_INFO; "afg.authority_set";
+					 "authority_set_id" => ?authorities.set_id(),
+					 "authorities" => {
+						let curr = authorities.current_authorities();
+						let voters = curr.voters();
+						let authorities: Vec<String> =
+							voters.iter().map(|(id, _)| id.to_string()).collect();
+						serde_json::to_string(&authorities)
+							.expect("authorities is always at least an empty vector; elements are always of type string")
+					 }
+				);
+				Ok(())
+			})
+			.then(|_| Ok(()));
+		let events = events.select(telemetry_on_connect.on_exit).then(|_| Ok(()));
+		telemetry_on_connect.executor.spawn(events);
+	}
+
 	let voters = authority_set.current_authorities();
 	let initial_environment = Arc::new(Environment {
 		inner: client.clone(),
@@ -660,12 +706,8 @@ pub fn run_grandpa_voter<B, E, Block: BlockT<Hash=H256>, N, RA>(
 }
 
 #[deprecated(since = "1.1", note = "Please switch to run_grandpa_voter.")]
-pub fn run_grandpa<B, E, Block: BlockT<Hash=H256>, N, RA>(
-	config: Config,
-	link: LinkHalf<B, E, Block, RA>,
-	network: N,
-	inherent_data_providers: InherentDataProviders,
-	on_exit: impl Future<Item=(),Error=()> + Clone + Send + 'static,
+pub fn run_grandpa<B, E, Block: BlockT<Hash=H256>, N, RA, X>(
+	grandpa_params: GrandpaParams<B, E, Block, N, RA, X>,
 ) -> ::client::error::Result<impl Future<Item=(),Error=()> + Send + 'static> where
 	Block::Hash: Ord,
 	B: Backend<Block, Blake2Hasher> + 'static,
@@ -676,6 +718,7 @@ pub fn run_grandpa<B, E, Block: BlockT<Hash=H256>, N, RA>(
 	DigestFor<Block>: Encode,
 	DigestItemFor<Block>: DigestItem<AuthorityId=AuthorityId>,
 	RA: Send + Sync + 'static,
+	X: Future<Item=(),Error=()> + Clone + Send + 'static,
 {
-	run_grandpa_voter(config, link, network, inherent_data_providers, on_exit)
+	run_grandpa_voter(grandpa_params)
 }

core/finality-grandpa/src/tests.rs

@@ -419,18 +419,20 @@ fn run_to_completion_with<F>(
 
 		fn assert_send<T: Send>(_: &T) { }
 
-		let voter = run_grandpa_voter(
-			Config {
+		let grandpa_params = GrandpaParams {
+			config: Config {
 				gossip_duration: TEST_GOSSIP_DURATION,
 				justification_period: 32,
 				local_key: Some(Arc::new(key.clone().into())),
 				name: Some(format!("peer#{}", peer_id)),
 			},
-			link,
-			MessageRouting::new(net.clone(), peer_id),
-			InherentDataProviders::new(),
-			Exit,
-		).expect("all in order with client and network");
+			link: link,
+			network: MessageRouting::new(net.clone(), peer_id),
+			inherent_data_providers: InherentDataProviders::new(),
+			on_exit: Exit,
+			telemetry_on_connect: None,
+		};
+		let voter = run_grandpa_voter(grandpa_params).expect("all in order with client and network");
 
 		assert_send(&voter);
 
@@ -517,18 +519,21 @@ fn finalize_3_voters_1_full_observer() {
 				.take_while(|n| Ok(n.header.number() < &20))
 				.for_each(move |_| Ok(()))
 		);
-		let voter = run_grandpa_voter(
-			Config {
+
+		let grandpa_params = GrandpaParams {
+			config: Config {
 				gossip_duration: TEST_GOSSIP_DURATION,
 				justification_period: 32,
 				local_key,
 				name: Some(format!("peer#{}", peer_id)),
 			},
-			link,
-			MessageRouting::new(net.clone(), peer_id),
-			InherentDataProviders::new(),
-			Exit,
-		).expect("all in order with client and network");
+			link: link,
+			network: MessageRouting::new(net.clone(), peer_id),
+			inherent_data_providers: InherentDataProviders::new(),
+			on_exit: Exit,
+			telemetry_on_connect: None,
+		};
+		let voter = run_grandpa_voter(grandpa_params).expect("all in order with client and network");
 
 		runtime.spawn(voter);
 	}
@@ -679,18 +684,20 @@ fn transition_3_voters_twice_1_full_observer() {
 					assert_eq!(set.pending_changes().count(), 0);
 				})
 		);
-		let voter = run_grandpa_voter(
-			Config {
+		let grandpa_params = GrandpaParams {
+			config: Config {
 				gossip_duration: TEST_GOSSIP_DURATION,
 				justification_period: 32,
 				local_key,
 				name: Some(format!("peer#{}", peer_id)),
 			},
-			link,
-			MessageRouting::new(net.clone(), peer_id),
-			InherentDataProviders::new(),
-			Exit,
-		).expect("all in order with client and network");
+			link: link,
+			network: MessageRouting::new(net.clone(), peer_id),
+			inherent_data_providers: InherentDataProviders::new(),
+			on_exit: Exit,
+			telemetry_on_connect: None,
+		};
+		let voter = run_grandpa_voter(grandpa_params).expect("all in order with client and network");
 
 		runtime.spawn(voter);
 	}
@@ -1081,18 +1088,20 @@ fn voter_persists_its_votes() {
 			let (_block_import, _, link) = net.lock().make_block_import(client.clone());
 			let link = link.lock().take().unwrap();
 
-			let mut voter = run_grandpa_voter(
-				Config {
+			let grandpa_params = GrandpaParams {
+				config: Config {
 					gossip_duration: TEST_GOSSIP_DURATION,
 					justification_period: 32,
 					local_key: Some(Arc::new(peers[0].clone().into())),
 					name: Some(format!("peer#{}", 0)),
 				},
-				link,
-				MessageRouting::new(net.clone(), 0),
-				InherentDataProviders::new(),
-				Exit,
-			).expect("all in order with client and network");
+				link: link,
+				network: MessageRouting::new(net.clone(), 0),
+				inherent_data_providers: InherentDataProviders::new(),
+				on_exit: Exit,
+				telemetry_on_connect: None,
+			};
+			let mut voter = run_grandpa_voter(grandpa_params).expect("all in order with client and network");
 
 			let voter = future::poll_fn(move || {
 				// we need to keep the block_import alive since it owns the

core/peerset/src/lib.rs

@@ -22,6 +22,7 @@ mod slots;
 use std::collections::VecDeque;
 use futures::{prelude::*, sync::mpsc, try_ready};
 use libp2p::PeerId;
+use linked_hash_map::LinkedHashMap;
 use log::trace;
 use lru_cache::LruCache;
 use slots::{SlotType, SlotState, Slots};