enhancement(observability): Add metrics to measure total event processing time (#24481)

* chore(core): Add `ingest_timestamp` to `EventMetadata`

We want to record the time at which events are ingested in order to track
latency as each travels through the topology. This is currently recorded for log
events using the Vector namespace in the `vector.ingest_timestamp` metadata
field, but we want it to be usable for all event types.  As such, we need a new
field in `struct EventMetadata`.

This change adds the new field as an `Option` so as to retain sane semantics for
`Default` implementations and to avoid extra calls to `Utc::now`.  The
`SourceSender::send` method sets this if the source doesn't so as to ensure
complete coverage.

For backward compatibility, this metadata is still inserted into the Vector log
namespace metadata, taken from this new field. Since this metadata is set up
before passing the events to the `SourceSender`, the ingest timestamp is set
manually in sources that can create Vector namespace logs.

* Add new `EwmaGauge` wrapper to handle both in one update step

* enhancement(observability): Add metrics to measure total event processing time

This adds an optional `trait BufferInstrumentation` hook to `struct
BufferSender` which is called at the very start of the buffer send path. We use
that hook to take the previously added universal event `ingest_timestamp`
metadata and from it calculate the total time spent processing the event,
including buffering delays. This time is emitted in internal metrics as a
`event_processing_time_seconds` histogram and
`event_processing_time_mean_seconds` gauge, the latter using an EWMA to smooth
the mean over time.

* Update wording on the lognamespacing tutorial docs.

* Pre-create metrics to avoid extraneous labels

* Fix tests

* Move setting ingest timestamp back into `insert_standard_vector_source_metadata`

* Assert an upper limit on the average latency gauge too

Author: bruceg

Cargo.lock

@@ -213,6 +213,7 @@ serial_test = { version = "3.2" }
 cfg-if.workspace = true
 clap.workspace = true
 clap_complete.workspace = true
+dashmap.workspace = true
 indoc.workspace = true
 paste.workspace = true
 pin-project.workspace = true

Cargo.toml

@@ -0,0 +1,4 @@
+Added the `event_processing_time_seconds` histogram and `event_processing_time_mean_seconds` gauge internal_metrics,
+exposing the total time events spend between the originating source and final sink in a topology.
+
+authors: bruceg

changelog.d/event_processing_time_metrics.enhancement.md

@@ -112,7 +112,9 @@ separate namespace to the event data.
 
 The actual value to be placed into the field.
 
-For the ingest timestamp this will be `chrono::Utc::now()`. Source type will be
+The ingest timestamp should be recorded once per batch (e.g. `let now =
+chrono::Utc::now();`) and passed into `insert_standard_vector_source_metadata`,
+which takes care of updating the event metadata. The source type will be
 the `NAME` property of the `Config` struct. `NAME` is provided by the
 `configurable_component` macro. You may need to include `use vector_config::NamedComponent;`.
 

docs/tutorials/lognamespacing.md

@@ -108,6 +108,13 @@ pub trait Bufferable: InMemoryBufferable + Encodable {}
 // Blanket implementation for anything that is already bufferable.
 impl<T> Bufferable for T where T: InMemoryBufferable + Encodable {}
 
+/// Hook for observing items as they are sent into a `BufferSender`.
+pub trait BufferInstrumentation<T: Bufferable>: Send + Sync + 'static {
+    /// Called immediately before the item is emitted to the underlying buffer.
+    /// The underlying type is stored in an `Arc`, so we cannot have `&mut self`.
+    fn on_send(&self, item: &T);
+}
+
 pub trait EventCount {
     fn event_count(&self) -> usize;
 }

lib/vector-buffers/src/lib.rs

@@ -16,15 +16,10 @@ use crossbeam_queue::{ArrayQueue, SegQueue};
 use futures::Stream;
 use metrics::{Gauge, Histogram, gauge, histogram};
 use tokio::sync::{Notify, OwnedSemaphorePermit, Semaphore, TryAcquireError};
-use vector_common::stats::AtomicEwma;
+use vector_common::stats::EwmaGauge;
 
 use crate::{InMemoryBufferable, config::MemoryBufferSize};
 
-/// The alpha value for the Exponentially Weighted Moving Average (EWMA) calculation. This is a
-/// measure of how much weight to give to the current value versus the previous values. A value of
-/// 0.9 results in a "half life" of 6-7 measurements.
-pub const DEFAULT_EWMA_ALPHA: f64 = 0.9;
-
 /// Error returned by `LimitedSender::send` when the receiver has disconnected.
 #[derive(Debug, PartialEq, Eq)]
 pub struct SendError<T>(pub T);
