enhancement(observability): Add metrics to measure transform processing time

Author: bruceg

changelog.d/component_processing_time-metrics.enhancement.md

@@ -0,0 +1,5 @@
+Added the `component_processing_time_seconds` histogram and
+`component_processing_time_mean_seconds` gauge internal_metrics, exposing the
+time an event spends in a single transform including the transform buffer.
+
+authors: bruceg

lib/vector-buffers/src/lib.rs

@@ -112,7 +112,7 @@ impl<T> Bufferable for T where T: InMemoryBufferable + Encodable {}
 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);
+    fn on_send(&self, item: &mut T);
 }
 
 pub trait EventCount {

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

@@ -205,9 +205,13 @@ impl<T: Bufferable> BufferSender<T> {
     }
 
     #[async_recursion]
-    pub async fn send(&mut self, item: T, send_reference: Option<Instant>) -> crate::Result<()> {
+    pub async fn send(
+        &mut self,
+        mut item: T,
+        send_reference: Option<Instant>,
+    ) -> crate::Result<()> {
         if let Some(instrumentation) = self.custom_instrumentation.as_ref() {
-            instrumentation.on_send(&item);
+            instrumentation.on_send(&mut item);
         }
         let item_sizing = self
             .usage_instrumentation

lib/vector-core/src/config/global_options.rs

@@ -156,9 +156,10 @@ pub struct GlobalOptions {
     /// The alpha value for the exponential weighted moving average (EWMA) of transform processing
     /// time metrics.
     ///
-    /// This controls how quickly the `event_processing_time_mean_seconds` gauge responds to new
-    /// observations. Values closer to 1.0 retain more of the previous value, leading to slower
-    /// adjustments. The default value of 0.9 is equivalent to a "half life" of 6-7 measurements.
+    /// This controls how quickly the `component_processing_time_mean_seconds` and
+    /// `event_processing_time_mean_seconds` gauges respond to new observations. Values closer to
+    /// 1.0 retain more of the previous value, leading to slower adjustments. The default value of
+    /// 0.9 is equivalent to a "half life" of 6-7 measurements.
     ///
     /// Must be between 0 and 1 exclusively (0 < alpha < 1).
     #[serde(default, skip_serializing_if = "crate::serde::is_default")]

lib/vector-core/src/event/array.rs

@@ -4,6 +4,7 @@
 
 use std::{iter, slice, sync::Arc, vec};
 
+use chrono::Utc;
 use futures::{Stream, stream};
 #[cfg(test)]
 use quickcheck::{Arbitrary, Gen};
@@ -172,6 +173,15 @@ impl EventArray {
         }
     }
 
+    /// Sets the `last_transform_timestamp` in the metadata for all events in this array. This is
+    /// used to initialize the timestamp before the first transform so that component processing
+    /// time can be measured.
+    pub fn set_last_transform_timestamp(&mut self, timestamp: chrono::DateTime<Utc>) {
+        for mut event in self.iter_events_mut() {
+            event.metadata_mut().set_last_transform_timestamp(timestamp);
+        }
+    }
+
     /// Iterate over references to this array's events.
     pub fn iter_events(&self) -> impl Iterator<Item = EventRef<'_>> {
         match self {

lib/vector-core/src/event/metadata.rs

@@ -84,6 +84,11 @@ pub(super) struct Inner {
     #[derivative(PartialEq = "ignore")]
     #[serde(default, skip)]
     pub(crate) ingest_timestamp: Option<DateTime<Utc>>,
+
+    /// The timestamp when the event last entered a transform buffer.
+    #[derivative(PartialEq = "ignore")]
+    #[serde(default, skip)]
+    pub(crate) last_transform_timestamp: Option<DateTime<Utc>>,
 }
 
 /// Metric Origin metadata for submission to Datadog.
@@ -256,6 +261,17 @@ impl EventMetadata {
     pub fn set_ingest_timestamp(&mut self, timestamp: DateTime<Utc>) {
         self.get_mut().ingest_timestamp = Some(timestamp);
     }
+
+    /// Returns the timestamp of the last transform buffer enqueue operation, if it exists.
+    #[must_use]
+    pub fn last_transform_timestamp(&self) -> Option<DateTime<Utc>> {
+        self.0.last_transform_timestamp
+    }
+
+    /// Sets the transform enqueue timestamp to the provided value.
+    pub fn set_last_transform_timestamp(&mut self, timestamp: DateTime<Utc>) {
+        self.get_mut().last_transform_timestamp = Some(timestamp);
+    }
 }
 
 impl Default for Inner {
@@ -272,6 +288,7 @@ impl Default for Inner {
             datadog_origin_metadata: None,
             source_event_id: Some(Uuid::new_v4()),
             ingest_timestamp: None,
+            last_transform_timestamp: None,
         }
     }
 }

lib/vector-core/src/event/proto.rs

@@ -689,6 +689,7 @@ impl From<Metadata> for EventMetadata {
             datadog_origin_metadata,
             source_event_id,
             ingest_timestamp: None,
+            last_transform_timestamp: None,
         }))
     }
 }

