feat(indexing): measure pipeline stage durations and waits

quickwit/quickwit-actors/src/actor.rs

@@ -111,6 +111,10 @@ pub trait Actor: Send + Sized + 'static {
         type_name::<Self>().to_string()
     }
 
+    /// Enable recording of duration metrics using the mailbox's actor name (normally `name()`).
+    /// When enabled, the name must be stable and contain no instance identifiers.
+    const RECORD_DURATION_METRICS: bool = false;
+
     /// The runner method makes it possible to decide the environment
     /// of execution of the Actor.
     ///

quickwit/quickwit-actors/src/actor_context.rs

@@ -28,6 +28,7 @@ use tracing::debug;
 #[cfg(any(test, feature = "testsuite"))]
 use crate::Universe;
 use crate::actor_state::AtomicState;
+use crate::duration_metrics::{ActorOperation, measure_future_duration};
 use crate::registry::ActorRegistry;
 use crate::spawn_builder::{SpawnBuilder, SpawnContext};
 use crate::{
@@ -128,6 +129,11 @@ impl<A: Actor> ActorContext<A> {
         &self.spawn_ctx.registry
     }
 
+    /// The actor name captured when its mailbox was created, without an instance identifier.
+    pub fn actor_name(&self) -> &str {
+        self.mailbox().actor_name()
+    }
+
     pub fn actor_instance_id(&self) -> &str {
         self.mailbox().actor_instance_id()
     }
@@ -252,12 +258,15 @@ impl<A: Actor> ActorContext<A> {
     {
         let _guard = self.protect_zone();
         debug!(from=%self.self_mailbox.actor_instance_id(), send=%mailbox.actor_instance_id(), msg=?msg);
-        mailbox
-            .send_message_with_backpressure_counter(
-                msg,
-                self.backpressure_micros_counter_opt.as_ref(),
-            )
-            .await
+        let send = mailbox.send_message_with_backpressure_counter(
+            msg,
+            self.backpressure_micros_counter_opt.as_ref(),
+        );
+        if A::RECORD_DURATION_METRICS {
+            measure_future_duration(self.actor_name(), ActorOperation::Send, send).await
+        } else {
+            send.await
+        }
     }
 
     pub async fn ask<DestActor, M, T>(

quickwit/quickwit-actors/src/channel_with_priority.rs

@@ -135,6 +135,7 @@ pub fn channel<T>(queue_capacity: QueueCapacity) -> (Sender<T>, Receiver<T>) {
         high_priority_rx,
         _high_priority_tx: high_priority_tx.clone(),
         pending_low_priority_message: LockedOption::none(),
+        send_to_receive_measurement: None,
         _clone_is_forbidden: CloneIsForbidden,
     };
     let sender = Sender {
@@ -180,6 +181,7 @@ pub struct Receiver<T> {
     high_priority_rx: flume::Receiver<T>,
     _high_priority_tx: flume::Sender<T>,
     pending_low_priority_message: LockedOption<T>,
+    send_to_receive_measurement: Option<fn(&mut T)>,
     _clone_is_forbidden: CloneIsForbidden,
 }
 
@@ -196,14 +198,41 @@ impl<T> Drop for Receiver<T> {
 }
 
 impl<T> Receiver<T> {
+    /// Set the send-to-receive measurement function, called before priority rechecks or parking.
+    /// Returning a parked message must not measure its send-to-receive duration again.
+    pub(crate) fn set_send_to_receive_measurement(&mut self, measure_send_to_receive: fn(&mut T)) {
+        self.send_to_receive_measurement = Some(measure_send_to_receive);
+    }
+
+    fn measure_send_to_receive(&self, mut message: T) -> T {
+        if let Some(measure_send_to_receive) = self.send_to_receive_measurement {
+            measure_send_to_receive(&mut message);
+        }
+        message
+    }
+
+    fn try_recv_from(&self, receiver: &flume::Receiver<T>) -> Result<T, TryRecvError> {
+        receiver
+            .try_recv()
+            .map(|message| self.measure_send_to_receive(message))
+    }
+
+    async fn recv_from(&self, receiver: &flume::Receiver<T>) -> Result<T, flume::RecvError> {
+        // Measure send-to-receive duration before select! drops the other receive future.
+        receiver
+            .recv_async()
+            .await
+            .map(|message| self.measure_send_to_receive(message))
+    }
+
     pub fn is_empty(&self) -> bool {
         self.low_priority_rx.is_empty()
             && self.pending_low_priority_message.is_none()
             && self.high_priority_rx.is_empty()
     }
 
     pub fn try_recv_high_priority_message(&self) -> Result<T, RecvError> {
-        match self.high_priority_rx.try_recv() {
+        match self.try_recv_from(&self.high_priority_rx) {
             Ok(msg) => Ok(msg),
             Err(TryRecvError::Disconnected) => {
                 unreachable!(
@@ -214,7 +243,7 @@ impl<T> Receiver<T> {
                 if self.low_priority_rx.is_disconnected() {
                     // We check that no new high priority message were sent
                     // in between.
-                    if let Ok(msg) = self.high_priority_rx.try_recv() {
+                    if let Ok(msg) = self.try_recv_from(&self.high_priority_rx) {
                         Ok(msg)
                     } else {
                         Err(RecvError::Disconnected)
@@ -227,23 +256,23 @@ impl<T> Receiver<T> {
     }
 
     pub fn try_recv(&self) -> Result<T, RecvError> {
-        if let Ok(msg) = self.high_priority_rx.try_recv() {
+        if let Ok(msg) = self.try_recv_from(&self.high_priority_rx) {
             return Ok(msg);
         }
         if let Some(pending_msg) = self.pending_low_priority_message.take() {
             return Ok(pending_msg);
         }
-        match self.low_priority_rx.try_recv() {
+        match self.try_recv_from(&self.low_priority_rx) {
             Ok(low_msg) => {
-                if let Ok(high_msg) = self.high_priority_rx.try_recv() {
+                if let Ok(high_msg) = self.try_recv_from(&self.high_priority_rx) {
                     self.pending_low_priority_message.place(low_msg);
                     Ok(high_msg)
                 } else {
                     Ok(low_msg)
                 }
             }
             Err(TryRecvError::Disconnected) => {
-                if let Ok(high_msg) = self.high_priority_rx.try_recv() {
+                if let Ok(high_msg) = self.try_recv_from(&self.high_priority_rx) {
                     Ok(high_msg)
                 } else {
                     Err(RecvError::Disconnected)
@@ -254,8 +283,7 @@ impl<T> Receiver<T> {
     }
 
     pub async fn recv_high_priority(&self) -> T {
-        self.high_priority_rx
-            .recv_async()
+        self.recv_from(&self.high_priority_rx)
             .await
             .expect("The Receiver owns the high priority Sender to avoid any disconnection.")
     }
@@ -271,7 +299,7 @@ impl<T> Receiver<T> {
             // We don't really care about fairness here.
             // We will double check if there is a command or not anyway.
             biased;
-            high_priority_msg_res = self.high_priority_rx.recv_async() => {
+            high_priority_msg_res = self.recv_from(&self.high_priority_rx) => {
                 match high_priority_msg_res {
                     Ok(high_priority_msg) => {
                         Ok(high_priority_msg)
@@ -281,7 +309,7 @@ impl<T> Receiver<T> {
                     },
                 }
             }
-            low_priority_msg_res = self.low_priority_rx.recv_async() => {
+            low_priority_msg_res = self.recv_from(&self.low_priority_rx) => {
                 match low_priority_msg_res {
                     Ok(low_priority_msg) => {
                         if let Ok(high_priority_msg) = self.try_recv_high_priority_message() {

quickwit/quickwit-actors/src/duration_metrics.rs

@@ -0,0 +1,126 @@
+// Copyright 2021-Present Datadog, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! `quickwit_actor_duration_seconds{actor,operation}` measures elapsed actor durations.
+//! The actor label uses the name captured at mailbox creation, never an instance identifier.
+//! Message handlers and work performed when the queue becomes empty each contribute a separate
+//! `work` observation. Spawned task lifetimes are not actor work observations.
+//!
+//! Completed futures are observed, including errors; cancellation and panic do not produce totals.
+//! These durations overlap: sends can happen inside work, and send-to-receive includes capacity
+//! waiting and queue residence. Indexing-operation durations also overlap actor work. Do not add
+//! these categories together or subtract independently computed histogram quantiles.
+
+use std::future::Future;
+use std::sync::Arc;
+use std::time::Duration;
+
+use quickwit_metrics::{LazyHistogram, histogram, lazy_histogram};
+use tokio::time::Instant;
+
+static ACTOR_DURATION_SECONDS: LazyHistogram = lazy_histogram!(
+    name: "duration_seconds",
+    description: "Elapsed seconds for actor work, sends, and message-delivery intervals",
+    subsystem: "actor",
+    buckets: vec![
+        0.000_001, 0.000_01, 0.000_1, 0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1.0, 2.5, 5.0,
+        10.0, 20.0, 30.0, 60.0, 120.0, 300.0, 600.0,
+    ],
+);
+
+#[derive(Clone, Copy, Debug)]
+pub(crate) enum ActorOperation {
+    /// Work performed in a message handler or when the message queue becomes empty.
+    Work,
+    /// Send a message, including waiting for channel capacity.
+    Send,
+    /// Interval from starting a send until the destination receives the message.
+    SendToReceive,
+    /// Interval from receipt until handling starts, including priority parking and actor-loop
+    /// yields.
+    ReceiveToHandler,
+}
+
+impl ActorOperation {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Work => "work",
+            Self::Send => "send",
+            Self::SendToReceive => "send_to_receive",
+            Self::ReceiveToHandler => "receive_to_handler",
+        }
+    }
+}
+
+/// Measure a future's duration on completion, including errors but not cancellation or panic.
+/// Callers control actor opt-in. Unlike `HistogramTimer`, this does not record on drop.
+pub(crate) async fn measure_future_duration<F: Future>(
+    actor_name: &str,
+    operation: ActorOperation,
+    future: F,
+) -> F::Output {
+    let started_at = Instant::now();
+    let output = future.await;
+    add_duration_sample(actor_name, operation, started_at.elapsed());
+    output
+}
+
+/// Add one elapsed-duration sample for an actor operation.
+pub(crate) fn add_duration_sample(actor_name: &str, operation: ActorOperation, duration: Duration) {
+    histogram!(parent: ACTOR_DURATION_SECONDS, "actor" => actor_name.to_owned(), "operation" => operation.as_str())
+        .observe(duration.as_secs_f64());
+}
+
+/// Send and receive timestamps for an actor message, with its destination actor's name.
+/// The send timestamp precedes waiting for channel capacity; both timestamps survive parking.
+pub(crate) struct ActorMessageTimestamps {
+    destination_actor_name: Arc<str>,
+    send_started_at: Instant,
+    received_at: Option<Instant>,
+}
+
+impl ActorMessageTimestamps {
+    pub(crate) fn new(destination_actor_name: Arc<str>) -> Self {
+        Self {
+            destination_actor_name,
+            send_started_at: Instant::now(),
+            received_at: None,
+        }
+    }
+
+    pub(crate) fn measure_send_to_receive(&mut self) {
+        let received_at = Instant::now();
+        assert!(
+            self.received_at.is_none(),
+            "an envelope can only be received once"
+        );
+        self.received_at = Some(received_at);
+        add_duration_sample(
+            &self.destination_actor_name,
+            ActorOperation::SendToReceive,
+            received_at - self.send_started_at,
+        );
+    }
+
+    pub(crate) fn measure_receive_to_handler(&self, handler_started_at: Instant) {
+        let received_at = self
+            .received_at
+            .expect("an envelope must be received before handling");
+        add_duration_sample(
+            &self.destination_actor_name,
+            ActorOperation::ReceiveToHandler,
+            handler_started_at - received_at,
+        );
+    }
+}

quickwit/quickwit-actors/src/envelope.rs

@@ -14,11 +14,14 @@
 
 use std::any::Any;
 use std::fmt;
+use std::sync::Arc;
 
 use async_trait::async_trait;
 use tokio::sync::oneshot;
+use tokio::time::Instant;
 
 use crate::actor::DeferableReplyHandler;
+use crate::duration_metrics::{ActorMessageTimestamps, ActorOperation, add_duration_sample};
 use crate::scheduler::NoAdvanceTimeGuard;
 use crate::{Actor, ActorContext, ActorExitStatus};
 
@@ -32,6 +35,8 @@ use crate::{Actor, ActorContext, ActorExitStatus};
 pub struct Envelope<A> {
     handler_envelope: Box<dyn EnvelopeT<A>>,
     _no_advance_time_guard: Option<NoAdvanceTimeGuard>,
+    // The mailbox creates timestamps only when the actor opts into duration metrics.
+    timestamps: Option<ActorMessageTimestamps>,
 }
 
 impl<A: Actor> Envelope<A> {
@@ -50,6 +55,12 @@ impl<A: Actor> Envelope<A> {
         }
     }
 
+    pub(crate) fn measure_send_to_receive(&mut self) {
+        if let Some(timestamps) = &mut self.timestamps {
+            timestamps.measure_send_to_receive();
+        }
+    }
+
     /// Executes the captured handle function.
     ///
     /// When exiting, also returns the message type name.
@@ -58,7 +69,21 @@ impl<A: Actor> Envelope<A> {
         actor: &mut A,
         ctx: &ActorContext<A>,
     ) -> Result<(), (ActorExitStatus, &'static str)> {
-        let handling_res = self.handler_envelope.handle_message(actor, ctx).await;
+        let handling_res = if let Some(timestamps) = &self.timestamps {
+            // The actor loop has already yielded. Share the boundary between dispatch and
+            // handler time, including the envelope adapter overhead.
+            let handler_started_at = Instant::now();
+            timestamps.measure_receive_to_handler(handler_started_at);
+            let result = self.handler_envelope.handle_message(actor, ctx).await;
+            add_duration_sample(
+                ctx.actor_name(),
+                ActorOperation::Work,
+                handler_started_at.elapsed(),
+            );
+            result
+        } else {
+            self.handler_envelope.handle_message(actor, ctx).await
+        };
         if let Err(exit_status) = handling_res {
             return Err((exit_status, self.handler_envelope.message_type_name()));
         }
@@ -145,6 +170,7 @@ where
 pub(crate) fn wrap_in_envelope<A, M>(
     msg: M,
     no_advance_time_guard: Option<NoAdvanceTimeGuard>,
+    actor_name: Option<Arc<str>>,
 ) -> (Envelope<A>, oneshot::Receiver<A::Reply>)
 where
     A: DeferableReplyHandler<M>,
@@ -155,6 +181,9 @@ where
     let envelope = Envelope {
         handler_envelope: Box::new(handler_envelope),
         _no_advance_time_guard: no_advance_time_guard,
+        // All mailbox send paths wrap immediately before their first channel send attempt.
+        // If the channel is full, the asynchronous retry keeps the original send timestamp.
+        timestamps: actor_name.map(ActorMessageTimestamps::new),
     };
     (envelope, response_rx)
 }

quickwit/quickwit-actors/src/lib.rs

@@ -34,6 +34,7 @@ mod actor_state;
 #[doc(hidden)]
 pub mod channel_with_priority;
 mod command;
+mod duration_metrics;
 mod envelope;
 mod mailbox;
 mod observation;