Throttle metrics

[ilinas]

Question

I am not sure if "throttle" is the right word...

I have metrics being scraped at 15 seconds interval, but I only want to send metrics to the sink at 45 second granularity. In other words, I only want to keep every third scrape.

I cannot change scrape_interval_secs, because I am receiving data from another Vector instance that I do not control.

Vector Config

Vector that I do not control:

sources:
  scrape_metrics:
    type: prometheus_scrape
    endpoints:
      - http://localhost/metrics
    scrape_interval_secs: 15

sinks:
  send_metrics:
    type: vector
    inputs:
      - scrape_metrics
    address: localhost:6000

Vector that I control:

sources:
  receive_metrics:
    type: vector
    address: localhost:6000

transforms:
  throttle_metrics:
    type: ???
    inputs:
      - receive_metrics
    scrape_interval_change_to_secs: 45

sinks:
  write_metrics:
    type: prometheus_remote_write
    inputs:
      - throttle_metrics
    endpoint: http://localhost/insert/0/prometheus/

Vector Logs

No response

[thomasqueirozb]

Hi @ilinas. What you want is basically a synchronous state between events. This is not something we currently support, but there are workarounds.

You can invoke external processes/access machine state such as the filesystem using the lua transform. You'd basically have to write a program that returns true, false, false depending on how many times is has been invoked. This is very prone to a race condition but since your metrics are only supposed to come in every 15 seconds it may work.

I spent some time and I come up with a pure Vector configuration without a lua transform, which leverages enrichment tables to keep internal state between events. Keep in mind that this is also very prone to a race condition - namely if an event comes in less than 3 seconds after another one.

sources:
  receive_metrics:
    type: vector
    address: localhost:6000

transforms:
  # Memory table only accepts log events
  to_log:
    type: metric_to_log
    inputs: [receive_metrics]

  one_per_scrape:
    type: dedupe
    inputs: [to_log]
    fields:
      match: [timestamp]   # first event per scrape timestamp survives

  update_counter:
    type: remap
    inputs: [one_per_scrape]
    source: |
      # ?? {} handles a missing record (first get / after a state reset)
      rec = get_enrichment_table_record("scrape_counter", {"key": "count"}) ?? {}
      current = to_int(rec.value) ?? 0
      . = { "count": current + 1 }

  # This runs concurrently with the counter write. Delay ensures that the
  # count gets written before we try to read it in the filter
  hold:
    type: delay
    inputs: [receive_metrics]
    delay_ms: 3000

  keep_every_third:
    type: filter
    inputs: [hold]
    condition:
      type: vrl
      source: |
        rec = get_enrichment_table_record("scrape_counter", {"key": "count"}) ?? {}
        mod(to_int(rec.value) ?? 0, 3) == 0

enrichment_tables:
  scrape_counter:
    type: memory
    inputs: [update_counter]
    ttl: 9223372036854775807 # ~292 billion years. We don't want the counter to expire.

sinks:
  write_metrics:
    type: prometheus_remote_write
    inputs: [keep_every_third]
    endpoint: http://localhost/insert/0/prometheus/

There is also the sample transform, but that one will drop events randomly, not in a predictable manner like you'd like. It doesn't accept metrics so you'd also need a log_to_metric and a metric_to_log transform.

[thomasqueirozb]

Update:

