fix(prometheus_exporter sink): allow disabling metric expiration via flush_period_secs=0

changelog.d/23519_prometheus_exporter_disable_expiration.enhancement.md

@@ -0,0 +1,9 @@
+The `prometheus_exporter` sink's `flush_period_secs` option now accepts `0` to disable metric
+expiration entirely. Previously, metrics with sparse or bursty updates (for example, high
+cardinality counters produced by `log_to_metric`) could be expired and re-added as a "new"
+series, causing gaps and apparent counter resets in downstream Prometheus queries even with a
+large `flush_period_secs` configured. Setting `flush_period_secs: 0` keeps all previously seen
+metrics for the lifetime of the sink; be aware this can result in unbounded memory growth if
+metric series cardinality is unbounded.
+
+authors: valerypetrov

src/sinks/prometheus/exporter.rs

@@ -18,7 +18,6 @@ use hyper::{
 };
 use indexmap::{IndexMap, map::Entry};
 use serde_with::serde_as;
-use snafu::Snafu;
 use stream_cancel::{Trigger, Tripwire};
 use tower::ServiceBuilder;
 use tower_http::compression::CompressionLayer;
@@ -48,16 +47,8 @@ use crate::{
     tls::{MaybeTlsSettings, TlsEnableableConfig},
 };
 
-const MIN_FLUSH_PERIOD_SECS: u64 = 1;
-
 const LOCK_FAILED: &str = "Prometheus exporter data lock is poisoned";
 
-#[derive(Debug, Snafu)]
-enum BuildError {
-    #[snafu(display("Flush period for sets must be greater or equal to {} secs", min))]
-    FlushPeriodTooShort { min: u64 },
-}
-
 /// Configuration for the `prometheus_exporter` sink.
 #[serde_as]
 #[configurable_component(sink(
