Add global metrics registry, pass to BeaconState

beacon_node/beacon_chain/src/beacon_chain.rs

@@ -1,5 +1,6 @@
 use crate::checkpoint::CheckPoint;
 use crate::errors::{BeaconChainError as Error, BlockProductionError};
+use crate::metrics::Metrics;
 use crate::persisted_beacon_chain::{PersistedBeaconChain, BEACON_CHAIN_DB_KEY};
 use fork_choice::{ForkChoice, ForkChoiceError};
 use log::{debug, trace};
@@ -96,6 +97,7 @@ pub struct BeaconChain<T: BeaconChainTypes> {
     pub state: RwLock<BeaconState<T::EthSpec>>,
     pub spec: ChainSpec,
     pub fork_choice: RwLock<T::ForkChoice>,
+    pub metrics: Metrics,
 }
 
 impl<T: BeaconChainTypes> BeaconChain<T> {
@@ -138,6 +140,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             canonical_head,
             spec,
             fork_choice: RwLock::new(fork_choice),
+            metrics: Metrics::new()?,
         })
     }
 
@@ -169,6 +172,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             state: RwLock::new(p.state),
             spec,
             fork_choice: RwLock::new(fork_choice),
+            metrics: Metrics::new()?,
         }))
     }
 
@@ -621,6 +625,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
     /// Will accept blocks from prior slots, however it will reject any block from a future slot.
     pub fn process_block(&self, block: BeaconBlock) -> Result<BlockProcessingOutcome, Error> {
         debug!("Processing block with slot {}...", block.slot);
+        self.metrics.blocks_processed.inc();
 
         let block_root = block.block_header().canonical_root();
 
@@ -704,6 +709,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
             self.update_state(state)?;
         }
 
+        self.metrics.valid_blocks_processed.inc();
+
         Ok(BlockProcessingOutcome::ValidBlock(ValidBlock::Processed))
     }
 
@@ -716,6 +723,7 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
         randao_reveal: Signature,
     ) -> Result<(BeaconBlock, BeaconState<T::EthSpec>), BlockProductionError> {
         debug!("Producing block at slot {}...", self.state.read().slot);
+        self.metrics.block_production_requests.inc();
 
         let mut state = self.state.read().clone();
 
@@ -766,6 +774,8 @@ impl<T: BeaconChainTypes> BeaconChain<T> {
 
         block.state_root = state_root;
 
+        self.metrics.block_production_successes.inc();
+
         Ok((block, state))
     }
 

beacon_node/beacon_chain/src/errors.rs

@@ -1,3 +1,4 @@
+use crate::metrics::Error as MetricsError;
 use fork_choice::ForkChoiceError;
 use state_processing::BlockProcessingError;
 use state_processing::SlotProcessingError;
@@ -25,10 +26,17 @@ pub enum BeaconChainError {
     MissingBeaconBlock(Hash256),
     MissingBeaconState(Hash256),
     SlotProcessingError(SlotProcessingError),
+    MetricsError(String),
 }
 
 easy_from_to!(SlotProcessingError, BeaconChainError);
 
+impl From<MetricsError> for BeaconChainError {
+    fn from(e: MetricsError) -> BeaconChainError {
+        BeaconChainError::MetricsError(format!("{:?}", e))
+    }
+}
+
 #[derive(Debug, PartialEq)]
 pub enum BlockProductionError {
     UnableToGetBlockRootFromState,

beacon_node/beacon_chain/src/lib.rs

@@ -1,6 +1,7 @@
 mod beacon_chain;
 mod checkpoint;
 mod errors;
+mod metrics;
 mod persisted_beacon_chain;
 
 pub use self::beacon_chain::{

beacon_node/beacon_chain/src/metrics.rs

@@ -0,0 +1,59 @@
+pub use prometheus::Error;
+use prometheus::{IntCounter, Opts, Registry};
+
+pub struct Metrics {
+    pub blocks_processed: IntCounter,
+    pub valid_blocks_processed: IntCounter,
+    pub block_production_requests: IntCounter,
+    pub block_production_successes: IntCounter,
+    pub attestation_production_requests: IntCounter,
+    pub attestation_production_successes: IntCounter,
+}
+
+impl Metrics {
+    pub fn new() -> Result<Self, Error> {
+        Ok(Self {
+            blocks_processed: {
+                let opts = Opts::new("blocks_processed", "total_blocks_processed");
+                IntCounter::with_opts(opts)?
+            },
+            valid_blocks_processed: {
+                let opts = Opts::new("valid_blocks_processed", "total_valid_blocks_processed");
+                IntCounter::with_opts(opts)?
+            },
+            block_production_requests: {
+                let opts = Opts::new("block_production_requests", "attempts_to_produce_new_block");
+                IntCounter::with_opts(opts)?
+            },
+            block_production_successes: {
+                let opts = Opts::new("block_production_successes", "blocks_successfully_produced");
+                IntCounter::with_opts(opts)?
+            },
+            attestation_production_requests: {
+                let opts = Opts::new(
+                    "attestation_production_requests",
+                    "total_attestation_production_requests",
+                );
+                IntCounter::with_opts(opts)?
+            },
+            attestation_production_successes: {
+                let opts = Opts::new(
+                    "attestation_production_successes",
+                    "total_attestation_production_successes",
+                );
+                IntCounter::with_opts(opts)?
+            },
+        })
+    }
+
+    pub fn register(&self, registry: &Registry) -> Result<(), Error> {
+        registry.register(Box::new(self.blocks_processed.clone()))?;
+        registry.register(Box::new(self.valid_blocks_processed.clone()))?;
+        registry.register(Box::new(self.block_production_requests.clone()))?;
+        registry.register(Box::new(self.block_production_successes.clone()))?;
+        registry.register(Box::new(self.attestation_production_requests.clone()))?;
+        registry.register(Box::new(self.attestation_production_successes.clone()))?;
+
+        Ok(())
+    }
+}