feat: add derivation pipeline queue size metrics (#263)

* add derivation pipeline queue size metrics

* lints

* update docker tag for the rollup-node

Author: frisitano

crates/derivation-pipeline/Cargo.toml

@@ -30,6 +30,7 @@ async-trait = { workspace = true, optional = true }
 futures.workspace = true
 metrics.workspace = true
 metrics-derive.workspace = true
+tokio.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
 

crates/derivation-pipeline/src/lib.rs

@@ -30,6 +30,7 @@ use rollup_node_providers::{BlockDataProvider, L1Provider};
 use scroll_alloy_rpc_types_engine::{BlockDataHint, ScrollPayloadAttributes};
 use scroll_codec::Codec;
 use scroll_db::{Database, DatabaseOperations};
+use tokio::time::Interval;
 
 /// A future that resolves to a stream of [`ScrollPayloadAttributesWithBatchInfo`].
 type DerivationPipelineFuture = Pin<
@@ -43,6 +44,9 @@ type DerivationPipelineFuture = Pin<
     >,
 >;
 
+/// The interval (in ms) at which the derivation pipeline should report queue size metrics.
+const QUEUE_METRICS_INTERVAL: u64 = 1000;
+
 /// A structure holding the current unresolved futures for the derivation pipeline.
 pub struct DerivationPipeline<P> {
     /// The current derivation pipeline futures polled.
@@ -59,6 +63,8 @@ pub struct DerivationPipeline<P> {
     waker: Option<Waker>,
     /// The metrics of the pipeline.
     metrics: DerivationPipelineMetrics,
+    /// The interval at which the derivation pipeline should report queue size metrics.
+    queue_metrics_interval: Interval,
 }
 
 impl<P: Debug> Debug for DerivationPipeline<P> {
@@ -92,6 +98,7 @@ where
             attributes_queue: Default::default(),
             waker: None,
             metrics: DerivationPipelineMetrics::default(),
+            queue_metrics_interval: delayed_interval(QUEUE_METRICS_INTERVAL),
         }
     }
 
@@ -160,6 +167,12 @@ where
         self.batch_queue.clear();
         self.pipeline_future = None;
     }
+
+    /// Emits the queue size metrics for the batch and payload attributes queues.
+    fn emit_queue_gauges(&self) {
+        self.metrics.batch_queue_size.set(self.batch_queue.len() as f64);
+        self.metrics.payload_attributes_queue_size.set(self.attributes_queue.len() as f64);
+    }
 }
 
 impl<P> Stream for DerivationPipeline<P>
@@ -171,6 +184,11 @@ where
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
 
+        // report queue size metrics if the interval has elapsed.
+        while this.queue_metrics_interval.poll_tick(cx).is_ready() {
+            this.emit_queue_gauges();
+        }
+
         // return attributes from the queue if any.
         if let Some(attribute) = this.attributes_queue.pop_front() {
             return Poll::Ready(Some(attribute.inner))
@@ -295,6 +313,14 @@ pub async fn derive<L1P: L1Provider + Sync + Send, L2P: BlockDataProvider + Sync
     Ok(attributes)
 }
 
+/// Creates a delayed interval that will not skip ticks if the interval is missed but will delay
+/// the next tick until the interval has passed.
+fn delayed_interval(interval: u64) -> Interval {
+    let mut interval = tokio::time::interval(tokio::time::Duration::from_millis(interval));
+    interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+    interval
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -388,6 +414,7 @@ mod tests {
             .into(),
             waker: None,
             metrics: Default::default(),
+            queue_metrics_interval: delayed_interval(QUEUE_METRICS_INTERVAL),
         };
 
         // flush and verify all relevant fields are emptied.
@@ -810,6 +837,7 @@ mod tests {
             attributes_queue: attributes,
             waker: None,
             metrics: Default::default(),
+            queue_metrics_interval: delayed_interval(QUEUE_METRICS_INTERVAL),
         }
     }
 

crates/derivation-pipeline/src/metrics.rs

@@ -9,4 +9,8 @@ pub struct DerivationPipelineMetrics {
     pub derived_blocks: Counter,
     /// The blocks per second derived by the pipeline.
     pub blocks_per_second: Gauge,
+    /// The number of batches in the queue.
+    pub batch_queue_size: Gauge,
+    /// The number of payload attributes in the queue.
+    pub payload_attributes_queue_size: Gauge,
 }

docker-compose/docker-compose.yml

@@ -19,7 +19,7 @@ services:
       - scroll-network
 
   rollup-node:
-    image: scrolltech/rollup-node:v0.0.1-rc20
+    image: scrolltech/rollup-node:v0.0.1-rc29
     container_name: rollup-node
     entrypoint: ["bash", "/launch_rollup_node.bash"]
     env_file:

docker-compose/resource/dashboards/rollup_node.json

@@ -619,6 +619,103 @@
       "title": "Derived L2 payload attributes",
       "type": "timeseries"
     },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "PBFA97CFB590B2093"
+      },
+      "description": "Live sizes of the derivation pipeline queues.",
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "axisBorderShow": false,
+            "axisCenteredZero": false,
+            "axisColorMode": "text",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "line",
+            "fillOpacity": 10,
+            "gradientMode": "none",
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "lineInterpolation": "linear",
+            "lineWidth": 2,
+            "pointSize": 3,
+            "scaleDistribution": {
+              "type": "linear"
+            },
+            "showPoints": "auto",
+            "spanNulls": true,
+            "stacking": {
+              "group": "A",
+              "mode": "none"
+            },
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green"
+              },
+              {
+                "color": "red",
+                "value": 40000
+              }
+            ]
+          },
+          "unit": "items"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 8,
+        "w": 12,
+        "x": 12,
+        "y": 17
+      },
+      "id": 14,
+      "options": {
+        "legend": {
+          "displayMode": "list",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "hideZeros": false,
+          "mode": "single",
+          "sort": "none"
+        }
+      },
+      "pluginVersion": "12.0.1",
+      "targets": [
+        {
+          "editorMode": "code",
+          "expr": "reth_derivation_pipeline_batch_queue_size",
+          "legendFormat": "batch_queue_size",
+          "range": true,
+          "refId": "A"
+        },
+        {
+          "editorMode": "code",
+          "expr": "reth_derivation_pipeline_payload_attributes_queue_size",
+          "legendFormat": "attributes_queue_size",
+          "range": true,
+          "refId": "B"
+        }
+      ],
+      "title": "Derivation queue sizes",
+      "type": "timeseries"
+    },
     {
       "collapsed": false,
       "gridPos": {