feat: PipelineMetrics common scheme, no-op version

bin/client/src/kona.rs

@@ -17,6 +17,7 @@ use kona_common_proc::client_entry;
 
 pub(crate) mod fault;
 use fault::{fpvm_handle_register, HINT_WRITER, ORACLE_READER};
+use kona_derive::metrics::PipelineMetrics;
 
 /// The size of the LRU cache in the oracle.
 const ORACLE_LRU_SIZE: usize = 1024;
@@ -42,6 +43,7 @@ fn main() -> Result<()> {
         let l1_provider = OracleL1ChainProvider::new(boot.clone(), oracle.clone());
         let l2_provider = OracleL2ChainProvider::new(boot.clone(), oracle.clone());
         let beacon = OracleBlobProvider::new(oracle.clone());
+        let metrics = PipelineMetrics::no_op();
 
         ////////////////////////////////////////////////////////////////
         //                   DERIVATION & EXECUTION                   //
@@ -54,6 +56,7 @@ fn main() -> Result<()> {
             beacon,
             l1_provider,
             l2_provider.clone(),
+            metrics,
         )
         .await?;
 

bin/client/src/l1/driver.rs

@@ -13,6 +13,7 @@ use core::fmt::Debug;
 use kona_derive::{
     attributes::StatefulAttributesBuilder,
     errors::{PipelineErrorKind, ResetError},
+    metrics::PipelineMetrics,
     pipeline::{DerivationPipeline, Pipeline, PipelineBuilder, StepResult},
     sources::EthereumDataSource,
     stages::{
@@ -37,6 +38,7 @@ use tracing::{error, info, warn};
 pub type OraclePipeline<O, B> = DerivationPipeline<
     OracleAttributesQueue<OracleDataProvider<O, B>, O>,
     OracleL2ChainProvider<O>,
+    PipelineMetrics,
 >;
 
 /// An oracle-backed Ethereum data source.
@@ -121,6 +123,7 @@ where
         blob_provider: B,
         mut chain_provider: OracleL1ChainProvider<O>,
         mut l2_chain_provider: OracleL2ChainProvider<O>,
+        metrics: PipelineMetrics,
     ) -> Result<Self> {
         let cfg = Arc::new(boot_info.rollup_config.clone());
 
@@ -158,6 +161,7 @@ where
             .chain_provider(chain_provider)
             .builder(attributes)
             .origin(l1_origin)
+            .metrics(metrics)
             .build();
 
         Ok(Self { l2_safe_head, l2_safe_head_header, pipeline })

crates/derive-alloy/src/pipeline.rs

@@ -1,7 +1,11 @@
 //! Helper to construct a [DerivationPipeline] using online types.
 
+use crate::{
+    AlloyChainProvider, AlloyL2ChainProvider, OnlineBeaconClient, OnlineBlobProviderWithFallback,
+};
 use kona_derive::{
     attributes::StatefulAttributesBuilder,
+    metrics::PipelineMetrics,
     pipeline::{DerivationPipeline, PipelineBuilder},
     sources::EthereumDataSource,
     stages::{
@@ -13,13 +17,12 @@ use op_alloy_genesis::RollupConfig;
 use op_alloy_protocol::BlockInfo;
 use std::sync::Arc;
 
-use crate::{
-    AlloyChainProvider, AlloyL2ChainProvider, OnlineBeaconClient, OnlineBlobProviderWithFallback,
-};
-
 /// An online derivation pipeline.
-pub type OnlinePipeline =
-    DerivationPipeline<OnlineAttributesQueue<OnlineDataProvider>, AlloyL2ChainProvider>;
+pub type OnlinePipeline = DerivationPipeline<
+    OnlineAttributesQueue<OnlineDataProvider>,
+    AlloyL2ChainProvider,
+    PipelineMetrics,
+>;
 
 /// An `online` Ethereum data source.
 pub type OnlineDataProvider = EthereumDataSource<
@@ -63,6 +66,7 @@ pub fn new_online_pipeline(
         .chain_provider(chain_provider)
         .builder(builder)
         .origin(origin)
+        .metrics(PipelineMetrics::no_op())
         .build()
 }
 

crates/derive/src/lib.rs

@@ -22,6 +22,7 @@ pub mod prelude {
 
 pub mod attributes;
 pub mod errors;
+pub mod metrics;
 pub mod pipeline;
 pub mod sources;
 pub mod stages;

crates/derive/src/metrics/mod.rs

@@ -0,0 +1,34 @@
+//! Metrics for the derivation pipeline.
+
+mod noop;
+
+use alloc::sync::Arc;
+use core::fmt::Debug;
+
+use crate::traits::{DerivationPipelineMetrics, StepResult};
+
+/// Composite metrics struct containing metrics for all stages.
+pub struct PipelineMetrics {
+    pub(crate) derivation_pipeline_metrics: Arc<dyn DerivationPipelineMetrics + Send + Sync>,
+    // todo: add more metrics here for each stage
+}
+
+impl Debug for PipelineMetrics {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.debug_struct("PipelineMetrics").finish()
+    }
+}
+
+impl DerivationPipelineMetrics for PipelineMetrics {
+    fn record_step_result(&self, result: &StepResult) {
+        self.derivation_pipeline_metrics.record_step_result(result)
+    }
+
+    fn inc_reset_signals(&self) {
+        self.derivation_pipeline_metrics.inc_reset_signals()
+    }
+
+    fn inc_flush_channel_signals(&self) {
+        self.derivation_pipeline_metrics.inc_flush_channel_signals()
+    }
+}

crates/derive/src/metrics/noop.rs

@@ -0,0 +1,28 @@
+use crate::{metrics::PipelineMetrics, pipeline::StepResult, traits::DerivationPipelineMetrics};
+use alloc::sync::Arc;
+
+impl PipelineMetrics {
+    /// No-op implementation for `PipelineMetrics`.
+    pub fn no_op() -> Self {
+        Self {
+            derivation_pipeline_metrics: Arc::new(NoopDerivationPipelineMetrics),
+            // todo: add more metrics here for each stage
+        }
+    }
+}
+
+/// No-op implementation of `DerivationPipelineMetrics`.
+#[derive(Debug)]
+struct NoopDerivationPipelineMetrics;
+
+impl DerivationPipelineMetrics for NoopDerivationPipelineMetrics {
+    fn record_step_result(&self, _result: &StepResult) {
+        // No-op
+    }
+    fn inc_reset_signals(&self) {
+        // No-op
+    }
+    fn inc_flush_channel_signals(&self) {
+        // No-op
+    }
+}

crates/derive/src/pipeline/builder.rs

@@ -2,6 +2,7 @@
 
 use super::{AttributesBuilder, DataAvailabilityProvider, DerivationPipeline};
 use crate::{
+    metrics::PipelineMetrics,
     stages::{
         AttributesQueue, BatchProvider, BatchStream, ChannelProvider, ChannelReader, FrameQueue,
         L1Retrieval, L1Traversal,
@@ -37,6 +38,7 @@ where
     builder: Option<B>,
     origin: Option<BlockInfo>,
     rollup_config: Option<Arc<RollupConfig>>,
+    metrics: Option<PipelineMetrics>,
 }
 
 impl<B, P, T, D> Default for PipelineBuilder<B, P, T, D>
@@ -54,6 +56,7 @@ where
             builder: None,
             origin: None,
             rollup_config: None,
+            metrics: None,
         }
     }
 }
@@ -106,14 +109,20 @@ where
         self
     }
 
+    /// Sets the metrics implementation for the pipeline.
+    pub fn metrics(mut self, metrics: PipelineMetrics) -> Self {
+        self.metrics = Some(metrics);
+        self
+    }
+
     /// Builds the pipeline.
-    pub fn build(self) -> DerivationPipeline<AttributesQueueStage<D, P, T, B>, T> {
+    pub fn build(self) -> DerivationPipeline<AttributesQueueStage<D, P, T, B>, T, PipelineMetrics> {
         self.into()
     }
 }
 
 impl<B, P, T, D> From<PipelineBuilder<B, P, T, D>>
-    for DerivationPipeline<AttributesQueueStage<D, P, T, B>, T>
+    for DerivationPipeline<AttributesQueueStage<D, P, T, B>, T, PipelineMetrics>
 where
     B: AttributesBuilder + Send + Debug,
     P: ChainProvider + Send + Sync + Debug,
@@ -127,7 +136,9 @@ where
         let l2_chain_provider = builder.l2_chain_provider.expect("chain_provider must be set");
         let dap_source = builder.dap_source.expect("dap_source must be set");
         let attributes_builder = builder.builder.expect("builder must be set");
+        let metrics = builder.metrics.expect("metrics must be set");
 
+        // todo: add metrics to the stages
         // Compose the stage stack.
         let mut l1_traversal = L1Traversal::new(chain_provider, Arc::clone(&rollup_config));
         l1_traversal.block = Some(builder.origin.expect("origin must be set"));
@@ -143,6 +154,6 @@ where
             AttributesQueue::new(rollup_config.clone(), batch_provider, attributes_builder);
 
         // Create the pipeline.
-        Self::new(attributes, rollup_config, l2_chain_provider)
+        Self::new(attributes, rollup_config, l2_chain_provider, metrics)
     }
 }