@@ -121,6 +112,10 @@ pub struct PrometheusExporterConfig {
     /// considered expired and is removed.
     ///
     /// Be sure to configure this value higher than your client’s scrape interval.
+    ///
+    /// Set to `0` to disable expiration entirely. Metrics will then accumulate for as long as the
+    /// sink runs, which can result in unbounded memory growth if metric series cardinality is
+    /// unbounded.
     #[serde(default = "default_flush_period_secs")]
     #[serde_as(as = "serde_with::DurationSeconds<u64>")]
     #[configurable(metadata(docs::human_name = "Flush Interval"))]
@@ -186,10 +181,12 @@ impl GenerateConfig for PrometheusExporterConfig {
 #[typetag::serde(name = "prometheus_exporter")]
 impl SinkConfig for PrometheusExporterConfig {
     async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        if self.flush_period_secs.as_secs() < MIN_FLUSH_PERIOD_SECS {
-            return Err(Box::new(BuildError::FlushPeriodTooShort {
-                min: MIN_FLUSH_PERIOD_SECS,
-            }));
+        if self.flush_period_secs.is_zero() {
+            warn!(
+                "Disabling `flush_period_secs` (setting it to `0`) disables metric expiration, \
+                 which can result in unbounded memory growth if metric series cardinality is \
+                 unbounded."
+            );
         }
 
         validate_quantiles(&self.quantiles)?;
@@ -240,8 +237,10 @@ impl MetricMetadata {
     }
 
     /// Whether or not the referenced metric has expired yet.
+    ///
+    /// A zero expiration window means expiration is disabled, so the metric never expires.
     pub fn has_expired(&self, now: Instant) -> bool {
-        now >= self.expires_at
+        !self.expiration_window.is_zero() && now >= self.expires_at
     }
 }
 
@@ -546,7 +545,9 @@ impl StreamSink<Event> for PrometheusExporter {
             // remove an expired metric depending on how things line up.  It'd be cool to _check_
             // for expired metrics more often, but we also don't want to check _way_ too often, like
             // every second, since then we're constantly iterating through every metric, etc etc.
-            if last_flush.elapsed() > self.config.flush_period_secs {
+            //
+            // A zero flush period means expiration is disabled, so we skip sweeping entirely.
+            if !flush_period.is_zero() && last_flush.elapsed() > flush_period {
                 last_flush = Instant::now();
 
                 let mut metrics = self.metrics.write().expect(LOCK_FAILED);
@@ -1445,6 +1446,37 @@ mod tests {
             .expect("gauge metric should exist");
         assert_eq!(actual_gauge.0.value(), expected_gauge.value());
     }
+
+    #[test]
+    fn metric_metadata_never_expires_when_disabled() {
+        let metadata = MetricMetadata::new(std::time::Duration::ZERO);
+
+        // Even long after "now", a metric with expiration disabled should never be considered
+        // expired.
+        let far_future = Instant::now() + std::time::Duration::from_secs(60 * 60 * 24 * 365);
+        assert!(!metadata.has_expired(far_future));
+    }
+
+    #[test]
+    fn metric_metadata_expires_when_enabled() {
+        let metadata = MetricMetadata::new(std::time::Duration::from_secs(1));
+
+        let past_expiration = Instant::now() + std::time::Duration::from_secs(2);
+        assert!(metadata.has_expired(past_expiration));
+    }
+
+    #[tokio::test]
+    async fn build_succeeds_with_flush_period_disabled() {
+        let (_guard, address) = next_addr();
+        let config = PrometheusExporterConfig {
+            address,
+            tls: None,
+            flush_period_secs: std::time::Duration::ZERO,
+            ..Default::default()
+        };
+
+        assert!(config.build(SinkContext::default()).await.is_ok());
+    }
 }
 
 #[cfg(all(test, feature = "prometheus-integration-tests"))]
@@ -1526,6 +1558,7 @@ mod integration_tests {
         time::sleep(time::Duration::from_millis(500)).await;
         reset_on_flush_period().await;
         expire_on_flush_period().await;
+        never_expire_when_flush_period_disabled().await;
     }
 
     async fn prometheus_scrapes_metrics() {
@@ -1670,4 +1703,45 @@ mod integration_tests {
         drop(tx);
         sink_handle.await.unwrap();
     }
+
+    async fn never_expire_when_flush_period_disabled() {
+        let config = PrometheusExporterConfig {
+            address: sink_exporter_address().parse().unwrap(),
+            // A flush period of zero disables metric expiration entirely.
+            flush_period_secs: Duration::from_secs(0),
+            ..Default::default()
+        };
+        let (sink, _) = config.build(SinkContext::default()).await.unwrap();
+        let (tx, rx) = mpsc::unbounded_channel();
+        let input_events = UnboundedReceiverStream::new(rx);
+
+        let input_events = input_events.map(Into::into);
+        let sink_handle = tokio::spawn(async move { sink.run(input_events).await.unwrap() });
+
+        // This metric is never updated again after this point. With expiration disabled it
+        // should still be present long after it would normally have expired.
+        let (name, event) = tests::create_metric_set(None, vec!["42"]);
+        tx.send(event).expect("Failed to send.");
+
+        // Wait a bit for the sink to process the event.
+        time::sleep(time::Duration::from_secs(1)).await;
+
+        let body = fetch_exporter_body().await;
+        assert!(body.contains(&name));
+
+        // Wait well past what would be the default/any reasonable flush period, sending other
+        // metrics along the way to trigger sweeps that would otherwise remove `name`.
+        for _ in 0..5 {
+            let (_, event) = tests::create_metric_gauge(None, 1.0);
+            tx.send(event).expect("Failed to send.");
+
+            time::sleep(time::Duration::from_secs(1)).await;
+        }
+
+        let body = fetch_exporter_body().await;
+        assert!(body.contains(&name));
+
+        drop(tx);
+        sink_handle.await.unwrap();
+    }
 }

website/cue/reference/components/sinks/generated/prometheus_exporter.cue

@@ -272,6 +272,10 @@ generated: components: sinks: prometheus_exporter: configuration: {
 			considered expired and is removed.
 
 			Be sure to configure this value higher than your client’s scrape interval.
+
+			Set to `0` to disable expiration entirely. Metrics will then accumulate for as long as the
+			sink runs, which can result in unbounded memory growth if metric series cardinality is
+			unbounded.
 			"""
 		required: false
 		type: uint: {