diff --git a/crates/rpc/rpc-engine-api/src/engine_api.rs b/crates/rpc/rpc-engine-api/src/engine_api.rs
index 21bd5e061..4d259ae00 100644
--- a/crates/rpc/rpc-engine-api/src/engine_api.rs
+++ b/crates/rpc/rpc-engine-api/src/engine_api.rs
@@ -565,9 +565,11 @@ where
     async fn new_payload_v1(&self, payload: ExecutionPayloadV1) -> RpcResult<PayloadStatus> {
         trace!(target: "rpc::engine", "Serving engine_newPayloadV1");
         let start = Instant::now();
+        let gas_used = payload.gas_used;
         let res = Self::new_payload_v1(self, payload).await;
-        self.inner.metrics.latency.new_payload_v1.record(start.elapsed());
-        self.inner.metrics.new_payload_response.update_response_metrics(&res);
+        let elapsed = start.elapsed();
+        self.inner.metrics.latency.new_payload_v1.record(elapsed);
+        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
         Ok(res?)
     }
 
@@ -576,9 +578,11 @@ where
     async fn new_payload_v2(&self, payload: ExecutionPayloadInputV2) -> RpcResult<PayloadStatus> {
         trace!(target: "rpc::engine", "Serving engine_newPayloadV2");
         let start = Instant::now();
+        let gas_used = payload.execution_payload.gas_used;
         let res = Self::new_payload_v2(self, payload).await;
-        self.inner.metrics.latency.new_payload_v2.record(start.elapsed());
-        self.inner.metrics.new_payload_response.update_response_metrics(&res);
+        let elapsed = start.elapsed();
+        self.inner.metrics.latency.new_payload_v2.record(elapsed);
+        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
         Ok(res?)
     }
 
@@ -592,10 +596,12 @@ where
     ) -> RpcResult<PayloadStatus> {
         trace!(target: "rpc::engine", "Serving engine_newPayloadV3");
         let start = Instant::now();
+        let gas_used = payload.payload_inner.payload_inner.gas_used;
         let res =
             Self::new_payload_v3(self, payload, versioned_hashes, parent_beacon_block_root).await;
-        self.inner.metrics.latency.new_payload_v3.record(start.elapsed());
-        self.inner.metrics.new_payload_response.update_response_metrics(&res);
+        let elapsed = start.elapsed();
+        self.inner.metrics.latency.new_payload_v3.record(elapsed);
+        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
         Ok(res?)
     }
 
@@ -607,10 +613,12 @@ where
     ) -> RpcResult<PayloadStatus> {
         trace!(target: "rpc::engine", "Serving engine_newPayloadV4");
         let start = Instant::now();
+        let gas_used = payload.payload_inner.payload_inner.payload_inner.gas_used;
         let res =
             Self::new_payload_v4(self, payload, versioned_hashes, parent_beacon_block_root).await;
-        self.inner.metrics.latency.new_payload_v4.record(start.elapsed());
-        self.inner.metrics.new_payload_response.update_response_metrics(&res);
+        let elapsed = start.elapsed();
+        self.inner.metrics.latency.new_payload_v4.record(elapsed);
+        self.inner.metrics.new_payload_response.update_response_metrics(&res, gas_used, elapsed);
         Ok(res?)
     }
 
diff --git a/crates/rpc/rpc-engine-api/src/metrics.rs b/crates/rpc/rpc-engine-api/src/metrics.rs
index b8679c84e..3c7b5c4b2 100644
--- a/crates/rpc/rpc-engine-api/src/metrics.rs
+++ b/crates/rpc/rpc-engine-api/src/metrics.rs
@@ -1,3 +1,5 @@
+use std::time::Duration;
+
 use crate::EngineApiError;
 use metrics::{Counter, Histogram};
 use reth_metrics::Metrics;
@@ -92,14 +94,27 @@ pub(crate) struct NewPayloadStatusResponseMetrics {
     /// The total count of new payload messages that were unsuccessful, i.e. we responded with an
     /// error type that is not a [PayloadStatusEnum].
     pub(crate) new_payload_error: Counter,
+    /// The total gas of valid new payload messages received.
+    pub(crate) new_payload_total_gas: Histogram,
+    /// The gas per second of valid new payload messages received.
+    pub(crate) new_payload_gas_per_second: Histogram,
 }
 
 impl NewPayloadStatusResponseMetrics {
     /// Increment the newPayload counter based on the given rpc result
-    pub(crate) fn update_response_metrics(&self, result: &Result<PayloadStatus, EngineApiError>) {
+    pub(crate) fn update_response_metrics(
+        &self,
+        result: &Result<PayloadStatus, EngineApiError>,
+        gas_used: u64,
+        time: Duration,
+    ) {
         match result {
             Ok(status) => match status.status {
-                PayloadStatusEnum::Valid => self.new_payload_valid.increment(1),
+                PayloadStatusEnum::Valid => {
+                    self.new_payload_valid.increment(1);
+                    self.new_payload_total_gas.record(gas_used as f64);
+                    self.new_payload_gas_per_second.record(gas_used as f64 / time.as_secs_f64());
+                }
                 PayloadStatusEnum::Syncing => self.new_payload_syncing.increment(1),
                 PayloadStatusEnum::Accepted => self.new_payload_accepted.increment(1),
                 PayloadStatusEnum::Invalid { .. } => self.new_payload_invalid.increment(1),