src/topology/builder.rs

@@ -6,6 +6,7 @@ use std::{
     time::Instant,
 };
 
+use chrono::Utc;
 use futures::{FutureExt, StreamExt, TryStreamExt, stream::FuturesOrdered};
 use futures_util::stream::FuturesUnordered;
 use metrics::gauge;
@@ -35,7 +36,7 @@ use vector_vrl_metrics::MetricsStorage;
 use super::{
     BuiltBuffer, ConfigDiff,
     fanout::{self, Fanout},
-    processing_time::ProcessingTimeRecorder,
+    processing_time::{ComponentProcessingTimeRecorder, ProcessingTimeRecorder},
     schema,
     task::{Task, TaskOutput, TaskResult},
 };
@@ -289,6 +290,7 @@ impl<'a> Builder<'a> {
                     {
                         array.set_output_id(&source);
                         array.set_source_type(source_type);
+                        array.set_last_transform_timestamp(Utc::now());
                         fanout
                             .send(array, Some(send_reference))
                             .await
@@ -525,14 +527,18 @@ impl<'a> Builder<'a> {
             };
 
             let metrics = ChannelMetricMetadata::new(TRANSFORM_CHANNEL_METRIC_PREFIX, None);
-            let (input_tx, input_rx) = TopologyBuilder::standalone_memory(
+            let (mut input_tx, input_rx) = TopologyBuilder::standalone_memory(
                 TOPOLOGY_BUFFER_SIZE,
                 WhenFull::Block,
                 &span,
                 Some(metrics),
                 self.config.global.buffer_utilization_ewma_alpha,
             );
 
+            input_tx.with_custom_instrumentation(ComponentProcessingTimeRecorder::new(
+                self.config.global.processing_time_ewma_alpha,
+            ));
+
             self.inputs
                 .insert(key.clone(), (input_tx, node.inputs.clone()));
 

src/topology/processing_time.rs

@@ -11,6 +11,8 @@ use crate::{config::ComponentKey, event::EventArray};
 const NANOS_PER_SECOND: f64 = 1_000_000_000.0;
 const EVENT_PROCESSING_TIME: &str = "event_processing_time_seconds";
 const EVENT_PROCESSING_TIME_MEAN: &str = "event_processing_time_mean_seconds";
+const COMPONENT_PROCESSING_TIME: &str = "component_processing_time_seconds";
+const COMPONENT_PROCESSING_TIME_MEAN: &str = "component_processing_time_mean_seconds";
 const DEFAULT_PROCESSING_TIME_EWMA_ALPHA: f64 = 0.9;
 
 #[cfg(test)]
@@ -87,7 +89,7 @@ impl ProcessingTimeRecorder {
 }
 
 impl BufferInstrumentation<EventArray> for ProcessingTimeRecorder {
-    fn on_send(&self, events: &EventArray) {
+    fn on_send(&self, events: &mut EventArray) {
         self.record_events(events);
     }
 }
@@ -124,6 +126,47 @@ impl Metrics {
     }
 }
 
+#[derive(Debug)]
+pub(crate) struct ComponentProcessingTimeRecorder {
+    histogram: Histogram,
+    gauge: EwmaGauge,
+}
+
+impl ComponentProcessingTimeRecorder {
+    pub(crate) fn new(ewma_alpha: Option<f64>) -> Self {
+        Self {
+            histogram: histogram!(COMPONENT_PROCESSING_TIME),
+            gauge: EwmaGauge::new(
+                gauge!(COMPONENT_PROCESSING_TIME_MEAN),
+                Some(ewma_alpha.unwrap_or(DEFAULT_PROCESSING_TIME_EWMA_ALPHA)),
+            ),
+        }
+    }
+
+    fn record(&self, latency_seconds: f64) {
+        self.histogram.record(latency_seconds);
+        self.gauge.record(latency_seconds);
+    }
+}
+
+impl BufferInstrumentation<EventArray> for ComponentProcessingTimeRecorder {
+    fn on_send(&self, events: &mut EventArray) {
+        let now = Utc::now();
+
+        for mut event in events.iter_events_mut() {
+            let metadata = event.metadata_mut();
+            if let Some(previous) = metadata.last_transform_timestamp()
+                && let Some(latency_ns) = now.signed_duration_since(previous).num_nanoseconds()
+                && latency_ns >= 0
+            {
+                self.record(latency_ns as f64 / NANOS_PER_SECOND);
+            }
+
+            metadata.set_last_transform_timestamp(now);
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

src/topology/test/processing_time.rs

@@ -14,24 +14,67 @@ use crate::{
     },
 };
 
+const EVENT_COUNT: usize = 100;
+const SOURCE_ID: &str = "latency_source";
+const TRANSFORM_ID: &str = "latency_delay";
+const TRANSFORM_TYPE: &str = "test_noop";
+const TRANSFORM_KIND: &str = "transform";
+const SINK_ID: &str = "latency_sink";
+
+struct ProcessingTimeTestRun {
+    metrics: Vec<Metric>,
+    elapsed_time: f64,
+}
+
 #[tokio::test]
 async fn sink_processing_time_metrics_emitted() {
+    let run = run_processing_time_topology().await;
+
+    assert_histogram_count(
+        &run.metrics,
+        "event_processing_time_seconds",
+        has_latency_tags,
+    );
+    assert_gauge_range(
+        &run.metrics,
+        "event_processing_time_mean_seconds",
+        has_latency_tags,
+        run.elapsed_time,
+    );
+}
+
+#[tokio::test]
+async fn component_processing_time_metrics_emitted() {
+    let run = run_processing_time_topology().await;
+
+    assert_histogram_count(
+        &run.metrics,
+        "component_processing_time_seconds",
+        has_component_tags,
+    );
+    assert_gauge_range(
+        &run.metrics,
+        "component_processing_time_mean_seconds",
+        has_component_tags,
+        run.elapsed_time,
+    );
+}
+
+async fn run_processing_time_topology() -> ProcessingTimeTestRun {
     trace_init();
 
     let controller = Controller::get().expect("metrics controller");
     controller.reset();
 
-    const EVENT_COUNT: usize = 100;
-
     let (mut source_tx, source_config) = basic_source();
     let transform_config = noop_transform();
     let (sink_done_tx, sink_done_rx) = oneshot::channel();
     let sink_config = completion_sink(EVENT_COUNT, sink_done_tx);
 
     let mut config = Config::builder();
-    config.add_source("latency_source", source_config);
-    config.add_transform("latency_delay", &["latency_source"], transform_config);
-    config.add_sink("latency_sink", &["latency_delay"], sink_config);
+    config.add_source(SOURCE_ID, source_config);
+    config.add_transform(TRANSFORM_ID, &[SOURCE_ID], transform_config);
+    config.add_sink(SINK_ID, &[TRANSFORM_ID], sink_config);
 
     let start_time = Instant::now();
     let (topology, _) = start_topology(config.build().unwrap(), false).await;
@@ -55,17 +98,17 @@ async fn sink_processing_time_metrics_emitted() {
     topology.stop().await;
     let elapsed_time = start_time.elapsed().as_secs_f64();
 
-    let metrics = controller.capture_metrics();
-    let sink_id = "latency_sink";
-    let source_id = "latency_source";
+    ProcessingTimeTestRun {
+        metrics: controller.capture_metrics(),
+        elapsed_time,
+    }
+}
 
+fn assert_histogram_count(metrics: &[Metric], metric_name: &str, tags_match: fn(&Metric) -> bool) {
     let histogram = metrics
         .iter()
-        .find(|metric| {
-            metric.name() == "event_processing_time_seconds"
-                && has_latency_tags(metric, sink_id, source_id)
-        })
-        .expect("event_processing_time_seconds histogram missing");
+        .find(|metric| metric.name() == metric_name && tags_match(metric))
+        .unwrap_or_else(|| panic!("{metric_name} histogram missing"));
 
     match histogram.value() {
         MetricValue::AggregatedHistogram { count, .. } => {
@@ -76,14 +119,18 @@ async fn sink_processing_time_metrics_emitted() {
         }
         other => panic!("expected aggregated histogram, got {other:?}"),
     }
+}
 
+fn assert_gauge_range(
+    metrics: &[Metric],
+    metric_name: &str,
+    tags_match: fn(&Metric) -> bool,
+    elapsed_time: f64,
+) {
     let gauge = metrics
         .iter()
-        .find(|metric| {
-            metric.name() == "event_processing_time_mean_seconds"
-                && has_latency_tags(metric, sink_id, source_id)
-        })
-        .expect("event_processing_time_mean_seconds gauge missing");
+        .find(|metric| metric.name() == metric_name && tags_match(metric))
+        .unwrap_or_else(|| panic!("{metric_name} gauge missing"));
 
     match gauge.value() {
         MetricValue::Gauge { value } => {
@@ -100,9 +147,17 @@ async fn sink_processing_time_metrics_emitted() {
     }
 }
 
-fn has_latency_tags(metric: &Metric, sink: &str, source: &str) -> bool {
+fn has_latency_tags(metric: &Metric) -> bool {
+    metric.tags().is_some_and(|tags| {
+        tags.get("source_component_id") == Some(SOURCE_ID)
+            && tags.get("sink_component_id") == Some(SINK_ID)
+    })
+}
+
+fn has_component_tags(metric: &Metric) -> bool {
     metric.tags().is_some_and(|tags| {
-        tags.get("source_component_id") == Some(source)
-            && tags.get("sink_component_id") == Some(sink)
+        tags.get("component_id") == Some(TRANSFORM_ID)
+            && tags.get("component_type") == Some(TRANSFORM_TYPE)
+            && tags.get("component_kind") == Some(TRANSFORM_KIND)
     })
 }