feat: add payload builder service metrics (#2347)

crates/payload/builder/Cargo.toml

@@ -19,6 +19,10 @@ reth-revm-primitives = { path = "../../revm/revm-primitives" }
 ## ethereum
 revm-primitives = "1.1"
 
+# metrics
+metrics = "0.20.1"
+reth-metrics-derive = { path = "../../metrics/metrics-derive" }
+
 ## async
 tokio = { version = "1", features = ["sync"] }
 tokio-stream = "0.1"

crates/payload/builder/src/lib.rs

@@ -18,6 +18,7 @@
 //!   - [PayloadJob]: a type that can yields (better) payloads over time.
 
 pub mod error;
+mod metrics;
 mod payload;
 mod service;
 mod traits;

crates/payload/builder/src/metrics.rs

@@ -0,0 +1,30 @@
+//! Payloadbuild service metrics.
+
+use metrics::{Counter, Gauge};
+use reth_metrics_derive::Metrics;
+
+/// Transaction pool metrics
+#[derive(Metrics)]
+#[metrics(scope = "payloads")]
+pub(crate) struct PayloadBuilderServiceMetrics {
+    /// Number of active jobs
+    pub(crate) active_jobs: Gauge,
+    /// Total number of initiated jobs
+    pub(crate) initiated_jobs: Counter,
+    /// Total number of failed jobs
+    pub(crate) failed_jobs: Counter,
+}
+
+impl PayloadBuilderServiceMetrics {
+    pub(crate) fn inc_initiated_jobs(&self) {
+        self.initiated_jobs.increment(1);
+    }
+
+    pub(crate) fn inc_failed_jobs(&self) {
+        self.failed_jobs.increment(1);
+    }
+
+    pub(crate) fn set_active_jobs(&self, value: usize) {
+        self.active_jobs.set(value as f64)
+    }
+}

crates/payload/builder/src/service.rs

@@ -4,8 +4,8 @@
 //! Once a new payload is created, it is continuously updated.
 
 use crate::{
-    error::PayloadBuilderError, traits::PayloadJobGenerator, BuiltPayload,
-    PayloadBuilderAttributes, PayloadJob,
+    error::PayloadBuilderError, metrics::PayloadBuilderServiceMetrics, traits::PayloadJobGenerator,
+    BuiltPayload, PayloadBuilderAttributes, PayloadJob,
 };
 use futures_util::stream::{StreamExt, TryStreamExt};
 use reth_rpc_types::engine::PayloadId;
@@ -106,6 +106,8 @@ where
     _service_tx: mpsc::UnboundedSender<PayloadServiceCommand>,
     /// Receiver half of the command channel.
     command_rx: UnboundedReceiverStream<PayloadServiceCommand>,
+    /// metrics for the payload builder service
+    metrics: PayloadBuilderServiceMetrics,
 }
 
 // === impl PayloadBuilderService ===
@@ -122,6 +124,7 @@ where
             payload_jobs: Vec::new(),
             _service_tx: service_tx.clone(),
             command_rx: UnboundedReceiverStream::new(command_rx),
+            metrics: Default::default(),
         };
         let handle = PayloadBuilderHandle { to_service: service_tx };
         (service, handle)
@@ -159,15 +162,19 @@ where
                 loop {
                     match job.try_poll_next_unpin(cx) {
                         Poll::Ready(Some(Ok(payload))) => {
+                            this.metrics.set_active_jobs(this.payload_jobs.len());
                             trace!(?payload, %id, "new payload");
                         }
                         Poll::Ready(Some(Err(err))) => {
                             warn!(?err, %id, "payload job failed; resolving payload");
+                            this.metrics.set_active_jobs(this.payload_jobs.len());
+                            this.metrics.inc_failed_jobs();
                             continue 'jobs
                         }
                         Poll::Ready(None) => {
                             // job is done
                             trace!(?id, "payload job finished");
+                            this.metrics.set_active_jobs(this.payload_jobs.len());
                             continue 'jobs
                         }
                         Poll::Pending => {
@@ -194,10 +201,12 @@ where
                             // no job for this payload yet, create one
                             match this.generator.new_payload_job(attr) {
                                 Ok(job) => {
+                                    this.metrics.inc_initiated_jobs();
                                     new_job = true;
                                     this.payload_jobs.push((job, id));
                                 }
                                 Err(err) => {
+                                    this.metrics.inc_failed_jobs();
                                     warn!(?err, %id, "failed to create payload job");
                                     res = Err(err);
                                 }