feat(derive): Batch Metrics

crates/protocol/derive/src/metrics/mod.rs

@@ -31,6 +31,12 @@
 
     /// Identifier for the gauge that tracks the maximum rlp byte size per channel.
     pub const PIPELINE_MAX_RLP_BYTES: &str = "kona_derive_max_rlp_bytes";
+
+    /// Identifier for the batch stream stage singular batch buffer size.
+    pub const PIPELINE_BATCH_BUFFER: &str = "kona_derive_batch_buffer";
+
+    /// Identifier for the batch stream stage batch memory overhead gauge.
+    pub const PIPELINE_BATCH_MEM: &str = "kona_derive_batch_mem";
 }
 
 impl Metrics {
@@ -84,16 +90,21 @@
             Self::PIPELINE_MAX_RLP_BYTES,
             "The maximum rlp byte size of a channel"
         );
+        metrics::describe_gauge!(
+            Self::PIPELINE_BATCH_BUFFER,
+            "The number of batches held in the batch stream stage"
+        );
+        metrics::describe_gauge!(
+            Self::PIPELINE_BATCH_MEM,
+            "The memory size of batches held in the batch stream stage"
+        );
     }
 
     /// Initializes metrics to 0 so they can be queried immediately.
     #[cfg(feature = "metrics")]
     pub fn zero() {
         kona_macros::set!(gauge, Self::PIPELINE_FRAME_QUEUE_BUFFER, 0);
-        kona_macros::set!(gauge, Self::PIPELINE_FRAME_QUEUE_MEM, 0);
         kona_macros::set!(gauge, Self::PIPELINE_CHANNEL_BUFFER, 0);
-        kona_macros::set!(gauge, Self::PIPELINE_CHANNEL_MEM, 0);
         kona_macros::set!(gauge, Self::PIPELINE_CHANNEL_TIMEOUT, 0);
-        kona_macros::set!(gauge, Self::PIPELINE_MAX_RLP_BYTES, 0);
     }
 }

crates/protocol/derive/src/stages/batch/batch_stream.rs

@@ -95,6 +95,13 @@ where
                 })?,
             );
         }
+        #[cfg(feature = "metrics")]
+        {
+            let batch_count = self.buffer.len() as f64;
+            kona_macros::set!(gauge, crate::metrics::Metrics::PIPELINE_BATCH_BUFFER, batch_count);
+            let batch_size = std::mem::size_of_val(&self.buffer) as f64;
+            kona_macros::set!(gauge, crate::metrics::Metrics::PIPELINE_BATCH_MEM, batch_size);
+        }
         Ok(())
     }
 }
@@ -122,6 +129,14 @@ where
         parent: L2BlockInfo,
         l1_origins: &[BlockInfo],
     ) -> PipelineResult<Batch> {
+        #[cfg(feature = "metrics")]
+        {
+            let batch_count = self.buffer.len() as f64;
+            kona_macros::set!(gauge, crate::metrics::Metrics::PIPELINE_BATCH_BUFFER, batch_count);
+            let batch_size = std::mem::size_of_val(&self.buffer) as f64;
+            kona_macros::set!(gauge, crate::metrics::Metrics::PIPELINE_BATCH_MEM, batch_size);
+        }
+
         // If the stage is not active, "pass" the next batch
         // through this stage to the BatchQueue stage.
         if !self.is_active()? {