refactor(stages): use MetricsListener for Execution stage gas metric

bin/reth/src/node/mod.rs

@@ -690,7 +690,7 @@ impl Command {
  if continuous { HeaderSyncMode::Continuous } else { HeaderSyncMode::Tip(tip_rx) };
  let pipeline = builder
  .with_tip_sender(tip_tx)
- .with_metrics_tx(metrics_tx)
+ .with_metrics_tx(metrics_tx.clone())
  .add_stages(
  DefaultStages::new(
  header_mode,
@@ -706,13 +706,16 @@ impl Command {
  .set(SenderRecoveryStage {
  commit_threshold: stage_config.sender_recovery.commit_threshold,
  })
- .set(ExecutionStage::new(
- factory,
- ExecutionStageThresholds {
- max_blocks: stage_config.execution.max_blocks,
- max_changes: stage_config.execution.max_changes,
- },
- ))
+ .set(
+ ExecutionStage::new(
+ factory,
+ ExecutionStageThresholds {
+ max_blocks: stage_config.execution.max_blocks,
+ max_changes: stage_config.execution.max_changes,
+ },
+ )
+ .with_metrics_tx(metrics_tx),
+ )
  .set(AccountHashingStage::new(
  stage_config.account_hashing.clean_threshold,
  stage_config.account_hashing.commit_threshold,

crates/stages/src/metrics/listener.rs

@@ -1,5 +1,6 @@
 use crate::metrics::SyncMetrics;
 use reth_primitives::{
+ constants::MGAS_TO_GAS,
  stage::{StageCheckpoint, StageId},
  BlockNumber,
 };
@@ -31,6 +32,11 @@ pub enum MetricEvent {
  /// If specified, `entities_total` metric is updated.
  max_block_number: Option<BlockNumber>,
  },
+ /// Execution stage processed some amount of gas.
+ ExecutionStageGas {
+ /// Gas processed.
+ gas: u64,
+ },
 }
 
 /// Metrics routine that listens to new metric events on the `events_rx` receiver.
@@ -71,6 +77,11 @@ impl MetricsListener {
  stage_metrics.entities_total.set(total as f64);
  }
  }
+ MetricEvent::ExecutionStageGas { gas } => self
+ .sync_metrics
+ .execution_stage
+ .mgas_processed_total
+ .increment(gas as f64 / MGAS_TO_GAS as f64),
  }
  }
 }

crates/stages/src/metrics/sync_metrics.rs

@@ -8,6 +8,7 @@ use std::collections::HashMap;
 #[derive(Debug, Default)]
 pub(crate) struct SyncMetrics {
  pub(crate) stages: HashMap<StageId, StageMetrics>,
+ pub(crate) execution_stage: ExecutionStageMetrics,
 }
 
 impl SyncMetrics {
@@ -29,3 +30,11 @@ pub(crate) struct StageMetrics {
  /// The number of total entities of the last commit for a stage, if applicable.
  pub(crate) entities_total: Gauge,
 }
+
+/// Execution stage metrics.
+#[derive(Metrics)]
+#[metrics(scope = "sync.execution")]
+pub(crate) struct ExecutionStageMetrics {
+ /// The total amount of gas processed (in millions)
+ pub(crate) mgas_processed_total: Gauge,
+}

crates/stages/src/stages/execution.rs

@@ -1,4 +1,7 @@
-use crate::{ExecInput, ExecOutput, Stage, StageError, UnwindInput, UnwindOutput};
+use crate::{
+ ExecInput, ExecOutput, MetricEvent, MetricEventsSender, Stage, StageError, UnwindInput,
+ UnwindOutput,
+};
 use reth_db::{
  cursor::{DbCursorRO, DbCursorRW, DbDupCursorRO},
  database::Database,
@@ -7,12 +10,7 @@ use reth_db::{
  transaction::{DbTx, DbTxMut},
 };
 use reth_interfaces::db::DatabaseError;
-use reth_metrics::{
- metrics::{self, Gauge},
- Metrics,
-};
 use reth_primitives::{
- constants::MGAS_TO_GAS,
  stage::{
  CheckpointBlockRange, EntitiesCheckpoint, ExecutionCheckpoint, StageCheckpoint, StageId,
  },
@@ -25,14 +23,6 @@ use reth_provider::{
 use std::{ops::RangeInclusive, time::Instant};
 use tracing::*;
 
-/// Execution stage metrics.
-#[derive(Metrics)]
-#[metrics(scope = "sync.execution")]
-pub struct ExecutionStageMetrics {
- /// The total amount of gas processed (in millions)
- mgas_processed_total: Gauge,
-}
-
 /// The execution stage executes all transactions and
 /// update history indexes.
 ///
@@ -64,7 +54,7 @@ pub struct ExecutionStageMetrics {
 // false positive, we cannot derive it if !DB: Debug.
 #[allow(missing_debug_implementations)]
 pub struct ExecutionStage<EF: ExecutorFactory> {
- metrics: ExecutionStageMetrics,
+ metrics_tx: Option<MetricEventsSender>,
  /// The stage's internal executor
  executor_factory: EF,
  /// The commit thresholds of the execution stage.
@@ -74,7 +64,7 @@ pub struct ExecutionStage<EF: ExecutorFactory> {
 impl<EF: ExecutorFactory> ExecutionStage<EF> {
  /// Create new execution stage with specified config.
  pub fn new(executor_factory: EF, thresholds: ExecutionStageThresholds) -> Self {
- Self { metrics: ExecutionStageMetrics::default(), executor_factory, thresholds }
+ Self { metrics_tx: None, executor_factory, thresholds }
  }
 
  /// Create an execution stage with the provided executor factory.
@@ -84,9 +74,15 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
  Self::new(executor_factory, ExecutionStageThresholds::default())
  }
 
+ /// Set the metric events sender.
+ pub fn with_metrics_tx(mut self, metrics_tx: MetricEventsSender) -> Self {
+ self.metrics_tx = Some(metrics_tx);
+ self
+ }
+
  /// Execute the stage.
  pub fn execute_inner<DB: Database>(
- &self,
+ &mut self,
  provider: &DatabaseProviderRW<'_, &DB>,
  input: ExecInput,
  ) -> Result<ExecOutput, StageError> {
@@ -129,9 +125,10 @@ impl<EF: ExecutorFactory> ExecutionStage<EF> {
  })?;
 
  // Gas metrics
- self.metrics
- .mgas_processed_total
- .increment(block.header.gas_used as f64 / MGAS_TO_GAS as f64);
+ if let Some(metrics_tx) = &mut self.metrics_tx {
+ let _ =
+ metrics_tx.send(MetricEvent::ExecutionStageGas { gas: block.header.gas_used });
+ }
 
  // Merge state changes
  state.extend(block_state);