Track chain worker request queue wait time

Author: ndr-ds

linera-core/src/chain_worker/actor.rs

@@ -16,6 +16,7 @@ use linera_base::{
     data_types::{ApplicationDescription, Blob, BlockHeight, Epoch, TimeDelta, Timestamp},
     hashed::Hashed,
     identifiers::{ApplicationId, BlobId, ChainId},
+    time::Instant,
 };
 use linera_chain::{
     data_types::{BlockProposal, MessageBundle, ProposedBlock},
@@ -39,6 +40,22 @@ use crate::{
     worker::{NetworkActions, WorkerError},
 };
 
+#[cfg(with_metrics)]
+mod metrics {
+    use std::sync::LazyLock;
+
+    use linera_base::prometheus_util::{exponential_bucket_latencies, register_histogram};
+    use prometheus::Histogram;
+
+    pub static CHAIN_WORKER_REQUEST_QUEUE_WAIT_TIME: LazyLock<Histogram> = LazyLock::new(|| {
+        register_histogram(
+            "chain_worker_request_queue_wait_time",
+            "Time (ms) a chain worker request waits in queue before being processed",
+            exponential_bucket_latencies(10_000.0),
+        )
+    });
+}
+
 /// A request for the [`ChainWorkerActor`].
 #[derive(Debug)]
 pub(crate) enum ChainWorkerRequest<Context>
@@ -261,6 +278,7 @@ where
         incoming_requests: mpsc::UnboundedReceiver<(
             ChainWorkerRequest<StorageClient::Context>,
             tracing::Span,
+            Instant,
         )>,
         is_tracked: bool,
     ) {
@@ -302,11 +320,19 @@ where
         mut incoming_requests: mpsc::UnboundedReceiver<(
             ChainWorkerRequest<StorageClient::Context>,
             tracing::Span,
+            Instant,
         )>,
     ) -> Result<(), WorkerError> {
         trace!("Starting `ChainWorkerActor`");
 
-        while let Some((request, span)) = incoming_requests.recv().await {
+        while let Some((request, span, _queued_at)) = incoming_requests.recv().await {
+            // Record how long the request waited in queue (in milliseconds)
+            #[cfg(with_metrics)]
+            {
+                let queue_wait_time_ms = _queued_at.elapsed().as_secs_f64() * 1000.0;
+                metrics::CHAIN_WORKER_REQUEST_QUEUE_WAIT_TIME.observe(queue_wait_time_ms);
+            }
+
             let (_service_runtime_thread, service_runtime_task, service_runtime_endpoint) =
                 if self.config.long_lived_services {
                     let actor = ServiceRuntimeActor::spawn(self.chain_id).await;
@@ -337,9 +363,17 @@ where
                 futures::select! {
                     () = self.sleep_until_timeout().fuse() => break,
                     maybe_request = incoming_requests.recv().fuse() => {
-                        let Some((request, span)) = maybe_request else {
+                        let Some((request, span, _queued_at)) = maybe_request else {
                             break; // Request sender was dropped.
                         };
+
+                        // Record how long the request waited in queue (in milliseconds)
+                        #[cfg(with_metrics)]
+                        {
+                            let queue_wait_time_ms = _queued_at.elapsed().as_secs_f64() * 1000.0;
+                            metrics::CHAIN_WORKER_REQUEST_QUEUE_WAIT_TIME.observe(queue_wait_time_ms);
+                        }
+
                         Box::pin(worker.handle_request(request)).instrument(span).await;
                     }
                 }

linera-core/src/worker.rs

@@ -15,6 +15,7 @@ use linera_base::{
     doc_scalar,
     hashed::Hashed,
     identifiers::{AccountOwner, ApplicationId, BlobId, ChainId, EventId, StreamId},
+    time::Instant,
 };
 #[cfg(with_testing)]
 use linera_chain::ChainExecutionContext;
@@ -389,6 +390,7 @@ where
 type ChainActorEndpoint<StorageClient> = mpsc::UnboundedSender<(
     ChainWorkerRequest<<StorageClient as Storage>::Context>,
     tracing::Span,
+    Instant,
 )>;
 
 pub(crate) type DeliveryNotifiers = HashMap<ChainId, DeliveryNotifier>;
@@ -886,7 +888,11 @@ where
         request: ChainWorkerRequest<StorageClient::Context>,
     ) -> Result<
         Option<
-            mpsc::UnboundedReceiver<(ChainWorkerRequest<StorageClient::Context>, tracing::Span)>,
+            mpsc::UnboundedReceiver<(
+                ChainWorkerRequest<StorageClient::Context>,
+                tracing::Span,
+                Instant,
+            )>,
         >,
         WorkerError,
     > {
@@ -899,7 +905,7 @@ where
             (sender, Some(receiver))
         };
 
-        if let Err(e) = sender.send((request, tracing::Span::current())) {
+        if let Err(e) = sender.send((request, tracing::Span::current(), Instant::now())) {
             // The actor was dropped. Give up without (re-)inserting the endpoint in the cache.
             return Err(WorkerError::ChainActorSendError {
                 chain_id,