Unify participation metrics

Author: paulhauner

Cargo.lock

@@ -95,7 +95,7 @@ cargo-fmt:
 check-benches:
 	cargo check --workspace --benches
 
-# Typechecks consensus code *without* allowing deprecated legacy arithmetic
+# Typechecks consensus code *without* allowing deprecated legacy arithmetic or metrics.
 check-consensus:
 	cargo check --manifest-path=consensus/state_processing/Cargo.toml --no-default-features
 

Makefile

@@ -58,14 +58,11 @@ use slot_clock::SlotClock;
 use ssz::Encode;
 use state_processing::{
     block_signature_verifier::{BlockSignatureVerifier, Error as BlockSignatureVerifierError},
-    per_block_processing,
-    per_epoch_processing::EpochProcessingSummary,
-    per_slot_processing,
+    per_block_processing, per_slot_processing,
     state_advance::partial_state_advance,
     BlockProcessingError, BlockSignatureStrategy, SlotProcessingError,
 };
 use std::borrow::Cow;
-use std::convert::TryFrom;
 use std::fs;
 use std::io::Write;
 use store::{Error as DBError, HotColdDB, HotStateSummary, KeyValueStore, StoreOp};
@@ -971,12 +968,19 @@ impl<'a, T: BeaconChainTypes> FullyVerifiedBlock<'a, T> {
             };
 
             if let Some(summary) = per_slot_processing(&mut state, Some(state_root), &chain.spec)? {
-                summaries.push(summary)
+                // Expose Prometheus metrics.
+                if let Err(e) = summary.observe_metrics() {
+                    error!(
+                        chain.log,
+                        "Failed to observe epoch summary metrics";
+                        "src" => "block_verification",
+                        "error" => ?e
+                    );
+                }
+                summaries.push(summary);
             }
         }
 
-        expose_participation_metrics(&summaries, &chain.log);
-
         // If the block is sufficiently recent, notify the validator monitor.
         if let Some(slot) = chain.slot_clock.now() {
             let epoch = slot.epoch(T::EthSpec::slots_per_epoch());
@@ -1440,62 +1444,6 @@ fn verify_header_signature<T: BeaconChainTypes>(
     }
 }
 
