Repository navigation
feat(aggregate transform): Add set_interval_ms option - #26597
gremlinops wants to merge 7 commits into
Conversation
When enabled, each incremental metric emitted by a flush gets interval_ms set to the configured flush interval, so downstream sinks such as datadog_metrics can treat aggregated counters as rates. Covers the system-time, event-time, and final flushes. Absolute and passthrough metrics are left unchanged. Defaults to false.
github-actions
Bot
added
docs review on hold
|
All contributors have signed the CLA ✍️ ✅ |
|
I have read the CLA Document and I hereby sign the CLA |
github-actions
Bot
added
the
docs review on hold
pront
left a comment
pront
left a comment
There was a problem hiding this comment.
Hi @gremlinops, thank you for making this PR. One question, did you confirm that this PR fixes #23762?
|
|
||
| /// Overwrites `interval_ms` on flushed incremental metrics with the flush interval, so a | ||
| /// counter aggregated over the window can be read as a rate over that window. | ||
| fn set_output_interval(&self, flushed: &mut [Event]) { |
There was a problem hiding this comment.
Both encoder.rs::series_to_proto_message and encoder_v3.rs::encode_metric_to_v3 calculate interval_ms / 1000 using integer division. Can we also handle fractional-second intervals in the Datadog rate path?
There was a problem hiding this comment.
Yeah... With interval_ms < 1000, interval_ms / 1000 is 0, so value / 0.0 sends inf (or NaN for a zero counter). With non-whole seconds such as 1500, it divides by 1 and overstates the rate by 1.5x.
Proposal for both series_to_proto_message and encode_metric_to_v3:
- Scale the value with the exact interval:
value / (interval_ms as f64 / 1000.0). - The wire
intervalfield is whole seconds in both protocols (int64 intervalin v2,uint64 intervalsin v3), so sendmax(1, round(interval_ms / 1000))there instead of truncating to 0.
Tradeoff: for a non-whole-second interval the point value is exact, but Datadog's as_count() (rate × interval) uses the rounded field, so a 500 ms window would read back as 2x its true count. The alternative is to keep the field and the scaling consistent (both rounded), which makes as_count() consistent but the rate inexact. I prefer the exact rate, because rates are what the type means, but I can fix to do either.
|
Apologies, I should have dug more. No, I didn't confirm the fix for #23762, I updated it to a "related". |
Both series encoders converted interval_ms to seconds with integer division. A counter with an interval under one second was divided by zero and sent as an infinite or NaN rate, and intervals that are not whole seconds overstated the rate. Scale the rate value by the exact interval in seconds, and send the whole-second interval field rounded to the nearest second, at least 1.
|
@codex review |
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 62d0f9179d
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| // so we need to get back to seconds from our milliseconds-based interval, and then | ||
| // divide our value by that amount as well. | ||
| let value = *value / (interval as f64); | ||
| let value = *value / interval.exact_secs; |
There was a problem hiding this comment.
When sub-second interval counters reach the Datadog sink, the default 2-second batch can contain multiple windows whose timestamps fall in the same whole second. sort_and_collapse_counters_by_series_and_timestamp sums those values but leaves the surviving metric's interval_ms unchanged, so two 500 ms windows become one summed value that this line still divides by 0.5, emitting a 2x rate in both V2 and V3. Make the collapse rate-aware by preserving the total covered duration (or avoid collapsing these interval-bearing windows) before applying the exact-seconds scaling.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 9a4c15a. When both collapsed counters carry an interval and a timestamp, the collapse now uses MetricData::update, which sums the values and merges the windows into one span. Without timestamps it sums the intervals. Tests cover both orders and the no-timestamp case.
| for event in flushed { | ||
| let metric = event.as_mut_metric(); | ||
| if metric.kind() == MetricKind::Incremental { | ||
| metric.data_mut().time.interval_ms = Some(interval_ms); |
There was a problem hiding this comment.
When event_time and set_interval_ms are both enabled, bucket aggregation retains an input sample timestamp while this assignment only adds the full bucket width. For a 10-second bucket [0, 10) whose last sample is at second 5, the Datadog encoder forwards a rate point at 5 with interval 10, which its intake semantics interpret as the window [-5, 5], misplacing the aggregate and overlapping adjacent buckets. Stamp the appropriate bucket boundary when attaching the interval so the timestamp and duration describe the same window.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Valid, fixed in 9a4c15a, but the reason differs: the Datadog Agent stamps count and rate series with the bucket start, and Vector's MetricData::update also treats the timestamp as the window start. So with event_time and set_interval_ms, incremental bucket metrics now get the bucket start as their timestamp. Behavior without the flag and for absolute metrics is unchanged. System-time mode keeps its current timestamps, since the exact window start is not known there
Combine the intervals of counters that the Datadog sink collapses because they share a series and second. Before, two 500 ms windows became one summed value that was still divided by 0.5, which doubled the rate. With event_time and set_interval_ms, set the timestamp of incremental bucket metrics to the bucket start, so the timestamp and the interval describe the same window. Regenerate the aggregate example config for the new option.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 9a4c15a4bf
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| for event in flushed { | ||
| let metric = event.as_mut_metric(); | ||
| if metric.kind() == MetricKind::Incremental { | ||
| metric.data_mut().time.timestamp = Some(bucket_start); |
There was a problem hiding this comment.
When event_time and set_interval_ms are enabled, this assignment stamps the rate at the bucket start, but the checked Datadog E2E intake model in tests/e2e/datadog/metrics/series.rs:31-35 interprets a rate as (timestamp - interval, timestamp). A bucket [0, 10) is therefore submitted as [-10, 0] rather than [0, 10]. Fresh evidence since the earlier comment is that the current fix now explicitly writes bucket_start; stamp bucket_key + interval_ms instead.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
I don't think this one is right. get_time_bucket in tests/e2e/datadog/metrics/series.rs only builds the key the e2e test uses to compare Agent and Vector payloads; both sides go through it, and the file notes their timestamps are not guaranteed to line up. The Datadog Agent itself stamps rate points with the bucket start: calculateBucketStart and Append(bucketTimestamp) in pkg/aggregator/time_sampler.go, then Points: {Ts: timestamp, Value: value / interval} in pkg/metrics/counter.go. Vector's datadog_agent source also keeps that timestamp as-is. So I kept the bucket start.
There was a problem hiding this comment.
Moot after caa499c: the bucket-start stamping was reverted along with the other timestamp changes.
| for event in flushed { | ||
| let metric = event.as_mut_metric(); | ||
| if metric.kind() == MetricKind::Incremental { | ||
| metric.data_mut().time.interval_ms = Some(interval_ms); |
There was a problem hiding this comment.
When set_interval_ms is used in the default system-time mode and inputs carry timestamps, record_sum retains the latest sample timestamp while this code attaches the full flush interval without replacing that timestamp. The Datadog E2E model in tests/e2e/datadog/metrics/series.rs:31-35 treats a rate timestamp as the interval end, so a 10-second flush window [0, 10) whose last sample arrived at second 5 is reported for [-5, 5]; capture and stamp the flush boundary on these outputs.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in ac4ea38. Opposite of what I said on the earlier thread about system-time mode. With set_interval_ms, a system-time flush now stamps its incremental metrics with the window start: the transform's creation time for the first window, then the time of the previous flush. That matches the bucket-start convention used in event-time mode and by the Datadog Agent. Without the flag, timestamps are unchanged.
| // Rates are scaled by their interval, so the collapsed counter must also cover both | ||
| // windows. `update` merges the windows from the timestamps and sums the values. | ||
| (Some(_), Some(_)) if left.timestamp().is_some() && right.timestamp().is_some() => { | ||
| if !right.data_mut().update(left.data()) { |
There was a problem hiding this comment.
When two timestamped rate counters in the same encoded second have an interval near u32::MAX and timestamps differing by even 1 ms, this new call reaches MetricData::update, whose merge at lib/vector-core/src/event/metric/data.rs:104-119 performs unchecked delta_t + interval. Since this commit explicitly accepts u32::MAX, debug builds panic and release builds wrap the interval (for exactly u32::MAX + 1, it becomes None), causing the sink to emit a wildly incorrect rate or even a count; use checked/saturating interval arithmetic or avoid this generic merge.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fixed in 497f5d8. MetricData::update now uses checked_add for the window and returns false on overflow, the same way it already handles a timestamp delta that does not fit in u32. That also covers the aggregate transform's existing record_sum path. Added a test with u32::MAX intervals
MetricData::update added the timestamp delta and an interval with unchecked u32 arithmetic. Intervals near u32::MAX panicked in debug builds and wrapped in release builds. Return false instead, as the function already does when the timestamp delta does not fit. Round the Datadog whole-second interval with integer arithmetic so it needs no float cast, and fix two pedantic lints in the new tests.
github-actions
Bot
added
the
domain: core
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 497f5d81fd
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| // Rates are scaled by their interval, so the collapsed counter must also cover both | ||
| // windows. `update` merges the windows from the timestamps and sums the values. | ||
| (Some(_), Some(_)) if left.timestamp().is_some() && right.timestamp().is_some() => { | ||
| if !right.data_mut().update(left.data()) { |
There was a problem hiding this comment.
When a batch contains at least three interval-bearing counters for one series in the same whole second and they arrive out of event-time order, this pairwise update can shrink an already-merged window. For example, consecutive windows [600,900), [0,300), and [300,600) ms are comparator-equal; the first update can produce [0,900), but the next MetricData::update replaces that end with the later operand's end and produces [0,600), so the encoder divides all three values by 0.6 seconds and overstates the rate by 50%. Fresh evidence since the earlier collapse finding is that the current MetricData::update fix is non-associative after it changes the survivor's timestamp; merge the minimum start and maximum end (with checked arithmetic) so the result is order-independent.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
The collapse change this refers to was reverted in caa499c, so this no longer applies here
| right.data_mut().time.interval_ms = | ||
| Some(right_interval.saturating_add(left_interval.get())); |
There was a problem hiding this comment.
When two untimestamped rate counters share a series and their intervals sum past u32::MAX (which is accepted for each input), this branch still sums both values but silently caps the duration. V2 and V3 then divide by that capped interval, so two back-to-back u32::MAX windows are emitted at twice their intended rate. Fresh evidence beyond the earlier timestamped-overflow finding is that this no-timestamp arm bypasses the new checked MetricData::update; use checked addition and avoid or otherwise correctly rescale an unrepresentable collapse.
Useful? React with 👍 / 👎.
With set_interval_ms in system-time mode, a flushed counter kept its latest sample timestamp, so the window it described could be off by up to one interval. Track when each window starts (transform creation, then each flush) and use that as the timestamp of flushed incremental metrics, as event-time mode already does with the bucket start.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ac4ea387ef
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
|
|
||
| // NOTE: The docs for `dedup_by` specify that if `left`/`right` are equal, then | ||
| // `left` is the element that gets removed. | ||
| match (left.interval_ms(), right.interval_ms()) { |
There was a problem hiding this comment.
When subsecond windows for one series land in different sink batches (for example with batch.max_events = 1, a small max_bytes, or a busy batch at its limit), this rate-aware merge cannot see both: run_inner performs batched_partitioned at src/sinks/datadog/metrics/sink.rs:120-130 before invoking this function at lines 137-147, and encode_timestamp at encoder.rs:765-769 truncates both points to the same second. The sink documents Datadog's last-write-wins behavior for that case at lines 132-134, so unequal 500 ms windows are sent separately and one window is lost instead of being combined into the correct per-second rate. Fresh evidence beyond the earlier intra-batch collapse report is this batching order; coalesce by encoded second before batching or retain merge state across batches.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Out of scope for this PR after caa499c: set_interval_ms now requires whole seconds, and whole-second windows from one producer never share an encoded second. Sub-second intervals from other sources hitting this across batches is pre-existing sink behavior
| for event in flushed { | ||
| let metric = event.as_mut_metric(); | ||
| if metric.kind() == MetricKind::Incremental { | ||
| metric.data_mut().time.interval_ms = Some(interval_ms); |
There was a problem hiding this comment.
When system-time mode starts with buffered input and set_interval_ms is enabled, tokio::time::interval's first tick is immediately ready; because the select! at lines 498-517 is unbiased, it can record a ready counter before choosing that tick and then flush it almost immediately. This assignment still labels that near-zero startup window with the full configured interval (and a timestamp from transform construction), so, for example, 10 counts with a 10-second configuration are encoded as 1/s even if they were collected into this window over only milliseconds. Start the timer at now + interval or avoid stamping an early first flush.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Fair point, but I'm leaving it out of this PR. The immediate first tick is existing aggregate behavior, and changing the flush schedule would affect every user of the transform, not only set_interval_ms. No counts are lost: as_count() (rate × interval) still returns the true total for that first point; only its per-second reading is low for the startup window. The timestamp part no longer applies, since window-start stamping was reverted in caa499c
Move the Datadog sink collapse, interval-merge, and window-timestamp changes to a follow-up PR, so this one only adds set_interval_ms and the exact-seconds Datadog rate encoding. Require a whole number of seconds for interval_ms when set_interval_ms is enabled, because Datadog sends the interval as whole seconds. The aggregate transform then never creates sub-second rate windows.
github-actions
Bot
removed
the
domain: core
|
I've trimmed this PR back to its scope: the opt-in To keep this PR safe without those, |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: caa499c375
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| // so we need to get back to seconds from our milliseconds-based interval, and then | ||
| // divide our value by that amount as well. | ||
| let value = *value / (interval as f64); | ||
| let value = *value / interval.exact_secs; |
There was a problem hiding this comment.
When a batch contains two interval-bearing counters for one series whose timestamps fall in the same whole second (for example, adjacent 500 ms windows), the sink's collapse sums the values but leaves one 500 ms interval, and this division emits twice the true rate in both V2 and V3. Fresh evidence since the earlier resolved thread is that this squashed target removed the interval-aware merge: src/sinks/datadog/metrics/sink.rs:230-250 still only adds the values. Preserve the total covered duration (or avoid collapsing these points) before applying the exact divisor.
Useful? React with 👍 / 👎.
| for event in flushed { | ||
| let metric = event.as_mut_metric(); | ||
| if metric.kind() == MetricKind::Incremental { | ||
| metric.data_mut().time.interval_ms = Some(interval_ms); |
There was a problem hiding this comment.
When set_interval_ms is enabled for timestamped inputs, this assignment pairs the full window width with the retained sample timestamp rather than the aggregation window start; for example, an event-time [0,10) bucket whose latest sample is at t=5 is represented as a [5,15) interval under MetricData::update's start-timestamp convention (lib/vector-core/src/event/metric/data.rs:111-119), and default system-time mode has the same shift. Fresh evidence since the earlier timestamp threads is that this squashed target removed both boundary stamps: event_time.rs:449-451 and flush_system_time now reconstruct metrics without changing the timestamp. Stamp the corresponding bucket/flush start when setting the interval.
Useful? React with 👍 / 👎.
| } else { | ||
| self.flush_system_time(output); | ||
| } | ||
| self.set_output_interval(&mut output[start..]); |
There was a problem hiding this comment.
When the input closes or a topology reload happens partway through the default system-time window, flush_final drains the partial aggregate and this call still stamps the entire configured interval. Thus a transform stopped one second after the previous flush with interval_ms = 10000 sends its count as a 10-second rate (the Datadog encoder divides by 10), underreporting that final window by 10x. Track the elapsed final-window duration or leave the interval unset on an incomplete final flush.
Useful? React with 👍 / 👎.
This branch has not been deployed
3 participants
Summary
This PR adds an opt-in
set_interval_msoption (defaultfalse) to theaggregatetransform. When the option is on, each incremental metric that a flush emits getsinterval_msset to the configured flushinterval_ms. Downstream components can then read an aggregated counter as a rate over the window. For example,datadog_metricssends counters that have an interval as Datadog rates.Today, users get this behavior with a
luaorremapstep afteraggregatethat setsinterval_msby hand. The value in that step must match the aggregate interval, and nothing checks that.Behavior when enabled:
Incrementalmetrics are changed. Absolute metrics are not changed, becauseMetricData::into_absolutealready clears the interval. Metrics that pass through unchanged are not changed either, because they are not aggregated over the window.interval_msthat already exists on input is overwritten with the flush interval.Aggregate::newrequiresinterval_msto be a whole number of seconds (a multiple of 1000) when the option is on, because Datadog sends the interval as whole seconds. So this option never creates sub-second rate windows.Aggregate::newrejectsinterval_ms > u32::MAXwhen the option is on, becauseMetricTime::interval_msisNonZeroU32. Configs that do not turn on the option get no new limit.The option is applied once, after the flush, on the slice of
outputthat the flush added. Events that are already in the caller's buffer are not changed.datadog_metricssink: fractional-second intervalsFrom review: both series encoders (
series_to_proto_messagefor v2,encode_metric_to_v3for v3) convertedinterval_msto seconds with integer division. Withinterval_ms < 1000the divisor was 0, so the rate wasinf(orNaNfor a zero counter). Intervals that are not whole seconds overstated the rate (1500 ms was divided by 1).Both encoders now share one
series_interval()helper:interval_ms as f64 / 1000.0).int64 intervalin v2,uint64 intervalsin v3), so it is sent asmax(1, round(secs))instead of being truncated to 0.Tradeoff: for a non-whole-second interval the rate is exact, but
as_count()(rate × interval) uses the rounded field. Whole-second intervals are not affected. A gauge with a sub-second interval now gets interval field 1 instead of 0, and its value does not change.Not in this PR
Review found more edge cases in rate windows. They touch shared code, so they move to a follow-up PR:
sort_and_collapse_counters_by_series_and_timestamp) adds values but keeps one interval. Two sub-second windows of one series in the same second are sent at about twice the rate.set_interval_mscannot produce this (it requires whole seconds); other sources with sub-second intervals can. Before this PR, those were sent asinf.as_count()(rate × interval) returns the true total.References
Closes: #15659
Related: #23762
Related: #24686. That draft sets the interval on counters unconditionally. This PR makes it opt-in, as @pront suggested on #15659, so existing
datadog_metricsoutput does not change unless a user turns the option on.Vector configuration
Output with
set_interval_ms: true:Output with
set_interval_ms: false(unchanged from today):How did you test this PR?
src/transforms/aggregate/tests.rs: config parsing and default, theu32limit, flag off, system-time flush and final flush, overwrite of an existing interval, passthrough and existing output buffer not changed, event-time bucketscargo test --lib --no-default-features --features transforms-aggregate transforms::aggregate: 38 passedcargo fmt --all -- --checkcargo clippy --lib --tests --no-default-features --features transforms-aggregate -- -D warningswebsite/cue/reference/components/transforms/generated/aggregate.cuewithvdev build component-docsvdev check changelog-fragmentssrc/sinks/datadog/metrics/encoder.rsandencoder_v3.rs:series_interval()from 1 ms tou32::MAX, decoded v2 payload for 500 ms and 1500 ms intervals, and a v3 payload-equality check (0.5 over 500 ms encodes the same as 1.0 over 1000 ms)cargo test --lib --no-default-features --features sinks-datadog_metrics,transforms-aggregate -- sinks::datadog::metrics transforms::aggregate: 120 passedcargo clippy --lib --tests --no-default-features --features sinks-datadog_metrics -- -D warningsmake check-generated-docs/ full-feature clippy (not run locally, because it needs the full default-feature build; CI covers these)set_interval_ms_requires_whole_seconds(1, 500, 1500, 10001 ms rejected with the option and accepted without it; 1000, 2000, 60000 ms accepted). Mutation check: with the validation removed, it failsinterval_ms: 1500withset_interval_ms: trueis rejected at startup with the new errorall_series_metric_types,smoke,default_series_version_uses_the_v3_intake): 3 passedwebsite/generated/example-configs/transforms/aggregate/advanced.yamlDoes this PR include user facing changes?
no-changeloglabel to this PR.Changelog fragments:
changelog.d/15659_aggregate_set_interval_ms.enhancement.mdchangelog.d/26597_datadog_metrics_fractional_interval.fix.md