Did some more digging. Looks like that lua transform runs sequentially and is able to retain state between runs. Here is a non-racy configuration that should work for your use case:

  sources:
    receive_metrics:
      type: vector
      address: 127.0.0.1:6000

  transforms:
    keep_every_third_scrape:
      type: lua
      version: "2"
      inputs: [receive_metrics]
      hooks:
        process: |
          function (event, emit)
            -- Scrape boundary: one scrape carries N metric events sharing a
            -- timestamp. Increment only when the timestamp changes.
            -- The key truncates sub-seconds: a real prometheus scrape stamps
            -- the whole batch with one timestamp, but rounding makes the
            -- boundary robust even if a source stamps events individually
            -- within the same second.
            local ts = event.metric and event.metric.timestamp

            -- Events without a timestamp can't be attributed to a scrape. Drop.
            if ts == nil then
              return
            end

            ts = string.format(
              "%04d-%02d-%02dT%02d:%02d:%02dZ",
              ts.year, ts.month, ts.day, ts.hour, ts.min, ts.sec
            )

            -- last_scrape_ts survives between runs
            if ts ~= last_scrape_ts then
              scrape_count = scrape_count + 1
              last_scrape_ts = ts
            end

            if scrape_count % 3 == 0 then
              emit(event)
            end
            -- else: drop. No emit = dropped.
          end
        init: |
          function ()
            scrape_count = 0
            last_scrape_ts = nil
          end

  sinks:
    write_metrics:
      type: prometheus_remote_write
      inputs: [keep_every_third_scrape]
      endpoint: http://localhost/insert/0/prometheus/

[ilinas]

Thanks @thomasqueirozb. Although I was hoping for a more generic solution. I was thinking, if this was to be implemented as a proper transform, what would be the best approach?

One possibility I am considering is to extend the reduce transform to support metrics. It already has parameters like expire_after_ms, flush_period_ms, and max_events that would allow handling arbitrary streams of metrics instead of relying on them being scraped at regular intervals. And merge_strategies would provide more useful ways to aggregate the metrics instead of discarding every n-th occurrence.

[thomasqueirozb]

implemented as a proper transform, what would be the best approach

Yes absolutely, this is not something we support today (at least not easily).

This seems like it could be a configuration option in the sample transform. The only thing today that makes it not work out is the statelessness of it. You'd encounter the same issue in the reduce transform since you have no way of grouping your events.

However, I think that the best way to solve your specific use case is to add atomic operation support to (memory) enrichment tables. That way it'd be possible to use a remap transform to tag every incoming event you get with a counter which is stored in an enrichment table, and is incremented every time a new event is processed. With an event counter you'd be able to drop events (inside remap) or use a filter transform to do so.

[ilinas]

Maybe my example was a bit too specific... In reality I cannot really known at what interval the metrics are being scraped. I cannot even assume that they have exactly the same timestamp, because in the end it's just a stream of events and could come from multiple sources.

What I really would like to be able to do, is to allow each unique metric, as defined by the labels, only once every n-th second regardless of how often it is scraped.

E.g. these are 2 unique metrics because of the unique host label:

vector_build_info{arch="x86_64", debug="false", host="abc-de-001", rust_version="1.92", version="0.56.0"} 1
vector_build_info{arch="x86_64", debug="false", host="abc-de-002", rust_version="1.92", version="0.56.0"} 1

And this is the same metric scraped multiple times:

vector_buffer_events{buffer_type="disk", component_kind="sink", host="abc-de-001"} 15
vector_buffer_events{buffer_type="disk", component_kind="sink", host="abc-de-001"} 23
vector_buffer_events{buffer_type="disk", component_kind="sink", host="abc-de-001"} 37

So if I receive all of these events within 45 seconds, I want to emit the following:

vector_build_info{arch="x86_64", debug="false", host="abc-de-001", rust_version="1.92", version="0.56.0"} 1
vector_build_info{arch="x86_64", debug="false", host="abc-de-002", rust_version="1.92", version="0.56.0"} 1
vector_buffer_events{buffer_type="disk", component_kind="sink", host="abc-de-001"} 37

Maybe in reality it's more like a dedupe?

[thomasqueirozb]

Maybe the dedupe transform would fit better.

There are a couple of caveats though:

1. It doesn't support metrics, so you'd need a metric_to_log and log_to_metric transform to use it today
2. You'd likely need to set fields.ignore so that all fields except actual metric data match (e.g. counter.value, gauge.value). See the metric data model for reference.