chore(observability): Refactor EWMA + Gauge into a new struct (#24556)

Author: bruceg

lib/vector-buffers/src/topology/channel/limited_queue.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/mod.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-common/src/stats/ewma_gauge.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.rs lib/vector-common/src/stats/mod.rslib/vector-common/src/stats.rs renamed to lib/vector-common/src/stats/mod.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;