Merge pull request #4199 from ProvableHQ/feat/individual-participation-metrics

[Feature] Export individual validator participation scores as gauges

Author: vicsn

Cargo.lock

@@ -1031,13 +1031,17 @@ impl<N: Network> Gateway<N> {
     // Logs the validator participation scores.
     #[cfg(feature = "telemetry")]
     fn log_participation_scores(&self) {
-        if let Ok(current_committee) = self.ledger.current_committee() {
+        if let Ok(committee_lookback) = self.ledger.get_committee_lookback_for_round(self.storage.current_round()) {
             // Retrieve the participation scores.
-            let participation_scores = self.validator_telemetry().get_participation_scores(&current_committee);
+            let participation_scores = self.validator_telemetry().get_participation_scores(&committee_lookback);
+
             // Log the participation scores.
             debug!("Participation Scores (in the last {} rounds):", self.storage.max_gc_rounds());
-            for (address, score) in participation_scores {
-                debug!("{}", format!("  {address} - {score:.2}%").dimmed());
+            for (address, (cert_score, sig_score)) in participation_scores {
+                debug!(
+                    "{}",
+                    format!("  {address} - certificates: {cert_score:.2}%  signatures: {sig_score:.2}%").dimmed()
+                );
             }
         }
     }

node/bft/src/gateway.rs

@@ -84,19 +84,21 @@ impl<N: Network> Telemetry<N> {
         }
     }
 
-    // TODO (raychu86): Consider using committee lookback here.
-    /// Fetch the participation scores for each validator in the committee set.
-    pub fn get_participation_scores(&self, committee: &Committee<N>) -> IndexMap<Address<N>, f64> {
+    /// Fetch the certificate and signature participation scores for each validator in the committee set.
+    /// Returns a map of `address` to `(certificate_score, signature_score)`.
+    pub fn get_participation_scores(&self, committee: &Committee<N>) -> IndexMap<Address<N>, (f64, f64)> {
         // Fetch the participation scores.
         let participation_scores = self.participation_scores.read();
-        // Calculate the weighted score for each validator.
+        // Return the individual scores for each validator.
         committee
             .members()
             .iter()
             .map(|(address, _)| {
-                let score =
-                    participation_scores.get(address).map(|(_, _, combined_score)| *combined_score).unwrap_or(0.0);
-                (*address, score)
+                let scores = participation_scores
+                    .get(address)
+                    .map(|(cert_score, sig_score, _)| (*cert_score, *sig_score))
+                    .unwrap_or((0.0, 0.0));
+                (*address, scores)
             })
             .collect()
     }
@@ -326,7 +328,7 @@ mod tests {
         let participation_scores = telemetry.get_participation_scores(&committee);
         assert_eq!(participation_scores.len(), committee.members().len());
         for (address, _) in committee.members() {
-            assert_eq!(*participation_scores.get(address).unwrap(), 0.0);
+            assert_eq!(*participation_scores.get(address).unwrap(), (0.0, 0.0));
         }
 
         // Calculate the participation scores.
@@ -335,7 +337,8 @@ mod tests {
         // Ensure that the participation scores are updated.
         let participation_scores = telemetry.get_participation_scores(&committee);
         for (address, _) in committee.members() {
-            assert!(*participation_scores.get(address).unwrap() > 0.0);
+            let (cert_score, sig_score) = *participation_scores.get(address).unwrap();
+            assert!(cert_score > 0.0 || sig_score > 0.0);
         }
 
         println!("{participation_scores:?}");

node/bft/src/helpers/telemetry.rs

@@ -594,10 +594,10 @@ impl<N: Network> Consensus<N> {
         ledger_update.advance_to_next_block(&block)?;
 
         #[cfg(feature = "telemetry")]
-        // Fetch the latest committee.
+        // Fetch the committee lookback for the latest round.
         // Note: Do not abort here if this returns an error, because the committee is only needed for telemetry,
         // not for block advancement itself.
-        let latest_committee = self.ledger.current_committee();
+        let latest_committee = self.ledger.get_committee_lookback_for_round(self.ledger.latest_round());
 
         // If the next block starts a new epoch, clear the existing solutions.
         if block_height.is_multiple_of(N::NUM_BLOCKS_PER_EPOCH) {
@@ -650,22 +650,29 @@ impl<N: Network> Consensus<N> {
             #[cfg(feature = "telemetry")]
             match latest_committee {
                 Ok(latest_committee) => {
-                    // Retrieve the latest participation scores.
+                    // Retrieve the individual participation scores.
                     let participation_scores = self
                         .bft()
                         .primary()
                         .gateway()
                         .validator_telemetry()
                         .get_participation_scores(&latest_committee);
 
-                    // Log the participation scores.
-                    for (address, participation_score) in participation_scores {
-                        metrics::histogram_label(
-                            metrics::consensus::VALIDATOR_PARTICIPATION,
+                    // Export the certificate and signature participation scores as individual gauges.
+                    for (address, (certificate_score, signature_score)) in participation_scores {
+                        let address_str = address.to_string();
+                        metrics::gauge_label(
+                            metrics::consensus::VALIDATOR_CERTIFICATE_PARTICIPATION,
                             "validator_address",
-                            address.to_string(),
-                            participation_score,
-                        )
+                            address_str.clone(),
+                            certificate_score,
+                        );
+                        metrics::gauge_label(
+                            metrics::consensus::VALIDATOR_SIGNATURE_PARTICIPATION,
+                            "validator_address",
+                            address_str,
+                            signature_score,
+                        );
                     }
                 }
                 Err(err) => warn!("{}", flatten_error(err.context("Failed to get latest committee for telemetry"))),

node/consensus/src/lib.rs

@@ -164,6 +164,10 @@ pub fn add_transmission_latency_metric<N: Network>(
     }
 }
 
+pub fn gauge_label<V: Into<f64>>(name: &'static str, label_key: &'static str, label_value: String, value: V) {
+    ::metrics::gauge!(name, label_key => label_value).set(value.into());
+}
+
 // Include the generated build information
 mod built_info {
     include!(concat!(env!("OUT_DIR"), "/built.rs"));

node/metrics/src/lib.rs

@@ -104,7 +104,8 @@ pub mod consensus {
     pub const TRANSMISSION_LATENCY: &str = "snarkos_consensus_transmission_latency";
     pub const STALE_UNCONFIRMED_TRANSACTIONS: &str = "snarkos_consensus_stale_unconfirmed_transactions";
     pub const STALE_UNCONFIRMED_SOLUTIONS: &str = "snarkos_consensus_stale_unconfirmed_solutions";
-    pub const VALIDATOR_PARTICIPATION: &str = "snarkos_consensus_validator_participation";
+    pub const VALIDATOR_CERTIFICATE_PARTICIPATION: &str = "snarkos_consensus_validator_certificate_participation";
+    pub const VALIDATOR_SIGNATURE_PARTICIPATION: &str = "snarkos_consensus_validator_signature_participation";
 }
 
 pub mod router {

node/metrics/src/names.rs

@@ -1048,15 +1048,25 @@ impl<N: Network, C: ConsensusStorage<N>, R: Routing<N>> Rest<N, C, R> {
     ) -> Result<impl axum::response::IntoResponse, RestError> {
         match rest.consensus {
             Some(consensus) => {
-                // Retrieve the latest committee.
-                let latest_committee = rest.ledger.latest_committee()?;
-                // Retrieve the latest participation scores.
-                let participation_scores = consensus
+                // Retrieve the committee lookback for the latest round.
+                let latest_round = rest.ledger.latest_round();
+                let committee_lookback = rest
+                    .ledger
+                    .get_committee_lookback_for_round(latest_round)?
+                    .ok_or_else(|| RestError::not_found(anyhow!("No committee found for round {latest_round}")))?;
+                // Retrieve the latest participation scores, combining certificate and signature scores.
+                let participation_scores: IndexMap<_, _> = consensus
                     .bft()
                     .primary()
                     .gateway()
                     .validator_telemetry()
-                    .get_participation_scores(&latest_committee);
+                    .get_participation_scores(&committee_lookback)
+                    .into_iter()
+                    .map(|(address, (cert_score, sig_score))| {
+                        let combined = ((0.9 * cert_score + 0.1 * sig_score) * 100.0).round() / 100.0;
+                        (address, combined)
+                    })
+                    .collect();
 
                 // Check if metadata is requested and return the participation scores with metadata if so.
                 if metadata.metadata.unwrap_or(false) {