From 34cb0550165f4c364be8caeda2884e2c71608022 Mon Sep 17 00:00:00 2001 From: Pascal Seitz Date: Mon, 19 Oct 2026 21:33:52 +0200 Subject: [PATCH] feat(indexing): measure pipeline stage durations and waits --- quickwit/Cargo.lock | 1 + quickwit/quickwit-actors/Cargo.toml | 1 + quickwit/quickwit-actors/src/actor.rs | 4 + quickwit/quickwit-actors/src/actor_context.rs | 21 +- .../src/channel_with_priority.rs | 48 ++- .../quickwit-actors/src/duration_metrics.rs | 134 ++++++++ quickwit/quickwit-actors/src/envelope.rs | 31 +- quickwit/quickwit-actors/src/lib.rs | 2 + quickwit/quickwit-actors/src/mailbox.rs | 17 +- quickwit/quickwit-actors/src/spawn_builder.rs | 21 +- quickwit/quickwit-actors/src/task_metrics.rs | 323 ++++++++++++++++++ .../src/actors/doc_processor.rs | 2 + .../src/actors/index_serializer.rs | 2 + .../quickwit-indexing/src/actors/indexer.rs | 22 +- .../quickwit-indexing/src/actors/packager.rs | 2 + .../quickwit-indexing/src/actors/publisher.rs | 33 +- .../quickwit-indexing/src/actors/sequencer.rs | 25 +- .../quickwit-indexing/src/actors/uploader.rs | 258 +++++++------- .../quickwit-indexing/src/duration_metrics.rs | 109 ++++++ quickwit/quickwit-indexing/src/lib.rs | 1 + .../src/source/ingest/mod.rs | 39 ++- .../src/source/ingest_api_source.rs | 20 +- quickwit/quickwit-indexing/src/source/mod.rs | 5 +- .../src/source/vec_source.rs | 5 +- 24 files changed, 956 insertions(+), 170 deletions(-) create mode 100644 quickwit/quickwit-actors/src/duration_metrics.rs create mode 100644 quickwit/quickwit-actors/src/task_metrics.rs create mode 100644 quickwit/quickwit-indexing/src/duration_metrics.rs diff --git a/quickwit/Cargo.lock b/quickwit/Cargo.lock index 336b249b2fb..c7cac96d449 100644 --- a/quickwit/Cargo.lock +++ b/quickwit/Cargo.lock @@ -7118,6 +7118,7 @@ dependencies = [ "sync_wrapper", "thiserror 2.0.18", "tokio", + "tokio-metrics", "tracing", ] diff --git a/quickwit/quickwit-actors/Cargo.toml b/quickwit/quickwit-actors/Cargo.toml index 08c63e62add..ab8d1dbbd55 100644 --- a/quickwit/quickwit-actors/Cargo.toml +++ b/quickwit/quickwit-actors/Cargo.toml @@ -20,6 +20,7 @@ serde_json = { workspace = true } sync_wrapper = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } +tokio-metrics = { workspace = true } tracing = { workspace = true } quickwit-common = { workspace = true } diff --git a/quickwit/quickwit-actors/src/actor.rs b/quickwit/quickwit-actors/src/actor.rs index bb5a48239a4..62cdc8a793a 100644 --- a/quickwit/quickwit-actors/src/actor.rs +++ b/quickwit/quickwit-actors/src/actor.rs @@ -111,6 +111,10 @@ pub trait Actor: Send + Sized + 'static { type_name::().to_string() } + /// Enable actor duration metrics, including wake-to-poll intervals, using the mailbox's actor + /// name (normally `name()`). 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. /// diff --git a/quickwit/quickwit-actors/src/actor_context.rs b/quickwit/quickwit-actors/src/actor_context.rs index 6308dd9ca6f..1b37f755cf6 100644 --- a/quickwit/quickwit-actors/src/actor_context.rs +++ b/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 ActorContext { &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 ActorContext { { 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( diff --git a/quickwit/quickwit-actors/src/channel_with_priority.rs b/quickwit/quickwit-actors/src/channel_with_priority.rs index d38936dcf2d..a9da4a45b36 100644 --- a/quickwit/quickwit-actors/src/channel_with_priority.rs +++ b/quickwit/quickwit-actors/src/channel_with_priority.rs @@ -135,6 +135,7 @@ pub fn channel(queue_capacity: QueueCapacity) -> (Sender, Receiver) { 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 { high_priority_rx: flume::Receiver, _high_priority_tx: flume::Sender, pending_low_priority_message: LockedOption, + send_to_receive_measurement: Option, _clone_is_forbidden: CloneIsForbidden, } @@ -196,6 +198,33 @@ impl Drop for Receiver { } impl Receiver { + /// 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) -> Result { + receiver + .try_recv() + .map(|message| self.measure_send_to_receive(message)) + } + + async fn recv_from(&self, receiver: &flume::Receiver) -> Result { + // 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() @@ -203,7 +232,7 @@ impl Receiver { } pub fn try_recv_high_priority_message(&self) -> Result { - 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 Receiver { 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,15 +256,15 @@ impl Receiver { } pub fn try_recv(&self) -> Result { - 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 { @@ -243,7 +272,7 @@ impl Receiver { } } 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 Receiver { } 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 Receiver { // 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 Receiver { }, } } - 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() { diff --git a/quickwit/quickwit-actors/src/duration_metrics.rs b/quickwit/quickwit-actors/src/duration_metrics.rs new file mode 100644 index 00000000000..ce4170f4dd2 --- /dev/null +++ b/quickwit/quickwit-actors/src/duration_metrics.rs @@ -0,0 +1,134 @@ +// 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. +//! +//! Work and send futures are observed on completion, including errors; cancellation and panic do +//! not produce totals. `wake_to_poll` records individual intervals after each task poll returns. +//! 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::{Histogram, 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, message delivery, and wake-to-poll 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, + /// Interval from first wake to next poll, including remaining execution if woken during a poll. + WakeToPoll, +} + +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", + Self::WakeToPoll => "wake_to_poll", + } + } +} + +/// 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( + 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) { + duration_histogram(actor_name, operation).observe(duration.as_secs_f64()); +} + +/// Get the shared histogram for this actor name and operation. +pub(crate) fn duration_histogram(actor_name: &str, operation: ActorOperation) -> Histogram { + histogram!(parent: ACTOR_DURATION_SECONDS, "actor" => actor_name.to_owned(), "operation" => operation.as_str()) +} + +/// 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, + send_started_at: Instant, + received_at: Option, +} + +impl ActorMessageTimestamps { + pub(crate) fn new(destination_actor_name: Arc) -> 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, + ); + } +} diff --git a/quickwit/quickwit-actors/src/envelope.rs b/quickwit/quickwit-actors/src/envelope.rs index a3ca6d6b9ad..75e80699523 100644 --- a/quickwit/quickwit-actors/src/envelope.rs +++ b/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 { handler_envelope: Box>, _no_advance_time_guard: Option, + // The mailbox creates timestamps only when the actor opts into duration metrics. + timestamps: Option, } impl Envelope { @@ -50,6 +55,12 @@ impl Envelope { } } + 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 Envelope { actor: &mut A, ctx: &ActorContext, ) -> 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( msg: M, no_advance_time_guard: Option, + actor_name: Option>, ) -> (Envelope, oneshot::Receiver) where A: DeferableReplyHandler, @@ -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) } diff --git a/quickwit/quickwit-actors/src/lib.rs b/quickwit/quickwit-actors/src/lib.rs index ca2e47e10d2..d178a9494da 100644 --- a/quickwit/quickwit-actors/src/lib.rs +++ b/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; @@ -41,6 +42,7 @@ mod registry; pub(crate) mod scheduler; mod spawn_builder; mod supervisor; +mod task_metrics; pub use scheduler::{SchedulerClient, start_scheduler}; diff --git a/quickwit/quickwit-actors/src/mailbox.rs b/quickwit/quickwit-actors/src/mailbox.rs index 52369458589..33272992092 100644 --- a/quickwit/quickwit-actors/src/mailbox.rs +++ b/quickwit/quickwit-actors/src/mailbox.rs @@ -106,6 +106,7 @@ struct Inner { pub(crate) tx: Sender>, scheduler_client_opt: Option, instance_id: String, + actor_name: Arc, } impl fmt::Debug for Mailbox { @@ -117,6 +118,10 @@ impl fmt::Debug for Mailbox { } impl Mailbox { + pub(crate) fn actor_name(&self) -> &str { + &self.inner.actor_name + } + pub fn actor_instance_id(&self) -> &str { &self.inner.instance_id } @@ -180,7 +185,11 @@ impl Mailbox { .scheduler_client_opt .as_ref() .map(|scheduler_client| scheduler_client.no_advance_time_guard()); - wrap_in_envelope(message, guard) + wrap_in_envelope( + message, + guard, + A::RECORD_DURATION_METRICS.then(|| self.inner.actor_name.clone()), + ) } /// Sends a message to the actor owning the associated inbox. @@ -396,12 +405,16 @@ pub(crate) fn create_mailbox( queue_capacity: QueueCapacity, scheduler_client_opt: Option, ) -> (Mailbox, Inbox) { - let (tx, rx) = crate::channel_with_priority::channel(queue_capacity); + let (tx, mut rx) = crate::channel_with_priority::channel(queue_capacity); + if A::RECORD_DURATION_METRICS { + rx.set_send_to_receive_measurement(Envelope::::measure_send_to_receive); + } let ref_count = Arc::new(AtomicUsize::new(1)); let mailbox = Mailbox { inner: Arc::new(Inner { tx, instance_id: quickwit_common::new_coolid(&actor_name), + actor_name: actor_name.into(), scheduler_client_opt, }), ref_count, diff --git a/quickwit/quickwit-actors/src/spawn_builder.rs b/quickwit/quickwit-actors/src/spawn_builder.rs index e8308693497..12cf87a9724 100644 --- a/quickwit/quickwit-actors/src/spawn_builder.rs +++ b/quickwit/quickwit-actors/src/spawn_builder.rs @@ -16,16 +16,19 @@ use std::fmt; use std::time::Duration; use anyhow::Context; +use futures::future::Either; use quickwit_metrics::Counter; use sync_wrapper::SyncWrapper; use tokio::sync::watch; use tracing::{debug, error}; +use crate::duration_metrics::{ActorOperation, duration_histogram, measure_future_duration}; use crate::envelope::Envelope; use crate::mailbox::{Inbox, create_mailbox}; use crate::registry::{ActorJoinHandle, ActorRegistry}; use crate::scheduler::{NoAdvanceTimeGuard, SchedulerClient}; use crate::supervisor::Supervisor; +use crate::task_metrics::measure_wake_to_poll; use crate::{ Actor, ActorContext, ActorExitStatus, ActorHandle, KillSwitch, Mailbox, QueueCapacity, }; @@ -174,6 +177,12 @@ impl SpawnBuilder { let ctx_clone = ctx.clone(); let loop_async_actor_future = async move { actor_loop(actor, inbox, no_advance_time_guard, ctx).await }; + let loop_async_actor_future = if A::RECORD_DURATION_METRICS { + let histogram = duration_histogram(ctx_clone.actor_name(), ActorOperation::WakeToPoll); + Either::Left(measure_wake_to_poll(loop_async_actor_future, histogram)) + } else { + Either::Right(loop_async_actor_future) + }; let join_handle = ActorJoinHandle::new(quickwit_common::spawn_named_task_on( loop_async_actor_future, std::any::type_name::(), @@ -327,11 +336,13 @@ impl ActorExecutionEnv { break; } } - self.actor - .get_mut() - .on_drained_messages(&self.ctx) - .await - .map_err(|exit_status| (exit_status, ActorExitPhase::OnDrainedMessaged))?; + let drain = self.actor.get_mut().on_drained_messages(&self.ctx); + let result = if A::RECORD_DURATION_METRICS { + measure_future_duration(self.ctx.actor_name(), ActorOperation::Work, drain).await + } else { + drain.await + }; + result.map_err(|exit_status| (exit_status, ActorExitPhase::OnDrainedMessaged))?; } if self.ctx.mailbox().is_last_mailbox() { // We double check here that the mailbox does not contain any messages, diff --git a/quickwit/quickwit-actors/src/task_metrics.rs b/quickwit/quickwit-actors/src/task_metrics.rs new file mode 100644 index 00000000000..3e2471f7a13 --- /dev/null +++ b/quickwit/quickwit-actors/src/task_metrics.rs @@ -0,0 +1,323 @@ +// 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. + +//! Individual wake-to-poll samples in +//! `quickwit_actor_duration_seconds{actor,operation="wake_to_poll"}`. Each actor task owns its +//! monitor; all instances of an actor name share the histogram. No task, worker, pipeline, or +//! actor-instance labels are added. +//! +//! TaskMonitor measures from the first wake until the next poll, including wakeups inside handler +//! awaits and framework yields. A wake during an ongoing poll includes that poll's remaining +//! execution, so this is not pure runtime-queue or off-CPU time. Initial spawn-to-first-poll delay, +//! time pending before a wakeup, and separately spawned tasks are excluded. +//! +//! Samples are recorded after each poll returns, while the actor is still running. A poll that +//! panics does not produce a sample. Correlate with the runtime's worker queue depth and busy +//! ratio. For example, the p99 wake-to-poll duration in seconds across Indexer instances is: +//! ```text +//! histogram_quantile(0.99, sum by (le) ( +//! rate(quickwit_actor_duration_seconds_bucket{actor="Indexer",operation="wake_to_poll"}[5m]) +//! )) +//! ``` + +use std::future::{Future, poll_fn}; +use std::pin::pin; +use std::time::Duration; + +use quickwit_metrics::Histogram; +use tokio_metrics::TaskMonitor; + +/// Record each wake-to-next-poll interval of one task in the supplied histogram. +/// Callers control opt-in and supply a histogram labeled with the stable actor name. +pub(crate) async fn measure_wake_to_poll(task: F, histogram: Histogram) -> F::Output { + // This monitor must instrument exactly one task. TaskMonitor adds at most one scheduling + // interval per poll, so the cumulative-duration delta is one sample, not an average. + let monitor = TaskMonitor::new(); + let mut task = pin!(monitor.instrument(task)); + let mut previous_count = 0; + let mut previous_duration = Duration::ZERO; + poll_fn(|context| { + let result = task.as_mut().poll(context); + let metrics = monitor.cumulative(); + if metrics.total_scheduled_count != previous_count { + assert_eq!( + metrics.total_scheduled_count, + previous_count + 1, + "one task poll can record at most one wake-to-poll interval" + ); + let duration = metrics.total_scheduled_duration - previous_duration; + histogram.observe(duration.as_secs_f64()); + previous_count = metrics.total_scheduled_count; + previous_duration = metrics.total_scheduled_duration; + } + result + }) + .await +} + +#[cfg(test)] +mod tests { + use std::sync::{Arc, Mutex}; + use std::task::{Context, Poll}; + use std::time::Instant; + + use async_trait::async_trait; + use futures::task::noop_waker; + use quickwit_metrics::__metrics as metrics; + use tokio::sync::{mpsc, oneshot}; + + use super::*; + use crate::duration_metrics::{ActorOperation, duration_histogram}; + use crate::{Actor, ActorContext, ActorExitStatus, Universe}; + + #[derive(Default)] + struct Samples(Mutex>); + + impl metrics::HistogramFn for Samples { + fn record(&self, value: f64) { + self.0.lock().unwrap().push(value); + } + } + + #[derive(Default)] + struct TestRecorder { + samples: Arc, + } + + impl TestRecorder { + fn histogram(&self, actor_name: &str) -> Histogram { + metrics::with_local_recorder(self, || { + duration_histogram(actor_name, ActorOperation::WakeToPoll) + }) + } + + fn samples(&self) -> Vec { + self.samples.0.lock().unwrap().clone() + } + } + + impl metrics::Recorder for TestRecorder { + fn describe_counter( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + fn describe_gauge( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + fn describe_histogram( + &self, + _: metrics::KeyName, + _: Option, + _: metrics::SharedString, + ) { + } + + fn register_counter( + &self, + _: &metrics::Key, + _: &metrics::Metadata<'_>, + ) -> metrics::Counter { + metrics::Counter::noop() + } + + fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge { + metrics::Gauge::noop() + } + + fn register_histogram( + &self, + key: &metrics::Key, + _: &metrics::Metadata<'_>, + ) -> metrics::Histogram { + if key + .labels() + .any(|label| label.key() == "operation" && label.value() == "wake_to_poll") + { + metrics::Histogram::from_arc(self.samples.clone()) + } else { + metrics::Histogram::noop() + } + } + } + + #[test] + fn test_individual_samples_and_polls_without_wakes() { + let recorder = TestRecorder::default(); + let histogram = recorder.histogram("IndividualSamplesTestActor"); + let (sender, mut receiver) = mpsc::unbounded_channel(); + let future = async move { + receiver.recv().await.unwrap(); + receiver.recv().await.unwrap(); + Err::<(), _>("completed error") + }; + let mut task = pin!(measure_wake_to_poll(future, histogram)); + let waker = noop_waker(); + let mut context = Context::from_waker(&waker); + assert!(task.as_mut().poll(&mut context).is_pending()); + assert!(recorder.samples().is_empty()); + + let started_at = Instant::now(); + sender.send(()).unwrap(); + std::thread::sleep(Duration::from_millis(2)); + assert!(task.as_mut().poll(&mut context).is_pending()); + assert_eq!(recorder.samples().len(), 1); + // A spurious poll without another wake must not repeat the previous sample. + assert!(task.as_mut().poll(&mut context).is_pending()); + assert_eq!(recorder.samples().len(), 1); + + sender.send(()).unwrap(); + std::thread::sleep(Duration::from_millis(2)); + assert_eq!( + task.as_mut().poll(&mut context), + Poll::Ready(Err("completed error")) + ); + let samples = recorder.samples(); + assert_eq!(samples.len(), 2); + assert!(samples.iter().all(|&sample| sample >= 0.002)); + // Each interval contributes only its own duration, not the cumulative total. + assert!(samples.iter().sum::() <= started_at.elapsed().as_secs_f64()); + } + + #[test] + fn test_first_poll_and_cancellation_do_not_record() { + let recorder = TestRecorder::default(); + let histogram = recorder.histogram("CancelledTaskTestActor"); + let waker = noop_waker(); + let mut context = Context::from_waker(&waker); + let mut ready_task = pin!(measure_wake_to_poll( + std::future::ready(42), + histogram.clone() + )); + assert_eq!(ready_task.as_mut().poll(&mut context), Poll::Ready(42)); + assert!(recorder.samples().is_empty()); + + let (sender, receiver) = oneshot::channel(); + let mut task = Box::pin(measure_wake_to_poll(receiver, histogram)); + assert!(task.as_mut().poll(&mut context).is_pending()); + sender.send(()).unwrap(); + drop(task); + assert!(recorder.samples().is_empty()); + } + + #[test] + fn test_tasks_have_separate_monitors_and_share_histogram() { + let recorder = TestRecorder::default(); + let first_histogram = recorder.histogram("SharedHistogramTestActor"); + let second_histogram = recorder.histogram("SharedHistogramTestActor"); + assert_eq!(first_histogram, second_histogram); + assert_eq!( + first_histogram.key().name(), + "quickwit_actor_duration_seconds" + ); + let labels: Vec<_> = first_histogram.key().labels().collect(); + assert_eq!(labels.len(), 2); + assert!( + labels + .iter() + .any(|label| label.key() == "actor" && label.value() == "SharedHistogramTestActor") + ); + assert!( + labels + .iter() + .any(|label| label.key() == "operation" && label.value() == "wake_to_poll") + ); + + let (first_sender, first_receiver) = oneshot::channel(); + let (second_sender, second_receiver) = oneshot::channel(); + let mut first_task = pin!(measure_wake_to_poll(first_receiver, first_histogram)); + let mut second_task = pin!(measure_wake_to_poll(second_receiver, second_histogram)); + let waker = noop_waker(); + let mut context = Context::from_waker(&waker); + assert!(first_task.as_mut().poll(&mut context).is_pending()); + assert!(second_task.as_mut().poll(&mut context).is_pending()); + first_sender.send(()).unwrap(); + assert!(first_task.as_mut().poll(&mut context).is_ready()); + assert_eq!(recorder.samples().len(), 1); + second_sender.send(()).unwrap(); + assert!(second_task.as_mut().poll(&mut context).is_ready()); + assert_eq!(recorder.samples().len(), 2); + } + + #[test] + fn test_wake_during_poll_includes_remaining_poll_time() { + let recorder = TestRecorder::default(); + let histogram = recorder.histogram("WakeDuringPollTestActor"); + let remaining_poll_time = Duration::from_millis(2); + let mut first_poll = true; + let future = poll_fn(|context| { + if !first_poll { + return Poll::Ready(()); + } + first_poll = false; + context.waker().wake_by_ref(); + std::thread::sleep(remaining_poll_time); + Poll::Pending + }); + let mut task = pin!(measure_wake_to_poll(future, histogram)); + let waker = noop_waker(); + let mut context = Context::from_waker(&waker); + assert!(task.as_mut().poll(&mut context).is_pending()); + assert!(task.as_mut().poll(&mut context).is_ready()); + let samples = recorder.samples(); + assert_eq!(samples.len(), 1); + assert!(samples[0] >= remaining_poll_time.as_secs_f64()); + } + + struct TestActor; + + #[async_trait] + impl Actor for TestActor { + type ObservableState = (); + const RECORD_DURATION_METRICS: bool = RECORD; + + fn name(&self) -> String { + if RECORD { + "MonitoredTaskTestActor".to_string() + } else { + "UnmonitoredTaskTestActor".to_string() + } + } + + fn observable_state(&self) {} + + async fn initialize(&mut self, _ctx: &ActorContext) -> Result<(), ActorExitStatus> { + tokio::task::yield_now().await; + Err(ActorExitStatus::Success) + } + } + + #[tokio::test] + async fn test_actor_spawn_monitoring_opt_in() { + // No process-wide Quickwit runtime initialization or background collector is needed. + let recorder = TestRecorder::default(); + let universe = Universe::new(); + let (_, monitored) = metrics::with_local_recorder(&recorder, || { + universe.spawn_builder().spawn(TestActor::) + }); + assert!(monitored.join().await.0.is_success()); + assert_eq!(recorder.samples().len(), 1); + let (_, unmonitored) = metrics::with_local_recorder(&recorder, || { + universe.spawn_builder().spawn(TestActor::) + }); + assert!(unmonitored.join().await.0.is_success()); + assert_eq!(recorder.samples().len(), 1); + } +} diff --git a/quickwit/quickwit-indexing/src/actors/doc_processor.rs b/quickwit/quickwit-indexing/src/actors/doc_processor.rs index 9c6294bea6d..24bde4a1191 100644 --- a/quickwit/quickwit-indexing/src/actors/doc_processor.rs +++ b/quickwit/quickwit-indexing/src/actors/doc_processor.rs @@ -534,6 +534,8 @@ struct VrlProgram {} #[async_trait] impl Actor for DocProcessor { + const RECORD_DURATION_METRICS: bool = true; + type ObservableState = Arc; fn observable_state(&self) -> Self::ObservableState { diff --git a/quickwit/quickwit-indexing/src/actors/index_serializer.rs b/quickwit/quickwit-indexing/src/actors/index_serializer.rs index c9c1a5561f2..9e6aa9a4a7c 100644 --- a/quickwit/quickwit-indexing/src/actors/index_serializer.rs +++ b/quickwit/quickwit-indexing/src/actors/index_serializer.rs @@ -49,6 +49,8 @@ impl IndexSerializer { #[async_trait] impl Actor for IndexSerializer { + const RECORD_DURATION_METRICS: bool = true; + type ObservableState = (); fn observable_state(&self) -> Self::ObservableState {} diff --git a/quickwit/quickwit-indexing/src/actors/indexer.rs b/quickwit/quickwit-indexing/src/actors/indexer.rs index 6cf97ab8fbf..4542dc3386b 100644 --- a/quickwit/quickwit-indexing/src/actors/indexer.rs +++ b/quickwit/quickwit-indexing/src/actors/indexer.rs @@ -53,6 +53,7 @@ use ulid::Ulid; use super::IndexSerializer; use super::cooperative_indexing::{CooperativeIndexingCycle, CooperativeIndexingPeriod}; use crate::docs_clustering::{DocIdClusterer, Fingerprinter}; +use crate::duration_metrics::{IndexingOperation, measure_future_duration}; use crate::metrics::SPLIT_BUILDERS; use crate::models::{ CommitTrigger, EmptySplit, IndexedSplitBatchBuilder, IndexedSplitBuilder, NewPublishLock, @@ -204,8 +205,12 @@ impl IndexerState { let cooperative_indexing_period = if let Some(cooperative_indexing) = &self.cooperative_indexing_opt { Some( - ctx.protect_future(cooperative_indexing.cooperative_indexing_period()) - .await, + measure_future_duration( + ctx.actor_name(), + IndexingOperation::IndexingPermitWait, + ctx.protect_future(cooperative_indexing.cooperative_indexing_period()), + ) + .await, ) } else { None @@ -214,13 +219,16 @@ impl IndexerState { let last_delete_opstamp_request = LastDeleteOpstampRequest { index_uid: Some(self.pipeline_id.index_uid.clone()), }; - let last_delete_opstamp_response = ctx - .protect_future( + let last_delete_opstamp_response = measure_future_duration( + ctx.actor_name(), + IndexingOperation::LastDeleteOpstampRequest, + ctx.protect_future( self.metastore .clone() .last_delete_opstamp(last_delete_opstamp_request), - ) - .await?; + ), + ) + .await?; let last_delete_opstamp = last_delete_opstamp_response.last_delete_opstamp; let checkpoint_delta = IndexCheckpointDelta { @@ -380,6 +388,8 @@ pub struct Indexer { #[async_trait] impl Actor for Indexer { + const RECORD_DURATION_METRICS: bool = true; + type ObservableState = IndexerCounters; fn observable_state(&self) -> Self::ObservableState { diff --git a/quickwit/quickwit-indexing/src/actors/packager.rs b/quickwit/quickwit-indexing/src/actors/packager.rs index 821c90a7968..68568bd92ce 100644 --- a/quickwit/quickwit-indexing/src/actors/packager.rs +++ b/quickwit/quickwit-indexing/src/actors/packager.rs @@ -92,6 +92,8 @@ impl Packager { #[async_trait] impl Actor for Packager { + const RECORD_DURATION_METRICS: bool = true; + type ObservableState = (); #[allow(clippy::unused_unit)] diff --git a/quickwit/quickwit-indexing/src/actors/publisher.rs b/quickwit/quickwit-indexing/src/actors/publisher.rs index db7d52dd479..3694d044636 100644 --- a/quickwit/quickwit-indexing/src/actors/publisher.rs +++ b/quickwit/quickwit-indexing/src/actors/publisher.rs @@ -26,6 +26,7 @@ use serde::Serialize; use tracing::{error, info, instrument, warn}; use crate::actors::MergePlanner; +use crate::duration_metrics::{IndexingOperation, measure_future_duration}; use crate::metrics::record_published_split; use crate::models::{NewSplits, PublishLock, SharedPublishToken, SplitsUpdate}; use crate::source::{SourceActor, SuggestTruncate}; @@ -104,7 +105,12 @@ impl Publisher { ); // The actor kill signal will propagate and eventually end up back here, and will try // to publish before exiting, so we kill the publish lock to prevent one final flush. - publish_lock.kill().await; + measure_future_duration( + ctx.actor_name(), + IndexingOperation::RevokePublishLock, + publish_lock.kill(), + ) + .await; let _ = ctx.send_exit_with_success(source_mailbox).await; Ok(()) } @@ -163,14 +169,25 @@ where Some(Duration::from_secs(3)), None, ] { - let Err(error) = ctx.protect_future(publish()).await else { + let Err(error) = measure_future_duration( + ctx.actor_name(), + IndexingOperation::PublishSplits, + ctx.protect_future(publish()), + ) + .await + else { return Ok(()); }; let retryable = matches!(error, MetastoreError::InvalidPublishToken { .. }); match retry_delay { Some(retry_delay) if retryable => { warn!(%error, operation = operation_name, "metastore publish failed, retrying"); - ctx.protect_future(ctx.sleep(retry_delay)).await; + measure_future_duration( + ctx.actor_name(), + IndexingOperation::PublishRetryDelay, + ctx.protect_future(ctx.sleep(retry_delay)), + ) + .await; } _ => { warn!(%error, operation = operation_name, retryable, "metastore publish failed, giving up after 3 tries"); @@ -185,6 +202,8 @@ where #[async_trait] impl Actor for Publisher { + const RECORD_DURATION_METRICS: bool = true; + type ObservableState = PublisherCounters; fn observable_state(&self) -> Self::ObservableState { @@ -245,7 +264,13 @@ impl Handler for Publisher { .iter() .map(|split| split.split_id.to_string()) .collect(); - let Some(guard) = publish_lock.acquire().await else { + let Some(guard) = measure_future_duration( + ctx.actor_name(), + IndexingOperation::AcquirePublishLock, + publish_lock.acquire(), + ) + .await + else { info!( split_ids=?split_ids, "Splits' publish lock is dead." diff --git a/quickwit/quickwit-indexing/src/actors/sequencer.rs b/quickwit/quickwit-indexing/src/actors/sequencer.rs index 5b801a245da..f2ddb3255f5 100644 --- a/quickwit/quickwit-indexing/src/actors/sequencer.rs +++ b/quickwit/quickwit-indexing/src/actors/sequencer.rs @@ -19,6 +19,8 @@ use async_trait::async_trait; use quickwit_actors::{Actor, ActorContext, ActorExitStatus, Handler, Mailbox, QueueCapacity}; use tokio::sync::oneshot; +use crate::duration_metrics::{IndexingOperation, measure_future_duration}; + /// The sequencer serves as a proxy to another actor, /// delivering message in a specific order. /// @@ -49,6 +51,8 @@ impl Sequencer { #[async_trait] impl Actor for Sequencer { + const RECORD_DURATION_METRICS: bool = A::RECORD_DURATION_METRICS; + type ObservableState = (); fn queue_capacity(&self) -> QueueCapacity { @@ -80,10 +84,18 @@ where message: oneshot::Receiver>, ctx: &ActorContext, ) -> Result<(), ActorExitStatus> { - let command = ctx - .protect_future(message) + let receive_command = ctx.protect_future(message); + let command = if Self::RECORD_DURATION_METRICS { + measure_future_duration( + ctx.actor_name(), + IndexingOperation::UploadResultWait, + receive_command, + ) .await - .context("failed to receive command from uploader")?; + } else { + receive_command.await + } + .context("failed to receive command from uploader")?; if let SequencerCommand::Proceed(msg) = command { ctx.send_message(&self.mailbox, msg) .await @@ -99,6 +111,13 @@ mod tests { use super::*; + #[test] + #[allow(clippy::assertions_on_constants)] + fn test_duration_metrics_follow_target_opt_in() { + assert!(Sequencer::::RECORD_DURATION_METRICS); + assert!(!Sequencer::::RECORD_DURATION_METRICS); + } + #[derive(Default)] struct SequencerTestActor { messages: Vec, diff --git a/quickwit/quickwit-indexing/src/actors/uploader.rs b/quickwit/quickwit-indexing/src/actors/uploader.rs index 9ee80f35610..5aad2fabef3 100644 --- a/quickwit/quickwit-indexing/src/actors/uploader.rs +++ b/quickwit/quickwit-indexing/src/actors/uploader.rs @@ -42,6 +42,7 @@ use tracing::{Instrument, Span, debug, error, info, instrument, warn}; use crate::actors::Publisher; use crate::actors::sequencer::{Sequencer, SequencerCommand}; +use crate::duration_metrics::{IndexingOperation, measure_future_duration}; use crate::merge_policy::{MergePolicy, MergeTask}; use crate::metrics::{AVAILABLE_CONCURRENT_UPLOAD_PERMITS, COMPONENT}; use crate::models::{ @@ -221,10 +222,13 @@ impl Uploader { let concurrent_upload_permits = concurrent_upload_permits_once_cell .get_or_init(|| Semaphore::const_new(self.max_concurrent_split_uploads)); concurrent_upload_permits_gauge.set(concurrent_upload_permits.available_permits() as f64); - concurrent_upload_permits - .acquire() - .await - .context("the uploader semaphore is closed. (this should never happen)") + measure_future_duration( + ctx.actor_name(), + IndexingOperation::UploadPermitWait, + concurrent_upload_permits.acquire(), + ) + .await + .context("the uploader semaphore is closed. (this should never happen)") } } @@ -236,6 +240,8 @@ pub struct UploaderCounters { #[async_trait] impl Actor for Uploader { + const RECORD_DURATION_METRICS: bool = true; + type ObservableState = UploaderCounters; #[allow(clippy::unused_unit)] @@ -302,117 +308,137 @@ impl Handler for Uploader { debug!(split_ids=?split_ids, "start-stage-and-store-splits"); let event_broker = self.event_broker.clone(); spawn_named_task( + // This task outlives the uploader handler and can run concurrently with other uploads. async move { - fail_point!("uploader:intask:before"); - - let mut split_metadata_list = Vec::with_capacity(batch.splits.len()); - let mut split_payloads = Vec::with_capacity(batch.splits.len()); - let mut report_splits: Vec = Vec::with_capacity(batch.splits.len()); - - for packaged_split in batch.splits.iter() { - if batch.publish_lock.is_dead() { - // TODO: Remove the junk right away? - info!("splits' publish lock is dead"); - - if let Err(error) = split_update_sender.discard() { - error!(?error, "failed to discard split"); + measure_future_duration( + ctx_clone.actor_name(), + IndexingOperation::UploadTask, + async { + fail_point!("uploader:intask:before"); + + let mut split_metadata_list = Vec::with_capacity(batch.splits.len()); + let mut split_payloads = Vec::with_capacity(batch.splits.len()); + let mut report_splits: Vec = + Vec::with_capacity(batch.splits.len()); + + for packaged_split in batch.splits.iter() { + if batch.publish_lock.is_dead() { + // TODO: Remove the junk right away? + info!("splits' publish lock is dead"); + + if let Err(error) = split_update_sender.discard() { + error!(?error, "failed to discard split"); + } + return; + } + + let (split_metadata, split_payload) = match prepare_split_for_upload( + packaged_split, + &merge_policy, + retention_policy.as_ref(), + ) { + Ok(prepared_split) => prepared_split, + Err(error) => { + error!( + ?error, + split_id = packaged_split.split_id(), + "failed to prepare split for upload" + ); + return; + } + }; + + report_splits.push(ReportSplit { + storage_uri: split_store.remote_uri().to_string(), + split_id: packaged_split.split_id().to_string(), + }); + + split_metadata_list.push(split_metadata); + split_payloads.push(split_payload); } - return; - } - - let (split_metadata, split_payload) = match prepare_split_for_upload( - packaged_split, - &merge_policy, - retention_policy.as_ref(), - ) { - Ok(prepared_split) => prepared_split, - Err(error) => { - error!( - ?error, - split_id = packaged_split.split_id(), - "failed to prepare split for upload" - ); + + let stage_splits_request = + match StageSplitsRequest::try_from_splits_metadata( + index_uid.clone(), + split_metadata_list.clone(), + ) { + Ok(stage_splits_request) => stage_splits_request, + Err(error) => { + error!(?error, "failed to create stage splits request"); + return; + } + }; + if let Err(error) = measure_future_duration( + ctx_clone.actor_name(), + IndexingOperation::StageSplits, + metastore.stage_splits(stage_splits_request), + ) + .await + { + error!(?error, "failed to stage splits"); return; + }; + + counters + .num_staged_splits + .fetch_add(split_metadata_list.len() as u64, Ordering::Relaxed); + + let mut packaged_splits_and_metadata = + Vec::with_capacity(batch.splits.len()); + + event_broker.publish(ReportSplitsRequest { report_splits }); + + for ((packaged_split, metadata), split_payload) in batch + .splits + .into_iter() + .zip(split_metadata_list) + .zip(split_payloads) + { + let upload_result = upload_split( + &packaged_split, + &metadata, + split_payload, + &split_store, + counters.clone(), + ctx_clone.actor_name(), + ) + .await; + + if let Err(error) = upload_result { + kill_switch.kill_with_fault(error.context(format!( + "failed to upload split `{}`", + packaged_split.split_id() + ))); + return; + } + packaged_splits_and_metadata.push((packaged_split, metadata)); } - }; - - report_splits.push(ReportSplit { - storage_uri: split_store.remote_uri().to_string(), - split_id: packaged_split.split_id().to_string(), - }); - - split_metadata_list.push(split_metadata); - split_payloads.push(split_payload); - } - let stage_splits_request = match StageSplitsRequest::try_from_splits_metadata( - index_uid.clone(), - split_metadata_list.clone(), - ) { - Ok(stage_splits_request) => stage_splits_request, - Err(error) => { - error!(?error, "failed to create stage splits request"); - return; - } - }; - if let Err(error) = metastore.stage_splits(stage_splits_request).await { - error!(?error, "failed to stage splits"); - return; - }; - - counters - .num_staged_splits - .fetch_add(split_metadata_list.len() as u64, Ordering::Relaxed); - - let mut packaged_splits_and_metadata = Vec::with_capacity(batch.splits.len()); - - event_broker.publish(ReportSplitsRequest { report_splits }); - - for ((packaged_split, metadata), split_payload) in batch - .splits - .into_iter() - .zip(split_metadata_list) - .zip(split_payloads) - { - let upload_result = upload_split( - &packaged_split, - &metadata, - split_payload, - &split_store, - counters.clone(), - ) - .await; - - if let Err(error) = upload_result { - kill_switch.kill_with_fault(error.context(format!( - "failed to upload split `{}`", - packaged_split.split_id() - ))); - return; - } - packaged_splits_and_metadata.push((packaged_split, metadata)); - } - - let splits_update = make_publish_operation( - index_uid, - packaged_splits_and_metadata, - batch.checkpoint_delta_opt, - batch.publish_lock, - batch.merge_task_opt, - batch.batch_parent_span, - ); - - let target = match &split_update_sender { - SplitsUpdateSender::Sequencer(_) => "sequencer", - SplitsUpdateSender::Publisher(_) => "publisher", - }; - if let Err(error) = split_update_sender.send(splits_update, &ctx_clone).await { - error!(?error, "failed to send splits update to {target}"); - return; - } - // We explicitly drop it in order to force move the permit guard into the async - // task. - mem::drop(permit_guard); + let splits_update = make_publish_operation( + index_uid, + packaged_splits_and_metadata, + batch.checkpoint_delta_opt, + batch.publish_lock, + batch.merge_task_opt, + batch.batch_parent_span, + ); + + let target = match &split_update_sender { + SplitsUpdateSender::Sequencer(_) => "sequencer", + SplitsUpdateSender::Publisher(_) => "publisher", + }; + if let Err(error) = + split_update_sender.send(splits_update, &ctx_clone).await + { + error!(?error, "failed to send splits update to {target}"); + return; + } + // We explicitly drop it in order to force move the permit guard into the + // async task. + mem::drop(permit_guard); + }, + ) + .await; } .instrument(Span::current()), "upload_single_task", @@ -569,14 +595,18 @@ async fn upload_split( split_payload: SplitPayload, split_store: &IndexingSplitStore, counters: UploaderCounters, + actor_name: &str, ) -> anyhow::Result<()> { - split_store - .store_split( + measure_future_duration( + actor_name, + IndexingOperation::StoreSplit, + split_store.store_split( split_metadata, packaged_split.split_scratch_directory.path(), Box::new(split_payload), - ) - .await?; + ), + ) + .await?; counters.num_uploaded_splits.fetch_add(1, Ordering::SeqCst); Ok(()) } diff --git a/quickwit/quickwit-indexing/src/duration_metrics.rs b/quickwit/quickwit-indexing/src/duration_metrics.rs new file mode 100644 index 00000000000..6e0ced06cc7 --- /dev/null +++ b/quickwit/quickwit-indexing/src/duration_metrics.rs @@ -0,0 +1,109 @@ +// 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. + +//! Elapsed indexing-operation durations, not CPU or isolated I/O time. +//! `quickwit_indexing_operation_duration_seconds{actor,operation}` uses the actor name captured +//! at mailbox creation. Spawned upload tasks use their owning uploader's name, not a worker name. +//! Task totals overlap nested operations and are separate from generic actor work totals. +//! Independently computed histogram quantiles must not be subtracted from each other. + +use std::future::Future; + +use quickwit_metrics::{LazyHistogram, histogram, lazy_histogram}; +use tokio::time::Instant; + +static INDEXING_OPERATION_DURATION_SECONDS: LazyHistogram = lazy_histogram!( + name: "operation_duration_seconds", + description: "Elapsed seconds for indexing operations, by owning actor and operation", + subsystem: "indexing", + 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 IndexingOperation { + /// Wait for an indexing permit and create its cooperative indexing period. + IndexingPermitWait, + /// Wait for an upload permit. + UploadPermitWait, + /// Wait for an upload result or discard command in the sequencer. + UploadResultWait, + /// Acquire the publish lock, including waiting for its current holder. + AcquirePublishLock, + /// Revoke the publish lock, including waiting for its current holder. + RevokePublishLock, + /// Sleep before retrying split publication. + PublishRetryDelay, + /// Request the last delete opstamp from the metastore. + LastDeleteOpstampRequest, + /// Stage split metadata in the metastore. + StageSplits, + /// Execute one split publication attempt against the metastore. + PublishSplits, + /// Acquire shards through the metastore request. + AcquireShards, + /// Subscribe to a shard's fetch stream. + SubscribeFetchStream, + /// Wait for the next fetch-stream message, including timeout completion. + FetchMessageWait, + /// Request legacy ingest records, including mailbox delivery and response waiting. + FetchRecordsRequest, + /// Request legacy ingest truncation, including mailbox delivery and response waiting. + TruncateRecordsRequest, + /// Store a split, excluding the uploaded-split counter update. + StoreSplit, + /// Run the entire spawned upload task, including preparation, staging, storage, and sending. + UploadTask, +} + +impl IndexingOperation { + fn as_str(self) -> &'static str { + match self { + Self::IndexingPermitWait => "indexing_permit_wait", + Self::UploadPermitWait => "upload_permit_wait", + Self::UploadResultWait => "upload_result_wait", + Self::AcquirePublishLock => "acquire_publish_lock", + Self::RevokePublishLock => "revoke_publish_lock", + Self::PublishRetryDelay => "publish_retry_delay", + Self::LastDeleteOpstampRequest => "last_delete_opstamp_request", + Self::StageSplits => "stage_splits", + Self::PublishSplits => "publish_splits", + Self::AcquireShards => "acquire_shards", + Self::SubscribeFetchStream => "subscribe_fetch_stream", + Self::FetchMessageWait => "fetch_message_wait", + Self::FetchRecordsRequest => "fetch_records_request", + Self::TruncateRecordsRequest => "truncate_records_request", + Self::StoreSplit => "store_split", + Self::UploadTask => "upload_task", + } + } +} + +/// Observe a future's elapsed duration only on completion, including errors. +/// Cancellation and panic do not produce samples. Callers control opt-in and must use the +/// owning actor's captured mailbox name, never an instance identifier. +pub(crate) async fn measure_future_duration( + actor_name: &str, + operation: IndexingOperation, + future: F, +) -> F::Output { + let started_at = Instant::now(); + let output = future.await; + let duration = started_at.elapsed(); + histogram!(parent: INDEXING_OPERATION_DURATION_SECONDS, "actor" => actor_name.to_owned(), "operation" => operation.as_str()) + .observe(duration.as_secs_f64()); + output +} diff --git a/quickwit/quickwit-indexing/src/lib.rs b/quickwit/quickwit-indexing/src/lib.rs index 288490b4a6d..8f5ebd625ed 100644 --- a/quickwit/quickwit-indexing/src/lib.rs +++ b/quickwit/quickwit-indexing/src/lib.rs @@ -42,6 +42,7 @@ pub use crate::split_store::{ pub mod actors; mod controlled_directory; pub mod docs_clustering; +mod duration_metrics; pub mod merge_policy; mod metrics; pub mod models; diff --git a/quickwit/quickwit-indexing/src/source/ingest/mod.rs b/quickwit/quickwit-indexing/src/source/ingest/mod.rs index ba27af469b2..9e95dced64a 100644 --- a/quickwit/quickwit-indexing/src/source/ingest/mod.rs +++ b/quickwit/quickwit-indexing/src/source/ingest/mod.rs @@ -50,6 +50,7 @@ use super::{ SourceRuntime, TypedSourceFactory, }; use crate::actors::DocProcessor; +use crate::duration_metrics::{IndexingOperation, measure_future_duration}; use crate::models::{LocalShardPositionsUpdate, NewPublishLock, PublishLock, SharedPublishToken}; pub struct IngestSourceFactory; @@ -432,7 +433,12 @@ impl IngestSource { ); self.assigned_shards.clear(); self.fetch_stream.reset(); - self.publish_lock.kill().await; + measure_future_duration( + ctx.actor_name(), + IndexingOperation::RevokePublishLock, + self.publish_lock.kill(), + ) + .await; self.publish_lock = PublishLock::default(); ctx.send_message( doc_processor_mailbox, @@ -455,7 +461,14 @@ impl Source for IngestSource { let now = time::Instant::now(); let deadline = now + *EMIT_BATCHES_TIMEOUT; loop { - match time::timeout_at(deadline, self.fetch_stream.next()).await { + // Include timeout completion: an idle source still waits for a fetch message. + match measure_future_duration( + ctx.actor_name(), + IndexingOperation::FetchMessageWait, + time::timeout_at(deadline, self.fetch_stream.next()), + ) + .await + { Ok(Ok(fetch_message)) => match fetch_message.message { Some(fetch_message::Message::Payload(fetch_payload)) => { self.process_fetch_payload(&mut batch_builder, fetch_payload)?; @@ -542,10 +555,13 @@ impl Source for IngestSource { shard_ids: shard_ids_to_acquire.clone(), publish_token: publish_token.to_string(), }; - let acquire_shards_response: AcquireShardsResponse = ctx - .protect_future(self.metastore.acquire_shards(acquire_shards_request)) - .await - .context("failed to acquire shards")?; + let acquire_shards_response: AcquireShardsResponse = measure_future_duration( + ctx.actor_name(), + IndexingOperation::AcquireShards, + ctx.protect_future(self.metastore.acquire_shards(acquire_shards_request)), + ) + .await + .context("failed to acquire shards")?; self.publish_token.store(Some(Arc::new(publish_token))); if acquire_shards_response.acquired_shards.len() != shard_ids_to_acquire.len() { @@ -585,15 +601,18 @@ impl Source for IngestSource { let status = if from_position_exclusive.is_eof() { IndexingStatus::Complete - } else if let Err(error) = ctx - .protect_future(self.fetch_stream.subscribe( + } else if let Err(error) = measure_future_duration( + ctx.actor_name(), + IndexingOperation::SubscribeFetchStream, + ctx.protect_future(self.fetch_stream.subscribe( ingester_id.clone(), index_uid, source_id, shard_id.clone(), from_position_exclusive, - )) - .await + )), + ) + .await { if let IngestV2Error::ShardNotFound { .. } = error { error!("failed to subscribe to shard `{shard_id}`: shard not found"); diff --git a/quickwit/quickwit-indexing/src/source/ingest_api_source.rs b/quickwit/quickwit-indexing/src/source/ingest_api_source.rs index 7f7cfc973c8..3a0ac9fe384 100644 --- a/quickwit/quickwit-indexing/src/source/ingest_api_source.rs +++ b/quickwit/quickwit-indexing/src/source/ingest_api_source.rs @@ -31,6 +31,7 @@ use tracing::{error, info}; use super::{BatchBuilder, Source, SourceContext, TypedSourceFactory}; use crate::actors::DocProcessor; +use crate::duration_metrics::{IndexingOperation, measure_future_duration}; use crate::source::SourceRuntime; /// Wait time for SourceActor before pooling for new documents. @@ -123,8 +124,12 @@ impl IngestApiSource { index_id: self.source_runtime.index_id().to_string(), up_to_position_included, }; - ctx.ask_for_res(&self.ingest_api_service, suggest_truncate_req) - .await?; + measure_future_duration( + ctx.actor_name(), + IndexingOperation::TruncateRecordsRequest, + ctx.ask_for_res(&self.ingest_api_service, suggest_truncate_req), + ) + .await?; Ok(()) } @@ -162,10 +167,13 @@ impl Source for IngestApiSource { let FetchResponse { first_position: first_position_opt, doc_batch: doc_batch_opt, - } = ctx - .ask_for_res(&self.ingest_api_service, fetch_req) - .await - .map_err(anyhow::Error::from)?; + } = measure_future_duration( + ctx.actor_name(), + IndexingOperation::FetchRecordsRequest, + ctx.ask_for_res(&self.ingest_api_service, fetch_req), + ) + .await + .map_err(anyhow::Error::from)?; // The `first_position_opt` being none means the doc_batch is empty and there is // no more document available, at least for the time being. diff --git a/quickwit/quickwit-indexing/src/source/mod.rs b/quickwit/quickwit-indexing/src/source/mod.rs index 372c22c72c5..8ead82fab26 100644 --- a/quickwit/quickwit-indexing/src/source/mod.rs +++ b/quickwit/quickwit-indexing/src/source/mod.rs @@ -346,10 +346,13 @@ pub struct AssignShards(pub Assignment); #[async_trait] impl Actor for SourceActor { + const RECORD_DURATION_METRICS: bool = true; + type ObservableState = JsonValue; fn name(&self) -> String { - self.source.name() + // Source::name() may include source IDs; actor names are also used as metric labels. + "SourceActor".to_string() } fn observable_state(&self) -> Self::ObservableState { diff --git a/quickwit/quickwit-indexing/src/source/vec_source.rs b/quickwit/quickwit-indexing/src/source/vec_source.rs index 53807028434..6a86dbc2cba 100644 --- a/quickwit/quickwit-indexing/src/source/vec_source.rs +++ b/quickwit/quickwit-indexing/src/source/vec_source.rs @@ -168,11 +168,12 @@ mod tests { }; let source_runtime = SourceRuntimeBuilder::new(index_uid, source_config).build(); let vec_source = VecSourceFactory::typed_create_source(source_runtime, params).await?; - let vec_source_actor = SourceActor::new(Box::new(vec_source), doc_processor_mailbox); assert_eq!( - vec_source_actor.name(), + vec_source.name(), r#"VecSource { source_id: "test-vec-source" }"# ); + let vec_source_actor = SourceActor::new(Box::new(vec_source), doc_processor_mailbox); + assert_eq!(vec_source_actor.name(), "SourceActor"); let (_vec_source_mailbox, vec_source_handle) = universe.spawn_builder().spawn(vec_source_actor); let (actor_termination, last_observation) = vec_source_handle.join().await;