Track beacon processor import result metrics (#6541)

* Track beacon processor import result metrics

* Update metric name

beacon_node/beacon_chain/src/block_verification.rs

@@ -92,6 +92,7 @@ use std::fs;
 use std::io::Write;
 use std::sync::Arc;
 use store::{Error as DBError, HotStateSummary, KeyValueStore, StoreOp};
+use strum::AsRefStr;
 use task_executor::JoinHandle;
 use types::{
     data_column_sidecar::DataColumnSidecarError, BeaconBlockRef, BeaconState, BeaconStateError,
@@ -137,7 +138,7 @@ const WRITE_BLOCK_PROCESSING_SSZ: bool = cfg!(feature = "write_ssz_files");
 ///
 /// - The block is malformed/invalid (indicated by all results other than `BeaconChainError`.
 /// - We encountered an error whilst trying to verify the block (a `BeaconChainError`).
-#[derive(Debug)]
+#[derive(Debug, AsRefStr)]
 pub enum BlockError {
     /// The parent block was unknown.
     ///

beacon_node/network/src/metrics.rs

@@ -2,7 +2,8 @@ use beacon_chain::{
     attestation_verification::Error as AttnError,
     light_client_finality_update_verification::Error as LightClientFinalityUpdateError,
     light_client_optimistic_update_verification::Error as LightClientOptimisticUpdateError,
-    sync_committee_verification::Error as SyncCommitteeError,
+    sync_committee_verification::Error as SyncCommitteeError, AvailabilityProcessingStatus,
+    BlockError,
 };
 use fnv::FnvHashMap;
 use lighthouse_network::{
@@ -11,12 +12,19 @@ use lighthouse_network::{
 };
 pub use metrics::*;
 use std::sync::{Arc, LazyLock};
+use strum::AsRefStr;
 use strum::IntoEnumIterator;
 use types::EthSpec;
 
 pub const SUCCESS: &str = "SUCCESS";
 pub const FAILURE: &str = "FAILURE";
 
+#[derive(Debug, AsRefStr)]
+pub(crate) enum BlockSource {
+    Gossip,
+    Rpc,
+}
+
 pub static BEACON_BLOCK_MESH_PEERS_PER_CLIENT: LazyLock<Result<IntGaugeVec>> =
     LazyLock::new(|| {
         try_create_int_gauge_vec(
@@ -59,6 +67,27 @@ pub static SYNC_COMMITTEE_SUBSCRIPTION_REQUESTS: LazyLock<Result<IntCounter>> =
         )
     });
 
+/*
+ * Beacon processor
+ */
+pub static BEACON_PROCESSOR_MISSING_COMPONENTS: LazyLock<Result<IntCounterVec>> = LazyLock::new(
+    || {
+        try_create_int_counter_vec(
+            "beacon_processor_missing_components_total",
+            "Total number of imported individual block components that resulted in missing components",
+            &["source", "component"],
+        )
+    },
+);
+pub static BEACON_PROCESSOR_IMPORT_ERRORS_PER_TYPE: LazyLock<Result<IntCounterVec>> =
+    LazyLock::new(|| {
+        try_create_int_counter_vec(
+            "beacon_processor_import_errors_total",
+            "Total number of block components that were not verified",
+            &["source", "component", "type"],
+        )
+    });
+
 /*
  * Gossip processor
  */
@@ -606,6 +635,37 @@ pub fn register_sync_committee_error(error: &SyncCommitteeError) {
     inc_counter_vec(&GOSSIP_SYNC_COMMITTEE_ERRORS_PER_TYPE, &[error.as_ref()]);
 }
 
+pub(crate) fn register_process_result_metrics(
+    result: &std::result::Result<AvailabilityProcessingStatus, BlockError>,
+    source: BlockSource,
+    block_component: &'static str,
+) {
+    match result {
+        Ok(status) => match status {
+            AvailabilityProcessingStatus::Imported { .. } => match source {
+                BlockSource::Gossip => {
+                    inc_counter(&BEACON_PROCESSOR_GOSSIP_BLOCK_IMPORTED_TOTAL);
+                }
+                BlockSource::Rpc => {
+                    inc_counter(&BEACON_PROCESSOR_RPC_BLOCK_IMPORTED_TOTAL);
+                }
+            },
+            AvailabilityProcessingStatus::MissingComponents { .. } => {
+                inc_counter_vec(
+                    &BEACON_PROCESSOR_MISSING_COMPONENTS,
+                    &[source.as_ref(), block_component],
+                );
+            }
+        },
+        Err(error) => {
+            inc_counter_vec(
+                &BEACON_PROCESSOR_IMPORT_ERRORS_PER_TYPE,
+                &[source.as_ref(), block_component, error.as_ref()],
+            );
+        }
+    }
+}
+
 pub fn from_result<T, E>(result: &std::result::Result<T, E>) -> &str {
     match result {
         Ok(_) => SUCCESS,

beacon_node/network/src/network_beacon_processor/gossip_methods.rs

@@ -1,5 +1,5 @@
 use crate::{
-    metrics,
+    metrics::{self, register_process_result_metrics},
     network_beacon_processor::{InvalidBlockStorage, NetworkBeaconProcessor},
     service::NetworkMessage,
     sync::SyncMessage,
@@ -915,12 +915,11 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         let blob_index = verified_blob.id().index;
 
         let result = self.chain.process_gossip_blob(verified_blob).await;
+        register_process_result_metrics(&result, metrics::BlockSource::Gossip, "blob");
 
         match &result {
             Ok(AvailabilityProcessingStatus::Imported(block_root)) => {
-                // Note: Reusing block imported metric here
+                info!(
-                metrics::inc_counter(&metrics::BEACON_PROCESSOR_GOSSIP_BLOCK_IMPORTED_TOTAL);
-                debug!(
                     self.log,
                     "Gossipsub blob processed - imported fully available block";
                     "block_root" => %block_root
@@ -989,18 +988,15 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         let data_column_slot = verified_data_column.slot();
         let data_column_index = verified_data_column.id().index;
 
-        match self
+        let result = self
             .chain
             .process_gossip_data_columns(vec![verified_data_column], || Ok(()))
-            .await
+            .await;
-        {
+        register_process_result_metrics(&result, metrics::BlockSource::Gossip, "data_column");
-            Ok(availability) => {
+
-                match availability {
+        match result {
+            Ok(availability) => match availability {
                 AvailabilityProcessingStatus::Imported(block_root) => {
-                        // Note: Reusing block imported metric here
-                        metrics::inc_counter(
-                            &metrics::BEACON_PROCESSOR_GOSSIP_BLOCK_IMPORTED_TOTAL,
-                        );
                     info!(
                         self.log,
                         "Gossipsub data column processed, imported fully available block";
@@ -1024,8 +1020,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
 
                     self.attempt_data_column_reconstruction(block_root).await;
                 }
-                }
+            },
-            }
             Err(BlockError::DuplicateFullyImported(_)) => {
                 debug!(
                     self.log,
@@ -1467,11 +1462,10 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                 NotifyExecutionLayer::Yes,
             )
             .await;
+        register_process_result_metrics(&result, metrics::BlockSource::Gossip, "block");
 
         match &result {
             Ok(AvailabilityProcessingStatus::Imported(block_root)) => {
-                metrics::inc_counter(&metrics::BEACON_PROCESSOR_GOSSIP_BLOCK_IMPORTED_TOTAL);
-
                 if reprocess_tx
                     .try_send(ReprocessQueueMessage::BlockImported {
                         block_root: *block_root,

beacon_node/network/src/network_beacon_processor/sync_methods.rs

@@ -1,4 +1,4 @@
-use crate::metrics;
+use crate::metrics::{self, register_process_result_metrics};
 use crate::network_beacon_processor::{NetworkBeaconProcessor, FUTURE_SLOT_TOLERANCE};
 use crate::sync::BatchProcessResult;
 use crate::sync::{
@@ -163,8 +163,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                 NotifyExecutionLayer::Yes,
             )
             .await;
-
+        register_process_result_metrics(&result, metrics::BlockSource::Rpc, "block");
-        metrics::inc_counter(&metrics::BEACON_PROCESSOR_RPC_BLOCK_IMPORTED_TOTAL);
 
         // RPC block imported, regardless of process type
         match result.as_ref() {
@@ -286,6 +285,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
         }
 
         let result = self.chain.process_rpc_blobs(slot, block_root, blobs).await;
+        register_process_result_metrics(&result, metrics::BlockSource::Rpc, "blobs");
 
         match &result {
             Ok(AvailabilityProcessingStatus::Imported(hash)) => {
@@ -343,6 +343,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
             .chain
             .process_rpc_custody_columns(custody_columns)
             .await;
+        register_process_result_metrics(&result, metrics::BlockSource::Rpc, "custody_columns");
 
         match &result {
             Ok(availability) => match availability {