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

[GreyLilac09]

Summary

Create new incremental_to_absolute transform

Useful for:

• avoiding duplicate metrics cache creation at the sink level
• creating a historical record of metrics to account for lossy connections/file-based back-filling

Problem it solves: #23018

Vector configuration

transforms:
  incremental_to_absolute:
    type: incremental_to_absolute
    cache:
      max_bytes: 100_000_00
      max_events: 1000
      time_to_idle: 300

How did you test this PR?

Example 1
Example configuration:

data_dir: ./vector-data-dir
sources:
  s0:
    type: static_metrics
    interval_secs: 1
    metrics:
      - name: response_time
        kind: incremental
        value:
          counter:
            value: 1
        tags: {}

transforms:
  t0:
    inputs:
      - s0
    type: incremental_to_absolute
    cache:
      max_events: 5
sinks:
  console:
    type: console
    inputs:
      - t0
    target: stdout
    encoding:
      codec: json
      json:
        pretty: true

Example output:

{
  "name": "response_time",
  "namespace": "static",
  "timestamp": "2025-07-16T04:02:06.446891Z",
  "kind": "absolute",
  "counter": {
    "value": 1.0
  }
}
{
  "name": "response_time",
  "namespace": "static",
  "timestamp": "2025-07-16T04:02:07.447752Z",
  "kind": "absolute",
  "counter": {
    "value": 2.0
  }
}
{
  "name": "response_time",
  "namespace": "static",
  "timestamp": "2025-07-16T04:02:08.447934Z",
  "kind": "absolute",
  "counter": {
    "value": 3.0
  }
}
{
  "name": "response_time",
  "namespace": "static",
  "timestamp": "2025-07-16T04:02:09.447506Z",
  "kind": "absolute",
  "counter": {
    "value": 4.0
  }
}

Example 2
Enforcing max_events. Note that max_events = 1 means the other item will just be evicted immediately, resulting in 1 being received always in this example

data_dir: ./vector-data-dir
sources:
  s0:
    type: static_metrics
    interval_secs: 1
    metrics:
      - name: m1
        kind: incremental
        value:
          counter:
            value: 1
        tags: {}
      - name: m2
        kind: incremental
        value:
          counter:
            value: 1
        tags: {}
transforms:
  t0:
    type: incremental_to_absolute
    inputs:
      - s0
    cache:
      max_events: 1
      time_to_idle: 10
sinks:
  console:
    type: console
    inputs:
      - t0
    target: stdout
    encoding:
      codec: json
      json:
        pretty: true

Output:

{
  "name": "m1",
  "namespace": "static",
  "timestamp": "2025-07-26T00:37:35.593501Z",
  "kind": "absolute",
  "counter": {
    "value": 1.0
  }
}
{
  "name": "m2",
  "namespace": "static",
  "timestamp": "2025-07-26T00:37:35.593512Z",
  "kind": "absolute",
  "counter": {
    "value": 1.0
  }
}
{
  "name": "m1",
  "namespace": "static",
  "timestamp": "2025-07-26T00:37:36.593826Z",
  "kind": "absolute",
  "counter": {
    "value": 1.0
  }
}
{
  "name": "m2",
  "namespace": "static",
  "timestamp": "2025-07-26T00:37:36.593830Z",
  "kind": "absolute",
  "counter": {
    "value": 1.0
  }
}

Change Type

• Bug fix
• New feature
• Non-functional (chore, refactoring, docs)
• Performance

Is this a breaking change?

• Yes
• No

Does this PR include user facing changes?

• Yes. Please add a changelog fragment based on our guidelines.
• No. A maintainer will apply the no-changelog label to this PR.

References

• Closes: Convert incremental to absolute counter values for statsd source #23018

Notes

• Please read our Vector contributor resources.
• Do not hesitate to use @vectordotdev/vector to reach out to us regarding this PR.
• Some CI checks run only after we manually approve them.
  • We recommend adding a pre-push hook, please see this template.
  • Alternatively, we recommend running the following locally before pushing to the remote branch:
    • cargo fmt --all
    • cargo clippy --workspace --all-targets -- -D warnings
    • cargo nextest run --workspace (alternatively, you can run cargo test --all)
• After a review is requested, please avoid force pushes to help us review incrementally.
  • Feel free to push as many commits as you want. They will be squashed into one before merging.
  • For example, you can run git merge origin master and git push.
• If this PR introduces changes Vector dependencies (modifies Cargo.lock), please
  run cargo vdev build licenses to regenerate the license inventory and commit the changes (if any). More details here.

[thomasqueirozb]

Hey @GreyLilac09, thanks for the PR. Please update the test plan in your description including a vector config and expected output. This part allows us to run your code with a config and easily validate what the expected output should look like.

Here is an example config you can build on top of:

sources:
  s0:
    type: static_metrics
    interval_secs: 1
    metrics:
      - name: response_time
        kind: incremental
        value:
          counter:
            value: 1
        tags: {}

transforms:
  t0:
    type: remap
    inputs:
      - s0
    source: |-
      .tags.output = "some value"

sinks:
  console:
    type: console
    inputs:
      - t0
    target: stdout
    encoding:
      codec: json
      json:
        pretty: true

or you can create one from scratch. Thanks!

[iadjivon]

All set from Docs!

[Copilot]

Pull Request Overview

This PR introduces a new incremental_to_absolute transform that converts incremental metrics to absolute metrics while preserving the cumulative values. This is useful for avoiding duplicate metric caches at sink levels and creating historical records for scenarios with lossy connections or file-based backfilling.

Key changes:

