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_live: 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]

@thomasqueirozb can you approve please, would love for this to be in the upcoming release

Unfortunately, we are already outside the merge window. You can always use the nightly builds though which should be available one day after this PR is merged.

[GreyLilac09]

@pront @thomasqueirozb Could you guys approve please? It's been ready for a while now

[thomasqueirozb]

Sorry for the delay!

(GitHub just completely deleted my review. Actual review is in the next comment)

[GreyLilac09]

@thomasqueirozb I think you need to approve?

[pront]

@thomasqueirozb I think you need to approve?

It's because All comments must be resolved., did we leave some threads open?

[GreyLilac09]

just resolved the last one

[pront]

Thank you for this contribution @GreyLilac09!