add dispute metrics, some chores (#3842)

* rename: MsgFilter -> MessageInterceptor

* feat: add dispute metrics

* fixup

* test fixins

* fix metrics

* dummysubsystem export and trait fn fix

* chore: fmt

* undo unwanted changes

* foo

* pfmt

* fixup

* fixup

* revert

* some more

* Update node/malus/Cargo.toml

Co-authored-by: Andronik Ordian <[email protected]>

* Update node/core/dispute-coordinator/src/metrics.rs

Co-authored-by: Andronik Ordian <[email protected]>

* Update node/core/dispute-coordinator/src/metrics.rs

Co-authored-by: Andronik Ordian <[email protected]>

* Update node/core/dispute-coordinator/src/metrics.rs

Co-authored-by: Andronik Ordian <[email protected]>

* add license header

* fix lockfile

* new with opts

* fmt

* Update node/core/dispute-coordinator/src/metrics.rs

* feature gate

Co-authored-by: Andronik Ordian <[email protected]>

Author: drahnr

Cargo.lock

@@ -30,6 +30,8 @@ use kvdb::KeyValueDB;
 use parity_scale_codec::{Decode, Encode, Error as CodecError};
 use sc_keystore::LocalKeystore;
 
+use crate::metrics::Metrics;
+
 const LOG_TARGET: &str = "parachain::dispute-coordinator";
 
 /// Timestamp based on the 1 Jan 1970 UNIX base, which is persistent across node restarts and OS reboots.
@@ -52,7 +54,7 @@ pub struct DisputeCoordinatorSubsystem {}
 
 impl DisputeCoordinatorSubsystem {
 	/// Create a new instance of the subsystem.
-	pub fn new(_: Arc<dyn KeyValueDB>, _: Config, _: Arc<LocalKeystore>) -> Self {
+	pub fn new(_: Arc<dyn KeyValueDB>, _: Config, _: Arc<LocalKeystore>, _: Metrics) -> Self {
 		DisputeCoordinatorSubsystem {}
 	}
 }

node/core/dispute-coordinator/src/dummy.rs

@@ -25,6 +25,8 @@
 //! another node, this will trigger the dispute participation subsystem to recover and validate the block and call
 //! back to this subsystem.
 
+mod metrics;
+
 #[cfg(feature = "disputes")]
 mod real;
 #[cfg(feature = "disputes")]

node/core/dispute-coordinator/src/lib.rs

@@ -0,0 +1,99 @@
+// Copyright 2020 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot 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.
+
+// Polkadot 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 Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+use polkadot_node_subsystem_util::metrics::{self, prometheus};
+
+#[derive(Clone)]
+struct MetricsInner {
+	/// Number of opened disputes.
+	open: prometheus::Counter<prometheus::U64>,
+	/// Votes of all disputes.
+	votes: prometheus::CounterVec<prometheus::U64>,
+	/// Conclusion across all disputes.
+	concluded: prometheus::CounterVec<prometheus::U64>,
+}
+
+/// Candidate validation metrics.
+#[derive(Default, Clone)]
+pub struct Metrics(Option<MetricsInner>);
+
+#[cfg(feature = "disputes")]
+impl Metrics {
+	pub(crate) fn on_open(&self) {
+		if let Some(metrics) = &self.0 {
+			metrics.open.inc();
+		}
+	}
+
+	pub(crate) fn on_valid_vote(&self) {
+		if let Some(metrics) = &self.0 {
+			metrics.votes.with_label_values(&["valid"]).inc();
+		}
+	}
+
+	pub(crate) fn on_invalid_vote(&self) {
+		if let Some(metrics) = &self.0 {
+			metrics.votes.with_label_values(&["invalid"]).inc();
+		}
+	}
+
+	pub(crate) fn on_concluded_valid(&self) {
+		if let Some(metrics) = &self.0 {
+			metrics.concluded.with_label_values(&["valid"]).inc();
+		}
+	}
+
+	pub(crate) fn on_concluded_invalid(&self) {
+		if let Some(metrics) = &self.0 {
+			metrics.concluded.with_label_values(&["invalid"]).inc();
+		}
+	}
+}
+
+impl metrics::Metrics for Metrics {
+	fn try_register(registry: &prometheus::Registry) -> Result<Self, prometheus::PrometheusError> {
+		let metrics = MetricsInner {
+			open: prometheus::register(
+				prometheus::Counter::with_opts(prometheus::Opts::new(
+					"parachain_candidate_disputes_total",
+					"Total number of raised disputes.",
+				))?,
+				registry,
+			)?,
+			concluded: prometheus::register(
+				prometheus::CounterVec::new(
+					prometheus::Opts::new(
+						"parachain_candidate_dispute_concluded",
+						"Concluded dispute votes, sorted by candidate is `valid` and `invalid`.",
+					),
+					&["validity"],
+				)?,
+				registry,
+			)?,
+			votes: prometheus::register(
+				prometheus::CounterVec::new(
+					prometheus::Opts::new(
+						"parachain_candidate_dispute_votes",
+						"Accumulated dispute votes, sorted by candidate is `valid` and `invalid`.",
+					),
+					&["validity"],
+				)?,
+				registry,
+			)?,
+		};
+		Ok(Metrics(Some(metrics)))
+	}
+}

node/core/dispute-coordinator/src/metrics.rs

@@ -56,6 +56,7 @@ use kvdb::KeyValueDB;
 use parity_scale_codec::{Decode, Encode, Error as CodecError};
 use sc_keystore::LocalKeystore;
 
+use crate::metrics::Metrics;
 use backend::{Backend, OverlayedBackend};
 use db::v1::{DbBackend, RecentDisputes};
 
@@ -116,12 +117,18 @@ pub struct DisputeCoordinatorSubsystem {
 	config: Config,
 	store: Arc<dyn KeyValueDB>,
 	keystore: Arc<LocalKeystore>,
+	metrics: Metrics,
 }
 
 impl DisputeCoordinatorSubsystem {
 	/// Create a new instance of the subsystem.
-	pub fn new(store: Arc<dyn KeyValueDB>, config: Config, keystore: Arc<LocalKeystore>) -> Self {
-		DisputeCoordinatorSubsystem { store, config, keystore }
+	pub fn new(
+		store: Arc<dyn KeyValueDB>,
+		config: Config,
+		keystore: Arc<LocalKeystore>,
+		metrics: Metrics,
+	) -> Self {
+		DisputeCoordinatorSubsystem { store, config, keystore, metrics }
 	}
 }
 
@@ -329,6 +336,7 @@ where
 		rolling_session_window: RollingSessionWindow::new(DISPUTE_WINDOW),
 		recovery_state: Participation::Pending,
 	};
+	let metrics = &subsystem.metrics;
 
 	loop {
 		let mut overlay_db = OverlayedBackend::new(backend);
@@ -348,7 +356,8 @@ where
 			},
 			FromOverseer::Signal(OverseerSignal::BlockFinalized(_, _)) => {},
 			FromOverseer::Communication { msg } =>
-				handle_incoming(ctx, &mut overlay_db, &mut state, msg, clock.now()).await?,
+				handle_incoming(ctx, &mut overlay_db, &mut state, msg, clock.now(), &metrics)
+					.await?,
 		}
 
 		if !overlay_db.is_empty() {
@@ -518,6 +527,7 @@ async fn handle_incoming(
 	state: &mut State,
 	message: DisputeCoordinatorMessage,
 	now: Timestamp,
+	metrics: &Metrics,
 ) -> Result<(), Error> {
 	match message {
 		DisputeCoordinatorMessage::ImportStatements {
@@ -537,6 +547,7 @@ async fn handle_incoming(
 				statements,
 				now,
 				pending_confirmation,
+				metrics,
 			)
 			.await?;
 		},
@@ -578,6 +589,7 @@ async fn handle_incoming(
 				session,
 				valid,
 				now,
+				metrics,
 			)
 			.await?;
 		},
@@ -635,6 +647,7 @@ async fn handle_import_statements(
 	statements: Vec<(SignedDisputeStatement, ValidatorIndex)>,
 	now: Timestamp,
 	pending_confirmation: oneshot::Sender<ImportStatementsResult>,
+	metrics: &Metrics,
 ) -> Result<(), Error> {
 	if state.highest_session.map_or(true, |h| session + DISPUTE_WINDOW < h) {
 		// It is not valid to participate in an ancient dispute (spam?).
@@ -694,6 +707,7 @@ async fn handle_import_statements(
 
 		match statement.statement().clone() {
 			DisputeStatement::Valid(valid_kind) => {
+				metrics.on_valid_vote();
 				insert_into_statement_vec(
 					&mut votes.valid,
 					valid_kind,
@@ -702,6 +716,7 @@ async fn handle_import_statements(
 				);
 			},
 			DisputeStatement::Invalid(invalid_kind) => {
+				metrics.on_invalid_vote();
 				insert_into_statement_vec(
 					&mut votes.invalid,
 					invalid_kind,
@@ -784,6 +799,14 @@ async fn handle_import_statements(
 				);
 				return Ok(())
 			}
+			metrics.on_open();
+
+			if concluded_valid {
+				metrics.on_concluded_valid();
+			}
+			if concluded_invalid {
+				metrics.on_concluded_invalid();
+			}
 		}
 
 		// Only write when updated and vote is available.
@@ -824,6 +847,7 @@ async fn issue_local_statement(
 	session: SessionIndex,
 	valid: bool,
 	now: Timestamp,
+	metrics: &Metrics,
 ) -> Result<(), Error> {
 	// Load session info.
 	let info = match state.rolling_session_window.session_info(session) {
@@ -857,7 +881,6 @@ async fn issue_local_statement(
 
 	let voted_indices: HashSet<_> = voted_indices.into_iter().collect();
 	let controlled_indices = find_controlled_validator_indices(&state.keystore, &validators[..]);
-
 	for index in controlled_indices {
 		if voted_indices.contains(&index) {
 			continue
@@ -914,6 +937,7 @@ async fn issue_local_statement(
 			statements,
 			now,
 			pending_confirmation,
+			metrics,
 		)
 		.await?;
 		match rx.await {

node/core/dispute-coordinator/src/real/mod.rs

@@ -286,6 +286,7 @@ impl TestState {
 			self.db.clone(),
 			self.config.clone(),
 			self.subsystem_keystore.clone(),
+			Metrics::default(),
 		);
 		let backend = DbBackend::new(self.db.clone(), self.config.column_config());
 		let subsystem_task = run(subsystem, ctx, backend, Box::new(self.clock.clone()));

node/core/dispute-coordinator/src/real/tests.rs

@@ -27,3 +27,8 @@ color-eyre = { version = "0.5.11", default-features = false }
 assert_matches = "1.5"
 structopt = "0.3.23"
 async-trait = "0.1.51"
+
+[dev-dependencies]
+polkadot-node-subsystem-test-helpers = { path = "../subsystem-test-helpers" }
+sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
+futures = { version = "0.3.17", features = ["thread-pool"] }