• Implements the core transform logic with TTL-based metric expiration
• Adds comprehensive documentation and configuration examples
• Includes unit tests covering incremental-to-absolute conversion and pass-through behavior for already-absolute metrics

Reviewed Changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 3 comments.

Show a summary per file

File	Description
src/transforms/incremental_to_absolute.rs	Core implementation of the transform with metric conversion logic and tests
website/cue/reference/components/transforms/incremental_to_absolute.cue	Documentation and configuration reference for the new transform
lib/vector-core/src/event/metric/data.rs	Updates into_absolute() method to remove interval_ms when converting metrics
src/transforms/mod.rs	Registers the new transform module
website/cue/reference/components.cue	Adds feature definition for the transform
Cargo.toml	Adds feature flags for the new transform
changelog.d/incremental_to_absolute_transform.feature.md	Changelog entry documenting the new feature

Comments suppressed due to low confidence (2)

website/cue/reference/components/transforms/incremental_to_absolute.cue:50

• The example title 'Aggregate over 5 seconds' is misleading. This transform converts incremental to absolute metrics but doesn't aggregate over time windows. A more accurate title would be 'Convert incremental counter to absolute'.

			title: "Aggregate over 5 seconds"

[GreyLilac09]

After thinking about this some more, I'm not so sure if just the expire_metrics_secs would be the right approach. The problem is that a lot of incremental counters are sparse (eg they could be incremented as a statsd counter every few hours or days), and the current approach just miss all of them. I think a better approach would be to implement the MetricSet (see https://github.com/vectordotdev/vector/blob/master/src/sinks/util/buffer/metrics/normalize.rs) as an LRU cache rather than as a IndexMap, and have a configurable max size.

1. I initially was thinking about this approach, but shied away from it because it would involve a much more substantial change to the code than I felt making at the time. However, I do think it's the right way to go about this.
2. The problem with just an LRU cache is someone might have a scenario where they need to be able to handle extremely high bursts of data without dropping. In this scenario, the expire_metrics_secs would be useful.
   a. in this scenario, the cache would just eventually grow to the max size and stay at that size until restart. If the LRU cache max size is say 256 MB, one month later Vector could just sit at 256 MB allocated memory for Vector, even if it hasn't received any of those incremental counters for 27 days
   b. they might not be able to predict the size of the burst ahead of time, and they need to be flexible, so having a fixed max size would cause a lot of inaccurate data when that burst comes and it exceeds the allowed size

Thus, I would propose an additional configuration, eg

transforms:
  incremental_to_absolute:
    type: incremental_to_absolute
    cache_max_size: 268435488 (default)
    expire_metrics_secs: 120s (default)

where cache_max_size is in bytes

Alternatively, maybe it would just be better to group the cache configs like we do for buffer and batch? So the config would be like

transforms:
  incremental_to_absolute:
    type: incremental_to_absolute
    cache:
      max_size: 268435488
      timeout_secs: 120s

Curious to hear your thoughts. I'd also eventually like to add this config to the prom remote write sink (eg from this PR).

I can also do add the LRU cache in a separate PR, but it would not be ideal to change the config later (eg. go from expire_metrics_secs to cache.timeout_secs) if it can be avoided, so if we go with the second config we'd probably want to do it in this PR.

[pront]

he problem is that a lot of incremental counters are sparse (eg they could be incremented as a statsd counter every few hours or days), and the current approach just miss all of them.

Can you explain with an example?

From a UX perspective the following is better:

transforms:
  incremental_to_absolute:
    type: incremental_to_absolute
    cache:
      max_size: 268435488
      timeout_secs: 120s

[GreyLilac09]

For example, if we increment (+1) count every 10 minutes, and the expire_metrics_secs is 5 minutes, that count would always just show up as 1 (unchanging) in prometheus and the increase in value is never logged

[GreyLilac09]

transforms:
  incremental_to_absolute:
    type: incremental_to_absolute
    cache:
      max_size: 268435488
      timeout_secs: 120s

@pront that makes sense, if this is the case we should also change the config of prom remote write sink (#23286) to be the same. I think the plan would be to modify this PR to use the LRU cache with this config, and in a separate follow-up PR modify the prom remote write config to have the same?

[pront]

transforms:
  incremental_to_absolute:
    type: incremental_to_absolute
    cache:
      max_size: 268435488
      timeout_secs: 120s

@pront that makes sense, if this is the case we should also change the config of prom remote write sink (#23286) to be the same. I think the plan would be to modify this PR to use the LRU cache with this config, and in a separate follow-up PR modify the prom remote write config to have the same?

Hi @GreyLilac09, this makes sense to me!

[GreyLilac09]

@pront I would appreciate an initial look at the new structure here. There's still some details I'm working to iron out (specifically, time_to_idle doesn't seem to be working) but I'd love to know if the structure is directionally correct.

[GreyLilac09]

not sure if this will just work, I'm looking to emit a metric for cache events and size

[GreyLilac09]

Copying same structure as BatchConfig

[GreyLilac09]

FYI, after some debugging, it looks like the TTI and max_events/max_bytes works, but they're enforced in the background so the max_events/max_bytes can exceed it temporarily

The alternative unsync::Cache (which enforces capacity/max events) is not thread-safe so I don't think we can use it

Our new approach can support it (no extra dependencies)

[GreyLilac09]

@pront I think it's ready for review. I originally went for a mini-moka approach, but I think 1. I'm not able to build it with cross due to new dependencies 2. I think we get more control and flexibility on behavior (eg being able to specify both max_events and max_bytes, rather than either) by just using the original cache impl and manually tracking/evicting from the LRU cache

[GreyLilac09]

@pront @thomasqueirozb gentle bump on a review--we'd like to get this in before next week's release if possible