-fn expose_participation_metrics(summaries: &[EpochProcessingSummary], log: &Logger) {
-    if !cfg!(feature = "participation_metrics") {
-        return;
-    }
-
-    for summary in summaries {
-        match summary.previous_epoch_target_attesting_balance() {
-            Ok(target_balance) => {
-                metrics::maybe_set_float_gauge(
-                    &metrics::PARTICIPATION_PREV_EPOCH_TARGET_ATTESTER,
-                    participation_ratio(
-                        target_balance,
-                        summary.previous_epoch_total_active_balance(),
-                    ),
-                );
-            }
-            Err(e) => error!(
-                log,
-                "Unable to read target balance";
-                "error" => ?e,
-            ),
-        }
-
-        match summary.previous_epoch_head_attesting_balance() {
-            Ok(head_balance) => {
-                metrics::maybe_set_float_gauge(
-                    &metrics::PARTICIPATION_PREV_EPOCH_HEAD_ATTESTER,
-                    participation_ratio(
-                        head_balance,
-                        summary.previous_epoch_total_active_balance(),
-                    ),
-                );
-            }
-            Err(e) => error!(
-                log,
-                "Unable to read head balance";
-                "error" => ?e,
-            ),
-        }
-    }
-}
-
-fn participation_ratio(section: u64, total: u64) -> Option<f64> {
-    // Reduce the precision to help ensure we fit inside a u32.
-    const PRECISION: u64 = 100_000_000;
-
-    let section: f64 = u32::try_from(section / PRECISION).ok()?.into();
-    let total: f64 = u32::try_from(total / PRECISION).ok()?.into();
-
-    if total > 0_f64 {
-        Some(section / total)
-    } else {
-        None
-    }
-}
-
 fn write_state<T: EthSpec>(prefix: &str, state: &BeaconState<T>, log: &Logger) {
     if WRITE_BLOCK_PROCESSING_SSZ {
         let root = state.tree_hash_root();

beacon_node/beacon_chain/src/block_verification.rs

@@ -330,17 +330,6 @@ lazy_static! {
     pub static ref OP_POOL_NUM_VOLUNTARY_EXITS: Result<IntGauge> =
         try_create_int_gauge("beacon_op_pool_voluntary_exits_total", "Count of voluntary exits in the op pool");
 
-    /*
-     * Participation Metrics
-     */
-    pub static ref PARTICIPATION_PREV_EPOCH_TARGET_ATTESTER: Result<Gauge> = try_create_float_gauge(
-        "beacon_participation_prev_epoch_target_attester",
-        "Ratio of target-attesting balances to total balances"
-    );
-    pub static ref PARTICIPATION_PREV_EPOCH_HEAD_ATTESTER: Result<Gauge> = try_create_float_gauge(
-        "beacon_participation_prev_epoch_head_attester",
-        "Ratio of head-attesting balances to total balances"
-    );
 
     /*
      * Attestation Observation Metrics

beacon_node/beacon_chain/src/metrics.rs

@@ -233,6 +233,16 @@ fn advance_head<T: BeaconChainTypes>(
     if let Some(summary) = per_slot_processing(&mut state, state_root, &beacon_chain.spec)
         .map_err(BeaconChainError::from)?
     {
+        // Expose Prometheus metrics.
+        if let Err(e) = summary.observe_metrics() {
+            error!(
+                log,
+                "Failed to observe epoch summary metrics";
+                "src" => "state_advance_timer",
+                "error" => ?e
+            );
+        }
+
         // Only notify the validator monitor for recent blocks.
         if state.current_epoch() + VALIDATOR_MONITOR_HISTORIC_EPOCHS as u64
             >= current_slot.epoch(T::EthSpec::slots_per_epoch())

beacon_node/beacon_chain/src/state_advance_timer.rs

@@ -29,11 +29,14 @@ eth2_hashing = "0.1.0"
 int_to_bytes = { path = "../int_to_bytes" }
 smallvec = "1.6.1"
 arbitrary = { version = "0.4.6", features = ["derive"], optional = true }
+lighthouse_metrics = { path = "../../common/lighthouse_metrics", optional = true }
+lazy_static = { version = "1.4.0", optional = true }
 
 [features]
-default = ["legacy-arith"]
+default = ["legacy-arith", "metrics"]
 fake_crypto = ["bls/fake_crypto"]
 legacy-arith = ["types/legacy-arith"]
+metrics = ["lighthouse_metrics", "lazy_static"]
 arbitrary-fuzz = [
   "arbitrary",
   "types/arbitrary-fuzz",

consensus/state_processing/Cargo.toml

@@ -14,6 +14,7 @@
 
 #[macro_use]
 mod macros;
+mod metrics;
 
 pub mod common;
 pub mod genesis;

consensus/state_processing/src/lib.rs

@@ -0,0 +1,30 @@
+#[cfg(feature = "metrics")]
+pub use only_if_enabled::*;
+
+#[cfg(feature = "metrics")]
+pub mod only_if_enabled {
+    use lazy_static::lazy_static;
+    pub use lighthouse_metrics::*;
+
+    lazy_static! {
+        /*
+         * Participation Metrics
+         */
+        pub static ref PARTICIPATION_PREV_EPOCH_HEAD_ATTESTING_GWEI_TOTAL: Result<IntGauge> = try_create_int_gauge(
+            "beacon_participation_prev_epoch_head_attesting_gwei_total",
+            "Total effective balance (gwei) of validators who attested to the head in the previous epoch"
+        );
+        pub static ref PARTICIPATION_PREV_EPOCH_TARGET_ATTESTING_GWEI_TOTAL: Result<IntGauge> = try_create_int_gauge(
+            "beacon_participation_prev_epoch_target_attesting_gwei_total",
+            "Total effective balance (gwei) of validators who attested to the target in the previous epoch"
+        );
+        pub static ref PARTICIPATION_PREV_EPOCH_SOURCE_ATTESTING_GWEI_TOTAL: Result<IntGauge> = try_create_int_gauge(
+            "beacon_participation_prev_epoch_source_attesting_gwei_total",
+            "Total effective balance (gwei) of validators who attested to the source in the previous epoch"
+        );
+        pub static ref PARTICIPATION_PREV_EPOCH_ACTIVE_GWEI_TOTAL: Result<IntGauge> = try_create_int_gauge(
+            "beacon_participation_prev_epoch_active_gwei_total",
+            "Total effective balance (gwei) of validators active in the previous epoch"
+        );
+    }
+}

consensus/state_processing/src/metrics.rs

@@ -303,6 +303,11 @@ impl ParticipationCache {
             .total_flag_balance(TIMELY_TARGET_FLAG_INDEX)
     }
 
+    pub fn previous_epoch_source_attesting_balance(&self) -> Result<u64, Error> {
+        self.previous_epoch_participation
+            .total_flag_balance(TIMELY_SOURCE_FLAG_INDEX)
+    }
+
     pub fn previous_epoch_head_attesting_balance(&self) -> Result<u64, Error> {
         self.previous_epoch_participation
             .total_flag_balance(TIMELY_HEAD_FLAG_INDEX)

consensus/state_processing/src/per_epoch_processing/altair/participation_cache.rs

@@ -3,6 +3,7 @@ use super::{
     base::{TotalBalances, ValidatorStatus},
     validator_statuses::InclusionInfo,
 };
+use crate::metrics;
 
 /// Provides a summary of validator participation during the epoch.
 #[derive(PartialEq, Debug)]
@@ -17,6 +18,29 @@ pub enum EpochProcessingSummary {
 }
 
 impl EpochProcessingSummary {
+    /// Updates some Prometheus metrics with some values in `self`.
+    #[cfg(feature = "metrics")]
+    pub fn observe_metrics(&self) -> Result<(), ParticipationCacheError> {
+        metrics::set_gauge(
+            &metrics::PARTICIPATION_PREV_EPOCH_HEAD_ATTESTING_GWEI_TOTAL,
+            self.previous_epoch_head_attesting_balance()? as i64,
+        );
+        metrics::set_gauge(
+            &metrics::PARTICIPATION_PREV_EPOCH_TARGET_ATTESTING_GWEI_TOTAL,
+            self.previous_epoch_target_attesting_balance()? as i64,
+        );
+        metrics::set_gauge(
+            &metrics::PARTICIPATION_PREV_EPOCH_SOURCE_ATTESTING_GWEI_TOTAL,
+            self.previous_epoch_source_attesting_balance()? as i64,
+        );
+        metrics::set_gauge(
+            &metrics::PARTICIPATION_PREV_EPOCH_ACTIVE_GWEI_TOTAL,
+            self.previous_epoch_total_active_balance() as i64,
+        );
+
+        Ok(())
+    }
+
     /// Returns the sum of the effective balance of all validators in the current epoch.
     pub fn current_epoch_total_active_balance(&self) -> u64 {
         match self {
@@ -127,6 +151,24 @@ impl EpochProcessingSummary {
         }
     }
 
+    /// Returns the sum of the effective balance of all validators in the previous epoch who
+    /// included an attestation that matched the source.
+    ///
+    /// ## Differences between Base and Altair
+    ///
+    /// - Base: any attestation can match the source.
+    /// - Altair: only "timely" attestations can match the source.
+    pub fn previous_epoch_source_attesting_balance(&self) -> Result<u64, ParticipationCacheError> {
+        match self {
+            EpochProcessingSummary::Base { total_balances, .. } => {
+                Ok(total_balances.previous_epoch_attesters())
+            }
+            EpochProcessingSummary::Altair {
+                participation_cache,
+            } => participation_cache.previous_epoch_source_attesting_balance(),
+        }
+    }
+
     /// Returns `true` if `val_index` was included in the active validator indices in the previous
     /// epoch *and* the validator is not slashed.
     ///