feat(sources): add source latency metric and fix source lag time on large batches (vectordotdev#24987)

* add send latency distribution to source sender

* add changelog

* use outputMetrics in the builder

* fix clippy

* apply suggestion

* only record the await point duration

* make sure metrics are updated if an error is returned

* add documentation for the added metrics

* fix source_lag_time_seconds measurement for large batches

Author: gwenaskell

changelog.d/fix_source_lag_time_chunked_batches.fix.md

@@ -0,0 +1,3 @@
+Fixed an incorrect source_lag_time_seconds measurement in sources that use `send_batch` with large event batches. When a batch was split into multiple chunks, the reference timestamp used to compute lag time was re-captured on each chunk send, causing the lag time for later chunks to be overstated by the amount of time spent waiting for the channel to accept earlier chunks. The reference timestamp is now captured once before iteration and shared across all chunks.
+
+authors: gwenaskell

changelog.d/source_sender_latency_metrics.enhancement.md

@@ -0,0 +1,5 @@
+Sources now record the distribution metrics `source_send_latency_seconds` (measuring the time spent
+blocking on a single events chunk send operation on the output) and `source_send_batch_latency_seconds`
+(encompassing all chunks within a received events batch).
+
+authors: gwenaskell

lib/vector-core/src/source_sender/builder.rs

@@ -1,17 +1,20 @@
 use std::{collections::HashMap, time::Duration};
 
-use metrics::{Histogram, histogram};
+use metrics::histogram;
 use vector_buffers::topology::channel::LimitedReceiver;
 use vector_common::internal_event::DEFAULT_OUTPUT;
 
-use super::{CHUNK_SIZE, LAG_TIME_NAME, Output, SourceSender, SourceSenderItem};
+use super::{
+    CHUNK_SIZE, LAG_TIME_NAME, Output, OutputMetrics, SEND_BATCH_LATENCY_NAME, SEND_LATENCY_NAME,
+    SourceSender, SourceSenderItem,
+};
 use crate::config::{ComponentKey, OutputId, SourceOutput};
 
 pub struct Builder {
     buf_size: usize,
     default_output: Option<Output>,
     named_outputs: HashMap<String, Output>,
-    lag_time: Option<Histogram>,
+    output_metrics: OutputMetrics,
     timeout: Option<Duration>,
     ewma_half_life_seconds: Option<f64>,
 }
@@ -22,7 +25,11 @@ impl Default for Builder {
             buf_size: CHUNK_SIZE,
             default_output: None,
             named_outputs: Default::default(),
-            lag_time: Some(histogram!(LAG_TIME_NAME)),
+            output_metrics: OutputMetrics::new(
+                Some(histogram!(LAG_TIME_NAME)),
+                Some(histogram!(SEND_LATENCY_NAME)),
+                Some(histogram!(SEND_BATCH_LATENCY_NAME)),
+            ),
             timeout: None,
             ewma_half_life_seconds: None,
         }
@@ -53,7 +60,6 @@ impl Builder {
         output: SourceOutput,
         component_key: ComponentKey,
     ) -> LimitedReceiver<SourceSenderItem> {
-        let lag_time = self.lag_time.clone();
         let log_definition = output.schema_definition.clone();
         let output_id = OutputId {
             component: component_key,
@@ -64,7 +70,7 @@ impl Builder {
                 let (output, rx) = Output::new_with_buffer(
                     self.buf_size,
                     DEFAULT_OUTPUT.to_owned(),
-                    lag_time,
+                    self.output_metrics.clone(),
                     log_definition,
                     output_id,
                     self.timeout,
@@ -77,7 +83,7 @@ impl Builder {
                 let (output, rx) = Output::new_with_buffer(
                     self.buf_size,
                     name.clone(),
-                    lag_time,
+                    self.output_metrics.clone(),
                     log_definition,
                     output_id,
                     self.timeout,

lib/vector-core/src/source_sender/mod.rs

@@ -14,7 +14,7 @@ mod tests;
 
 pub use builder::Builder;
 pub use errors::SendError;
-use output::Output;
+use output::{Output, OutputMetrics};
 pub use sender::{SourceSender, SourceSenderItem};
 
 pub const CHUNK_SIZE: usize = 1000;
@@ -23,3 +23,5 @@ pub const CHUNK_SIZE: usize = 1000;
 const TEST_BUFFER_SIZE: usize = 100;
 
 const LAG_TIME_NAME: &str = "source_lag_time_seconds";
+const SEND_LATENCY_NAME: &str = "source_send_latency_seconds";
+const SEND_BATCH_LATENCY_NAME: &str = "source_send_batch_latency_seconds";

lib/vector-core/src/source_sender/output.rs

@@ -85,7 +85,7 @@ impl Drop for UnsentEventCount {
 #[derive(Clone)]
 pub(super) struct Output {
     sender: LimitedSender<SourceSenderItem>,
-    lag_time: Option<Histogram>,
+    metrics: OutputMetrics,
     events_sent: Registered<EventsSent>,
     /// The schema definition that will be attached to Log events sent through here
     log_definition: Option<Arc<Definition>>,
@@ -95,6 +95,27 @@ pub(super) struct Output {
     timeout: Option<Duration>,
 }
 
+#[derive(Clone, Default)]
+pub(super) struct OutputMetrics {
+    lag_time: Option<Histogram>,
+    send_latency: Option<Histogram>,
+    send_batch_latency: Option<Histogram>,
+}
+
+impl OutputMetrics {
+    pub(super) fn new(
+        lag_time: Option<Histogram>,
+        send_latency: Option<Histogram>,
+        send_batch_latency: Option<Histogram>,
+    ) -> Self {
+        Self {
+            lag_time,
+            send_latency,
+            send_batch_latency,
+        }
+    }
+}
+
 #[expect(clippy::missing_fields_in_debug)]
 impl fmt::Debug for Output {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
@@ -111,19 +132,20 @@ impl Output {
     pub(super) fn new_with_buffer(
         n: usize,
         output: String,
-        lag_time: Option<Histogram>,
+        metrics: OutputMetrics,
         log_definition: Option<Arc<Definition>>,
         output_id: OutputId,
         timeout: Option<Duration>,
         ewma_half_life_seconds: Option<f64>,
     ) -> (Self, LimitedReceiver<SourceSenderItem>) {
         let limit = MemoryBufferSize::MaxEvents(NonZeroUsize::new(n).unwrap());
-        let metrics = ChannelMetricMetadata::new(UTILIZATION_METRIC_PREFIX, Some(output.clone()));
-        let (tx, rx) = channel::limited(limit, Some(metrics), ewma_half_life_seconds);
+        let channel_metrics =
+            ChannelMetricMetadata::new(UTILIZATION_METRIC_PREFIX, Some(output.clone()));
+        let (tx, rx) = channel::limited(limit, Some(channel_metrics), ewma_half_life_seconds);
         (
             Self {
                 sender: tx,
-                lag_time,
+                metrics,
                 events_sent: internal_event::register(EventsSent::from(internal_event::Output(
                     Some(output.into()),
                 ))),
@@ -136,12 +158,21 @@ impl Output {
     }
 
     pub(super) async fn send(
+        &mut self,
+        events: EventArray,
+        unsent_event_count: &mut UnsentEventCount,
+    ) -> Result<(), SendError> {
+        let reference = Utc::now().timestamp_millis();
+        self.send_inner(events, unsent_event_count, reference).await
+    }
+
+    async fn send_inner(
         &mut self,
         mut events: EventArray,
         unsent_event_count: &mut UnsentEventCount,
+        reference: i64,
     ) -> Result<(), SendError> {
         let send_reference = Instant::now();
-        let reference = Utc::now().timestamp_millis();
         events
             .iter_events()
             .for_each(|event| self.emit_lag_time(event, reference));
@@ -156,7 +187,17 @@ impl Output {
 
         let byte_size = events.estimated_json_encoded_size_of();
         let count = events.len();
-        self.send_with_timeout(events, send_reference).await?;
+
+        let send_start = Instant::now();
+
+        let send_result = self.send_with_timeout(events, send_reference).await;
+
+        if let Some(send_latency) = &self.metrics.send_latency {
+            send_latency.record(send_start.elapsed().as_secs_f64());
+        }
+
+        send_result?;
+
         self.events_sent.emit(CountByteSize(count, byte_size));
         unsent_event_count.decr(count);
         Ok(())
@@ -218,33 +259,47 @@ impl Output {
         I: IntoIterator<Item = E>,
         <I as IntoIterator>::IntoIter: ExactSizeIterator,
     {
+        // Capture a single reference timestamp for the entire batch so that lag time
+        // measurements are not inflated by channel-send latency for later chunks.
+        let reference = Utc::now().timestamp_millis();
+
         // It's possible that the caller stops polling this future while it is blocked waiting
         // on `self.send()`. When that happens, we use `UnsentEventCount` to correctly emit
         // `ComponentEventsDropped` events.
         let events = events.into_iter().map(Into::into);
         let mut unsent_event_count = UnsentEventCount::new(events.len());
+        let send_batch_start = Instant::now();
+
         for events in array::events_into_arrays(events, Some(CHUNK_SIZE)) {
-            self.send(events, &mut unsent_event_count)
+            self.send_inner(events, &mut unsent_event_count, reference)
                 .await
-                .inspect_err(|error| match error {
-                    SendError::Timeout => {
-                        unsent_event_count.timed_out();
+                .inspect_err(|error| {
+                    match error {
+                        SendError::Timeout => {
+                            unsent_event_count.timed_out();
+                        }
+                        SendError::Closed => {
+                            // The unsent event count is discarded here because the callee emits the
+                            // `StreamClosedError`.
+                            unsent_event_count.discard();
+                        }
                     }
-                    SendError::Closed => {
-                        // The unsent event count is discarded here because the callee emits the
-                        // `StreamClosedError`.
-                        unsent_event_count.discard();
+                    if let Some(send_batch_latency) = &self.metrics.send_batch_latency {
+                        send_batch_latency.record(send_batch_start.elapsed().as_secs_f64());
                     }
                 })?;
         }
+        if let Some(send_batch_latency) = &self.metrics.send_batch_latency {
+            send_batch_latency.record(send_batch_start.elapsed().as_secs_f64());
+        }
         Ok(())
     }
 
     /// Calculate the difference between the reference time and the
     /// timestamp stored in the given event reference, and emit the
     /// different, as expressed in milliseconds, as a histogram.
     pub(super) fn emit_lag_time(&self, event: EventRef<'_>, reference: i64) {
-        if let Some(lag_time_metric) = &self.lag_time {
+        if let Some(lag_time_metric) = &self.metrics.lag_time {
             let timestamp = match event {
                 EventRef::Log(log) => {
                     log_schema()

lib/vector-core/src/source_sender/sender.rs

@@ -22,7 +22,9 @@ use vector_common::{
 
 use super::{Builder, Output, SendError};
 #[cfg(any(test, feature = "test"))]
-use super::{LAG_TIME_NAME, TEST_BUFFER_SIZE};
+use super::{
+    LAG_TIME_NAME, OutputMetrics, SEND_BATCH_LATENCY_NAME, SEND_LATENCY_NAME, TEST_BUFFER_SIZE,
+};
 use crate::{
     EstimatedJsonEncodedSizeOf,
     event::{Event, EventArray, EventContainer, array::EventArrayIntoIter},
@@ -108,14 +110,16 @@ impl SourceSender {
         timeout: Option<Duration>,
     ) -> (Self, LimitedReceiver<SourceSenderItem>) {
         let lag_time = Some(histogram!(LAG_TIME_NAME));
+        let send_latency = Some(histogram!(SEND_LATENCY_NAME));
+        let send_batch_latency = Some(histogram!(SEND_BATCH_LATENCY_NAME));
         let output_id = OutputId {
             component: "test".to_string().into(),
             port: None,
         };
         let (default_output, rx) = Output::new_with_buffer(
             n,
             DEFAULT_OUTPUT.to_owned(),
-            lag_time,
+            OutputMetrics::new(lag_time, send_latency, send_batch_latency),
             None,
             output_id,
             timeout,
@@ -192,8 +196,15 @@ impl SourceSender {
             component: "test".to_string().into(),
             port: Some(name.clone()),
         };
-        let (output, recv) =
-            Output::new_with_buffer(100, name.clone(), None, None, output_id, None, None);
+        let (output, recv) = Output::new_with_buffer(
+            100,
+            name.clone(),
+            OutputMetrics::default(),
+            None,
+            output_id,
+            None,
+            None,
+        );
         let recv = recv.into_stream().map(move |mut item| {
             item.events.iter_events_mut().for_each(|mut event| {
                 let metadata = event.metadata_mut();

website/cue/reference/components/sources/internal_metrics.cue

@@ -774,6 +774,18 @@ components: sources: internal_metrics: {
 			default_namespace: "vector"
 			tags:              _component_tags
 		}
+		source_send_batch_latency_seconds: {
+			description:       "The time elapsed blocking on the downstream channel to accept an entire batch of events received at the source"
+			type:              "histogram"
+			default_namespace: "vector"
+			tags:              _component_tags
+		}
+		source_send_latency_seconds: {
+			description:       "The time elapsed blocking on the downstream channel to accept a single chunk from a batch of events received at the source"
+			type:              "histogram"
+			default_namespace: "vector"
+			tags:              _component_tags
+		}
 		source_buffer_max_byte_size: {
 			description:       "The maximum number of bytes the source buffer can hold. The outputs of the source send data to this buffer."
 			type:              "gauge"