@@ -114,8 +109,7 @@ impl ChannelMetricMetadata {
 struct Metrics {
     histogram: Histogram,
     gauge: Gauge,
-    mean_gauge: Gauge,
-    ewma: Arc<AtomicEwma>,
+    mean_gauge: EwmaGauge,
     // We hold a handle to the max gauge to avoid it being dropped by the metrics collector, but
     // since the value is static, we never need to update it. The compiler detects this as an unused
     // field, so we need to suppress the warning here.
@@ -150,37 +144,36 @@ impl Metrics {
         let histogram_name = format!("{prefix}_utilization");
         let gauge_name = format!("{prefix}_utilization_level");
         let mean_name = format!("{prefix}_utilization_mean");
-        let ewma = Arc::new(AtomicEwma::new(ewma_alpha.unwrap_or(DEFAULT_EWMA_ALPHA)));
         #[cfg(test)]
         let recorded_values = Arc::new(Mutex::new(Vec::new()));
         if let Some(label_value) = output {
             let max_gauge = gauge!(max_gauge_name, "output" => label_value.clone());
             max_gauge.set(max_value);
+            let mean_gauge_handle = gauge!(mean_name, "output" => label_value.clone());
             // DEPRECATED: buffer-bytes-events-metrics
             let legacy_max_gauge = gauge!(legacy_max_gauge_name, "output" => label_value.clone());
             legacy_max_gauge.set(max_value);
             Self {
                 histogram: histogram!(histogram_name, "output" => label_value.clone()),
                 gauge: gauge!(gauge_name, "output" => label_value.clone()),
-                mean_gauge: gauge!(mean_name, "output" => label_value.clone()),
+                mean_gauge: EwmaGauge::new(mean_gauge_handle, ewma_alpha),
                 max_gauge,
-                ewma,
                 legacy_max_gauge,
                 #[cfg(test)]
                 recorded_values,
             }
         } else {
             let max_gauge = gauge!(max_gauge_name);
             max_gauge.set(max_value);
+            let mean_gauge_handle = gauge!(mean_name);
             // DEPRECATED: buffer-bytes-events-metrics
             let legacy_max_gauge = gauge!(legacy_max_gauge_name);
             legacy_max_gauge.set(max_value);
             Self {
                 histogram: histogram!(histogram_name),
                 gauge: gauge!(gauge_name),
-                mean_gauge: gauge!(mean_name),
+                mean_gauge: EwmaGauge::new(mean_gauge_handle, ewma_alpha),
                 max_gauge,
-                ewma,
                 legacy_max_gauge,
                 #[cfg(test)]
                 recorded_values,
@@ -192,8 +185,7 @@ impl Metrics {
     fn record(&self, value: usize) {
         self.histogram.record(value as f64);
         self.gauge.set(value as f64);
-        let avg = self.ewma.update(value as f64);
-        self.mean_gauge.set(avg);
+        self.mean_gauge.record(value as f64);
         #[cfg(test)]
         if let Ok(mut recorded) = self.recorded_values.lock() {
             recorded.push(value);

lib/vector-buffers/src/topology/channel/limited_queue.rs

@@ -3,10 +3,11 @@ mod receiver;
 mod sender;
 
 pub use limited_queue::{
-    ChannelMetricMetadata, DEFAULT_EWMA_ALPHA, LimitedReceiver, LimitedSender, SendError, limited,
+    ChannelMetricMetadata, LimitedReceiver, LimitedSender, SendError, limited,
 };
 pub use receiver::*;
 pub use sender::*;
+pub use vector_common::stats::DEFAULT_EWMA_ALPHA;
 
 #[cfg(test)]
 mod tests;

lib/vector-buffers/src/topology/channel/mod.rs

@@ -8,7 +8,7 @@ use vector_common::internal_event::{InternalEventHandle, Registered, register};
 
 use super::limited_queue::LimitedSender;
 use crate::{
-    Bufferable, WhenFull,
+    BufferInstrumentation, Bufferable, WhenFull,
     buffer_usage_data::BufferUsageHandle,
     internal_events::BufferSendDuration,
     variants::disk_v2::{self, ProductionFilesystem},
@@ -134,9 +134,11 @@ pub struct BufferSender<T: Bufferable> {
     base: SenderAdapter<T>,
     overflow: Option<Box<BufferSender<T>>>,
     when_full: WhenFull,
-    instrumentation: Option<BufferUsageHandle>,
+    usage_instrumentation: Option<BufferUsageHandle>,
     #[derivative(Debug = "ignore")]
     send_duration: Option<Registered<BufferSendDuration>>,
+    #[derivative(Debug = "ignore")]
+    custom_instrumentation: Option<Arc<dyn BufferInstrumentation<T>>>,
 }
 
 impl<T: Bufferable> BufferSender<T> {
@@ -146,8 +148,9 @@ impl<T: Bufferable> BufferSender<T> {
             base,
             overflow: None,
             when_full,
-            instrumentation: None,
+            usage_instrumentation: None,
             send_duration: None,
+            custom_instrumentation: None,
         }
     }
 
@@ -157,8 +160,9 @@ impl<T: Bufferable> BufferSender<T> {
             base,
             overflow: Some(Box::new(overflow)),
             when_full: WhenFull::Overflow,
-            instrumentation: None,
+            usage_instrumentation: None,
             send_duration: None,
+            custom_instrumentation: None,
         }
     }
 
@@ -174,14 +178,19 @@ impl<T: Bufferable> BufferSender<T> {
 
     /// Configures this sender to instrument the items passing through it.
     pub fn with_usage_instrumentation(&mut self, handle: BufferUsageHandle) {
-        self.instrumentation = Some(handle);
+        self.usage_instrumentation = Some(handle);
     }
 
     /// Configures this sender to instrument the send duration.
     pub fn with_send_duration_instrumentation(&mut self, stage: usize, span: &Span) {
         let _enter = span.enter();
         self.send_duration = Some(register(BufferSendDuration { stage }));
     }
+
+    /// Configures this sender to invoke a custom instrumentation hook.
+    pub fn with_custom_instrumentation(&mut self, instrumentation: impl BufferInstrumentation<T>) {
+        self.custom_instrumentation = Some(Arc::new(instrumentation));
+    }
 }
 
 impl<T: Bufferable> BufferSender<T> {
@@ -197,14 +206,17 @@ impl<T: Bufferable> BufferSender<T> {
 
     #[async_recursion]
     pub async fn send(&mut self, item: T, send_reference: Option<Instant>) -> crate::Result<()> {
+        if let Some(instrumentation) = self.custom_instrumentation.as_ref() {
+            instrumentation.on_send(&item);
+        }
         let item_sizing = self
-            .instrumentation
+            .usage_instrumentation
             .as_ref()
             .map(|_| (item.event_count(), item.size_of()));
 
         let mut was_dropped = false;
 
-        if let Some(instrumentation) = self.instrumentation.as_ref()
+        if let Some(instrumentation) = self.usage_instrumentation.as_ref()
             && let Some((item_count, item_size)) = item_sizing
         {
             instrumentation
@@ -229,7 +241,7 @@ impl<T: Bufferable> BufferSender<T> {
             }
         }
 
-        if let Some(instrumentation) = self.instrumentation.as_ref()
+        if let Some(instrumentation) = self.usage_instrumentation.as_ref()
             && let Some((item_count, item_size)) = item_sizing
             && was_dropped
         {

lib/vector-buffers/src/topology/channel/sender.rs

@@ -0,0 +1,32 @@
+use std::sync::Arc;
+
+use metrics::Gauge;
+
+use super::AtomicEwma;
+
+/// The default alpha parameter used when constructing EWMA-backed gauges.
+pub const DEFAULT_EWMA_ALPHA: f64 = 0.9;
+
+/// Couples a [`Gauge`] with an [`AtomicEwma`] so gauge readings reflect the EWMA.
+#[derive(Clone, Debug)]
+pub struct EwmaGauge {
+    gauge: Gauge,
+    // Note that the `Gauge` internally is equivalent to an `Arc<AtomicF64>` so we need to use the
+    // same semantics for the EWMA calculation as well.
+    ewma: Arc<AtomicEwma>,
+}
+
+impl EwmaGauge {
+    #[must_use]
+    pub fn new(gauge: Gauge, alpha: Option<f64>) -> Self {
+        let alpha = alpha.unwrap_or(DEFAULT_EWMA_ALPHA);
+        let ewma = Arc::new(AtomicEwma::new(alpha));
+        Self { gauge, ewma }
+    }
+
+    /// Records a new value, updates the EWMA, and sets the gauge accordingly.
+    pub fn record(&self, value: f64) {
+        let average = self.ewma.update(value);
+        self.gauge.set(average);
+    }
+}

lib/vector-common/src/stats/ewma_gauge.rs

@@ -1,5 +1,9 @@
 #![allow(missing_docs)]
 
+pub mod ewma_gauge;
+
+pub use ewma_gauge::{DEFAULT_EWMA_ALPHA, EwmaGauge};
+
 use std::sync::atomic::Ordering;
 
 use crate::atomic::AtomicF64;