feat(new transform): Add new incremental_to_absolute transform and change to MetricSet to an LRU cache with optional capacity policies

Cargo.toml

@@ -375,7 +375,7 @@ itertools.workspace = true
 k8s-openapi = { version = "0.22.0", default-features = false, features = ["v1_26"], optional = true }
 kube = { version = "0.93.0", default-features = false, features = ["client", "openssl-tls", "runtime"], optional = true }
 listenfd = { version = "1.0.2", default-features = false, optional = true }
-lru = { version = "0.16.0", default-features = false, optional = true }
+lru = { version = "0.16.0", default-features = false }
 maxminddb = { version = "0.26.0", default-features = false, optional = true, features = ["simdutf8"] }
 md-5 = { version = "0.10", default-features = false, optional = true }
 mongodb = { version = "2.8.2", default-features = false, features = ["tokio-runtime"], optional = true }
@@ -711,6 +711,7 @@ transforms-logs = [
 transforms-metrics = [
   "transforms-aggregate",
   "transforms-filter",
+  "transforms-incremental_to_absolute",
   "transforms-log_to_metric",
   "transforms-lua",
   "transforms-metric_to_log",
@@ -723,6 +724,7 @@ transforms-aggregate = []
 transforms-aws_ec2_metadata = ["dep:arc-swap"]
 transforms-dedupe = ["transforms-impl-dedupe"]
 transforms-filter = []
+transforms-incremental_to_absolute = []
 transforms-window = []
 transforms-log_to_metric = []
 transforms-lua = ["dep:mlua", "vector-lib/lua"]
@@ -737,7 +739,7 @@ transforms-throttle = ["dep:governor"]
 
 # Implementations of transforms
 transforms-impl-sample = []
-transforms-impl-dedupe = ["dep:lru"]
+transforms-impl-dedupe = []
 transforms-impl-reduce = []
 
 # Sinks
@@ -851,7 +853,7 @@ sinks-opentelemetry = ["sinks-http"]
 sinks-papertrail = ["dep:syslog"]
 sinks-prometheus = ["dep:base64", "dep:prost", "vector-lib/prometheus"]
 sinks-postgres = ["dep:sqlx"]
-sinks-pulsar = ["dep:apache-avro", "dep:pulsar", "dep:lru"]
+sinks-pulsar = ["dep:apache-avro", "dep:pulsar"]
 sinks-redis = ["dep:redis"]
 sinks-sematext = ["sinks-elasticsearch", "sinks-influxdb"]
 sinks-socket = ["sinks-utils-udp"]

changelog.d/incremental_to_absolute_transform.feature.md

@@ -0,0 +1,6 @@
+Add a new `incremental_to_absolute` transform which converts incremental metrics to absolute metrics. This is useful for
+use cases when sending metrics to a sink is lossy or you want to get a historical record of metrics, in which case
+incremental metrics may be inaccurate since any gaps in metrics sent will result in an inaccurate reading of the ending
+value.
+
+authors: GreyLilac09

lib/vector-core/src/event/metric/data.rs

@@ -58,11 +58,14 @@ impl MetricData {
 
     /// Consumes this metric, returning it as an absolute metric.
     ///
-    /// If the metric was already absolute, nothing is changed.
+    /// The `interval_ms` is set to `None`. If the metric was already absolute, nothing else is changed.
     #[must_use]
     pub fn into_absolute(self) -> Self {
         Self {
-            time: self.time,
+            time: MetricTime {
+                timestamp: self.time.timestamp,
+                interval_ms: None,
+            },
             kind: MetricKind::Absolute,
             value: self.value,
         }

src/sinks/util/buffer/metrics/mod.rs

@@ -57,7 +57,12 @@ impl Batch for MetricsBuffer {
         } else {
             let max_events = self.max_events;
             self.metrics
-                .get_or_insert_with(|| MetricSet::with_capacity(max_events))
+                .get_or_insert_with(|| {
+                    MetricSet::new(MetricSetSettings {
+                        max_events: Some(max_events),
+                        ..Default::default()
+                    })
+                })
                 .insert_update(item);
             PushResult::Ok(self.num_items() >= self.max_events)
         }