enhancement(observability)!: Time-weight buffer utilization means (#24697)

* enhancement(observability): Time-weight buffer utilization means

Under a fully loaded pipeline, the typical source will batch its received events
into arrays, the size of which is currently capped at 1,000 events. The
channel buffers for transforms, however, have a capacity limit of 100 events.
This means their actual utilization under non-trivial load will bounce between 0
and 1,000 events, and the weighted mean value will just hover around 500. This
is true both when infrequent batches are sent and when the pipeline is at
capacity, making that mean value useless.

This change adds a duration-weighting calculation to the mean that weights
values held for longer times more than those held for shorter times. This will
cause the above two situations to reflect much different values in the mean.

* Adjustments to docs

Author: bruceg

changelog.d/time-weighted-buffer-utilization-mean.breaking.md

@@ -0,0 +1,9 @@
+The `*buffer_utilization_mean` metrics have been enhanced to use time-weighted
+averaging which make them more representative of the actual buffer utilization
+over time.
+
+This change is breaking due to the replacement of the existing
+`buffer_utilization_ewma_alpha` config option with
+`buffer_utilization_ewma_half_life_seconds`.
+
+authors: bruceg

lib/vector-buffers/src/topology/builder.rs

@@ -191,13 +191,13 @@ impl<T: Bufferable> TopologyBuilder<T> {
         when_full: WhenFull,
         receiver_span: &Span,
         metadata: Option<ChannelMetricMetadata>,
-        ewma_alpha: Option<f64>,
+        ewma_half_life_seconds: Option<f64>,
     ) -> (BufferSender<T>, BufferReceiver<T>) {
         let usage_handle = BufferUsageHandle::noop();
         usage_handle.set_buffer_limits(None, Some(max_events.get()));
 
         let limit = MemoryBufferSize::MaxEvents(max_events);
-        let (sender, receiver) = limited(limit, metadata, ewma_alpha);
+        let (sender, receiver) = limited(limit, metadata, ewma_half_life_seconds);
 
         let mode = match when_full {
             WhenFull::Overflow => WhenFull::Block,

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

@@ -7,6 +7,7 @@ use std::{
         Arc,
         atomic::{AtomicUsize, Ordering},
     },
+    time::Instant,
 };
 
 #[cfg(test)]
@@ -17,10 +18,12 @@ 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::EwmaGauge;
+use vector_common::stats::TimeEwmaGauge;
 
 use crate::{InMemoryBufferable, config::MemoryBufferSize};
 
+pub const DEFAULT_EWMA_HALF_LIFE_SECONDS: f64 = 5.0;
+
 /// Error returned by `LimitedSender::send` when the receiver has disconnected.
 #[derive(Debug, PartialEq, Eq)]
 pub struct SendError<T>(pub T);
@@ -110,7 +113,7 @@ impl ChannelMetricMetadata {
 struct Metrics {
     histogram: Histogram,
     gauge: Gauge,
-    mean_gauge: EwmaGauge,
+    mean_gauge: TimeEwmaGauge,
     // 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.
@@ -127,8 +130,10 @@ impl Metrics {
     fn new(
         limit: MemoryBufferSize,
         metadata: ChannelMetricMetadata,
-        ewma_alpha: Option<f64>,
+        ewma_half_life_seconds: Option<f64>,
     ) -> Self {
+        let ewma_half_life_seconds =
+            ewma_half_life_seconds.unwrap_or(DEFAULT_EWMA_HALF_LIFE_SECONDS);
         let ChannelMetricMetadata { prefix, output } = metadata;
         let (legacy_suffix, gauge_suffix, max_value) = match limit {
             MemoryBufferSize::MaxEvents(max_events) => (
@@ -157,7 +162,7 @@ impl Metrics {
             Self {
                 histogram: histogram!(histogram_name, "output" => label_value.clone()),
                 gauge: gauge!(gauge_name, "output" => label_value.clone()),
-                mean_gauge: EwmaGauge::new(mean_gauge_handle, ewma_alpha),
+                mean_gauge: TimeEwmaGauge::new(mean_gauge_handle, ewma_half_life_seconds),
                 max_gauge,
                 legacy_max_gauge,
                 #[cfg(test)]
@@ -173,7 +178,7 @@ impl Metrics {
             Self {
                 histogram: histogram!(histogram_name),
                 gauge: gauge!(gauge_name),
-                mean_gauge: EwmaGauge::new(mean_gauge_handle, ewma_alpha),
+                mean_gauge: TimeEwmaGauge::new(mean_gauge_handle, ewma_half_life_seconds),
                 max_gauge,
                 legacy_max_gauge,
                 #[cfg(test)]
@@ -183,10 +188,10 @@ impl Metrics {
     }
 
     #[expect(clippy::cast_precision_loss)]
-    fn record(&self, value: usize) {
+    fn record(&self, value: usize, reference: Instant) {
         self.histogram.record(value as f64);
         self.gauge.set(value as f64);
-        self.mean_gauge.record(value as f64);
+        self.mean_gauge.record(value as f64, reference);
         #[cfg(test)]
         if let Ok(mut recorded) = self.recorded_values.lock() {
             recorded.push(value);
@@ -221,10 +226,11 @@ impl<T: Send + Sync + Debug + 'static> Inner<T> {
     fn new(
         limit: MemoryBufferSize,
         metric_metadata: Option<ChannelMetricMetadata>,
-        ewma_alpha: Option<f64>,
+        ewma_half_life_seconds: Option<f64>,
     ) -> Self {
         let read_waker = Arc::new(Notify::new());
-        let metrics = metric_metadata.map(|metadata| Metrics::new(limit, metadata, ewma_alpha));
+        let metrics =
+            metric_metadata.map(|metadata| Metrics::new(limit, metadata, ewma_half_life_seconds));
         match limit {
             MemoryBufferSize::MaxEvents(max_events) => Inner {
                 data: Arc::new(ArrayQueue::new(max_events.get())),
@@ -256,7 +262,7 @@ impl<T: Send + Sync + Debug + 'static> Inner<T> {
             // acquired fewer permits than their true size, `size` is the correct utilization since
             // the queue must have been empty for the oversized acquire to succeed.
             let utilization = size.max(self.used_capacity());
-            metrics.record(utilization);
+            metrics.record(utilization, Instant::now());
         }
         self.data.push((permits, item));
         self.read_waker.notify_one();
@@ -275,7 +281,7 @@ impl<T> Inner<T> {
                 // been released yet, used_capacity is stable against racing senders acquiring those
                 // permits.
                 let utilization = self.used_capacity().saturating_sub(permit.num_permits());
-                metrics.record(utilization);
+                metrics.record(utilization, Instant::now());
             }
             // Release permits after recording so a waiting sender cannot enqueue a new item
             // before this pop's utilization measurement is taken.
@@ -446,9 +452,9 @@ impl<T> Drop for LimitedReceiver<T> {
 pub fn limited<T: InMemoryBufferable + fmt::Debug>(
     limit: MemoryBufferSize,
     metric_metadata: Option<ChannelMetricMetadata>,
-    ewma_alpha: Option<f64>,
+    ewma_half_life_seconds: Option<f64>,
 ) -> (LimitedSender<T>, LimitedReceiver<T>) {
-    let inner = Inner::new(limit, metric_metadata, ewma_alpha);
+    let inner = Inner::new(limit, metric_metadata, ewma_half_life_seconds);
 
     let sender = LimitedSender {
         inner: inner.clone(),

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

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

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

@@ -1,11 +1,9 @@
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
+use std::time::Instant;
 
 use metrics::Gauge;
 
-use super::AtomicEwma;
-
-/// The default alpha parameter used when constructing EWMA-backed gauges.
-pub const DEFAULT_EWMA_ALPHA: f64 = 0.9;
+use super::{AtomicEwma, TimeEwma};
 
 /// Couples a [`Gauge`] with an [`AtomicEwma`] so gauge readings reflect the EWMA.
 #[derive(Clone, Debug)]
@@ -19,7 +17,7 @@ pub struct EwmaGauge {
 impl EwmaGauge {
     #[must_use]
     pub fn new(gauge: Gauge, alpha: Option<f64>) -> Self {
-        let alpha = alpha.unwrap_or(DEFAULT_EWMA_ALPHA);
+        let alpha = alpha.unwrap_or(super::DEFAULT_EWMA_ALPHA);
         let ewma = Arc::new(AtomicEwma::new(alpha));
         Self { gauge, ewma }
     }
@@ -30,3 +28,31 @@ impl EwmaGauge {
         self.gauge.set(average);
     }
 }
+
+/// Couples a [`Gauge`] with a [`TimeEwma`] so gauge readings reflect the EWMA. Since `TimeEwma` has
+/// an internal state consisting of multiple values, this gauge requires a mutex to protect the
+/// state update.
+#[derive(Clone, Debug)]
+pub struct TimeEwmaGauge {
+    gauge: Gauge,
+    ewma: Arc<Mutex<TimeEwma>>,
+}
+
+impl TimeEwmaGauge {
+    #[must_use]
+    pub fn new(gauge: Gauge, half_life_seconds: f64) -> Self {
+        let ewma = Arc::new(Mutex::new(TimeEwma::new(half_life_seconds)));
+        Self { gauge, ewma }
+    }
+
+    /// Records a new value, updates the EWMA, and sets the gauge accordingly.
+    ///
+    /// # Panics
+    ///
+    /// Panics if the EWMA mutex is poisoned.
+    pub fn record(&self, value: f64, reference: Instant) {
+        let mut ewma = self.ewma.lock().expect("time ewma gauge mutex poisoned");
+        let average = ewma.update(value, reference);
+        self.gauge.set(average);
+    }
+}

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

@@ -1,13 +1,18 @@
 #![allow(missing_docs)]
 
-pub mod ewma_gauge;
+mod ewma_gauge;
+mod time_ewma;
 
-pub use ewma_gauge::{DEFAULT_EWMA_ALPHA, EwmaGauge};
+pub use ewma_gauge::{EwmaGauge, TimeEwmaGauge};
+pub use time_ewma::TimeEwma;
 
 use std::sync::atomic::Ordering;
 
 use crate::atomic::AtomicF64;
 
+/// The default alpha parameter used when constructing EWMA-backed gauges.
+pub const DEFAULT_EWMA_ALPHA: f64 = 0.9;
+
 /// Exponentially Weighted Moving Average
 #[derive(Clone, Copy, Debug)]
 pub struct Ewma {

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

@@ -0,0 +1,84 @@
+use std::time::Instant;
+
+#[derive(Clone, Copy, Debug)]
+struct State {
+    average: f64,
+    point: f64,
+    reference: Instant,
+}
+
+/// Continuous-Time Exponentially Weighted Moving Average.
+///
+/// This is used to average values that are observed at irregular intervals but have a fixed value
+/// between the observations, AKA a piecewise-constant signal sampled at change points. Instead of
+/// an "alpha" parameter, this uses a "half-life" parameter which is the time it takes for the
+/// average to decay to half of its value, measured in seconds.
+#[derive(Clone, Copy, Debug)]
+pub struct TimeEwma {
+    state: Option<State>,
+    half_life_seconds: f64,
+}
+
+impl TimeEwma {
+    #[must_use]
+    pub const fn new(half_life_seconds: f64) -> Self {
+        Self {
+            state: None,
+            half_life_seconds,
+        }
+    }
+
+    #[must_use]
+    pub fn average(&self) -> Option<f64> {
+        self.state.map(|state| state.average)
+    }
+
+    /// Update the current average and return it for convenience. Note that this average will "lag"
+    /// the current observation because the new average is based on the previous point and the
+    /// duration during which it was held constant. If the reference time is before the previous
+    /// update, the update is ignored and the previous average is returned.
+    pub fn update(&mut self, point: f64, reference: Instant) -> f64 {
+        let average = match self.state {
+            None => point,
+            Some(state) => {
+                if let Some(duration) = reference.checked_duration_since(state.reference) {
+                    let k = (-duration.as_secs_f64() / self.half_life_seconds).exp2();
+                    // The elapsed duration applies to the previously observed point, since that value
+                    // was held constant between observations.
+                    k * state.average + (1.0 - k) * state.point
+                } else {
+                    state.average
+                }
+            }
+        };
+        self.state = Some(State {
+            average,
+            point,
+            reference,
+        });
+        average
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::TimeEwma;
+    use std::time::{Duration, Instant};
+
+    #[test]
+    #[expect(clippy::float_cmp, reason = "exact values for this test")]
+    fn time_ewma_uses_previous_point_duration() {
+        let mut ewma = TimeEwma::new(1.0);
+        let t0 = Instant::now();
+        let t1 = t0 + Duration::from_secs(1);
+        let t2 = t1 + Duration::from_secs(1);
+
+        assert_eq!(ewma.average(), None);
+        assert_eq!(ewma.update(0.0, t0), 0.0);
+        assert_eq!(ewma.average(), Some(0.0));
+        assert_eq!(ewma.update(10.0, t1), 0.0);
+        assert_eq!(ewma.average(), Some(0.0));
+        assert_eq!(ewma.update(10.0, t2), 5.0);
+        assert_eq!(ewma.average(), Some(5.0));
+    }
+}

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

@@ -10,6 +10,16 @@ use super::{
 };
 use crate::serde::bool_or_struct;
 
+#[expect(
+    clippy::ref_option,
+    reason = "we have to follow the serde calling convention"
+)]
+fn is_default_buffer_utilization_ewma_half_life_seconds(value: &Option<f64>) -> bool {
+    value.is_none_or(|seconds| {
+        seconds == vector_buffers::topology::channel::DEFAULT_EWMA_HALF_LIFE_SECONDS
+    })
+}
+
 #[derive(Debug, Snafu)]
 pub(crate) enum DataDirError {
     #[snafu(display("data_dir option required, but not given here or globally"))]
@@ -140,18 +150,25 @@ pub struct GlobalOptions {
     #[serde(skip_serializing_if = "crate::serde::is_default")]
     pub expire_metrics_per_metric_set: Option<Vec<PerMetricSetExpiration>>,
 
-    /// The alpha value for the exponential weighted moving average (EWMA) of source and transform
-    /// buffer utilization metrics.
+    /// The half-life, in seconds, for the exponential weighted moving average (EWMA) of source
+    /// and transform buffer utilization metrics.
     ///
     /// This controls how quickly the `*_buffer_utilization_mean` 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.
+    /// observations. Longer half-lives retain more of the previous value, leading to slower
+    /// adjustments.
     ///
-    /// Must be between 0 and 1 exclusively (0 < alpha < 1).
-    #[serde(default, skip_serializing_if = "crate::serde::is_default")]
-    #[configurable(validation(range(min = 0.0, max = 1.0)))]
+    /// - Lower values (< 1): Metrics update quickly but may be volatile
+    /// - Default (5): Balanced between responsiveness and stability
+    /// - Higher values (> 5): Smooth, stable metrics that update slowly
+    ///
+    /// Adjust based on whether you need fast detection of buffer issues (lower)
+    /// or want to see sustained trends without noise (higher).
+    ///
+    /// Must be greater than 0.
+    #[serde(skip_serializing_if = "is_default_buffer_utilization_ewma_half_life_seconds")]
+    #[configurable(validation(range(min = 0.0)))]
     #[configurable(metadata(docs::advanced))]
-    pub buffer_utilization_ewma_alpha: Option<f64>,
+    pub buffer_utilization_ewma_half_life_seconds: Option<f64>,
 
     /// The alpha value for the exponential weighted moving average (EWMA) of transform latency
     /// metrics.
@@ -321,9 +338,9 @@ impl GlobalOptions {
                 expire_metrics: self.expire_metrics.or(with.expire_metrics),
                 expire_metrics_secs: self.expire_metrics_secs.or(with.expire_metrics_secs),
                 expire_metrics_per_metric_set: merged_expire_metrics_per_metric_set,
-                buffer_utilization_ewma_alpha: self
-                    .buffer_utilization_ewma_alpha
-                    .or(with.buffer_utilization_ewma_alpha),
+                buffer_utilization_ewma_half_life_seconds: self
+                    .buffer_utilization_ewma_half_life_seconds
+                    .or(with.buffer_utilization_ewma_half_life_seconds),
                 latency_ewma_alpha: self.latency_ewma_alpha.or(with.latency_ewma_alpha),
                 metrics_storage_refresh_period: self
                     .metrics_storage_refresh_period