From 9f84be26efed757070dc169034c61e9023231e21 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Sat, 5 Sep 2026 22:27:59 +0100 Subject: [PATCH] Fix sum()/avg() merging sparse across staggered raw timestamps (ticket 042) merge_series_exact required an exact-millisecond timestamp match to combine sub-series, so sum(requests) by (path)-style queries over raw occurrence counts produced one point per underlying (label, timestamp) pair instead of a meaningful combined total, since real occurrence timestamps across different label groups essentially never coincide. Added a regression test confirming the sparse behavior, then replaced the merge with merge_series_windowed: a single-sub-series group is passed through unchanged (preserving existing single-label-group resolution exactly), and groups with multiple sub-series are bucketed into fixed 1s windows before summing/averaging, mirroring the fixed-width bucketing rate() already uses. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01XFUNZwCvz3kcod3352ZR2o --- src/engine.rs | 107 ++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 100 insertions(+), 7 deletions(-) diff --git a/src/engine.rs b/src/engine.rs index 71f8144..6cec839 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1176,14 +1176,30 @@ fn windowed_rate_series(points: &[(i64, f64)], range_ms: i64) -> Vec<(i64, f64)> .collect() } -/// Merges several series by exact timestamp match: values sharing a timestamp are summed (or averaged). -/// This is the plain `sum()`/`avg()` behavior — appropriate when the underlying points are occurrence -/// markers rather than a continuous reading, so there's nothing to carry forward between timestamps. -fn merge_series_exact(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { +/// Fixed window width (ms) used by `merge_series_windowed` to combine multiple raw-occurrence +/// sub-series into one time series — the same fixed-width bucketing style `histogram()`/the +/// `Timeline` query already use elsewhere in this file, just applied along the time axis instead of +/// the value axis. Occurrence timestamps across different label groups (e.g. different hosts serving +/// the same path) essentially never land on the same millisecond, so summing/averaging only requires +/// matching *this* window to combine into "total over time" rather than going sparse (ticket 042). +const MERGE_WINDOW_MS: i64 = 1_000; + +/// Merges several series into one combined time series (ticket 042). A single sub-series is returned +/// untouched — there's only one contributor, so there's nothing to merge and no reason to lose +/// resolution. With two or more sub-series, points are bucketed into fixed `MERGE_WINDOW_MS`-wide +/// windows (rather than requiring an exact-millisecond timestamp match) before summing/averaging +/// within each window, so raw occurrence markers from different label groups combine into a +/// meaningful "total over time" instead of each window reflecting only whichever sub-series happened +/// to report at that exact millisecond. +fn merge_series_windowed(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { + if sub_series.len() <= 1 { + return sub_series.first().cloned().unwrap_or_default(); + } let mut grouped: BTreeMap> = BTreeMap::new(); for points in sub_series { for (timestamp_ms, value) in points { - grouped.entry(*timestamp_ms).or_default().push(*value); + let window_start = timestamp_ms.div_euclid(MERGE_WINDOW_MS) * MERGE_WINDOW_MS; + grouped.entry(window_start).or_default().push(*value); } } grouped @@ -1198,7 +1214,7 @@ fn merge_series_exact(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, /// Merges several already-windowed rate series by sweeping the union of their timestamps, carrying each /// sub-series' last-known value forward (0.0 before its first point) so every step reflects every -/// sub-series' most recent reading — unlike `merge_series_exact`, which only combines exact-timestamp matches. +/// sub-series' most recent reading — unlike `merge_series_windowed`, which buckets by fixed time windows instead. fn merge_series_locf(sub_series: &[Vec<(i64, f64)>], is_avg: bool) -> Vec<(i64, f64)> { let mut all_timestamps: std::collections::BTreeSet = std::collections::BTreeSet::new(); for points in sub_series { @@ -1276,7 +1292,7 @@ fn evaluate_promql( let points = if rate_range_ms.is_some() { merge_series_locf(&sub_series, is_avg) } else { - merge_series_exact(&sub_series, is_avg) + merge_series_windowed(&sub_series, is_avg) }; PromqlSeriesDto { labels: group_labels, points } }) @@ -2025,6 +2041,83 @@ mod tests { assert_eq!(series[1].points, vec![(0, 1.0)]); } + fn declare_requests_by_host_metric(index: &mut BTreeMap) { + declare_metric( + index, + metric_request( + "requests", + r"request path=(?P\S+) host=(?P\S+)", + vec![ + ("path", "path", MetricPrimitiveKind::String), + ("host", "host", MetricPrimitiveKind::String), + ], + ), + ) + .unwrap(); + } + + #[test] + fn sum_by_merges_staggered_raw_timestamps_across_label_groups_instead_of_going_sparse() { + // Regression test for ticket 042: `sum(requests) by (path)` groups together several + // underlying (path, host) series. Their raw occurrence timestamps are staggered (as real + // occurrence timestamps essentially always are), so a merge that only combines exact- + // timestamp matches (the old `merge_series_exact` behavior) would leave each point + // reflecting just one host instead of the combined total across hosts. + let mut index = BTreeMap::new(); + declare_requests_by_host_metric(&mut index); + ingest_log(&mut index, 0, "request path=/a host=web1"); + ingest_log(&mut index, 1, "request path=/a host=web2"); + ingest_log(&mut index, 2, "request path=/a host=web3"); + + let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + + assert_eq!(series.len(), 1); + assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); + // The three staggered per-host occurrences merge into one combined-total point rather than + // three separate points each showing a lone host's count of 1. + let total: f64 = series[0].points.iter().map(|(_, v)| v).sum(); + assert_eq!(total, 3.0, "expected all three staggered occurrences to sum to 3.0, got {:?}", series[0].points); + assert!( + series[0].points.len() < 3, + "expected staggered timestamps to be bucketed together rather than staying sparse/near-empty, got {:?}", + series[0].points + ); + } + + #[test] + fn avg_by_merges_staggered_raw_timestamps_across_label_groups() { + let mut index = BTreeMap::new(); + declare_requests_by_host_metric(&mut index); + ingest_log(&mut index, 0, "request path=/a host=web1"); + ingest_log(&mut index, 1, "request path=/a host=web2"); + + let parsed = promql::parse_query("avg(requests) by (path)").unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + + assert_eq!(series.len(), 1); + assert_eq!(series[0].labels.get("path").map(String::as_str), Some("/a")); + // Both occurrences (value 1.0 each) land in the same merged bucket, so the average stays 1.0 + // rather than reporting two separate points that each only saw one host. + assert_eq!(series[0].points, vec![(0, 1.0)]); + } + + #[test] + fn sum_by_still_gives_a_single_label_group_its_own_line_at_full_resolution() { + // Existing single-label-group behavior (only one underlying series feeds each output group) + // must stay untouched by the multi-series windowing fix. + let mut index = BTreeMap::new(); + declare_requests_metric(&mut index); + ingest_log(&mut index, 0, "request path=/a"); + ingest_log(&mut index, 1, "request path=/a"); + + let parsed = promql::parse_query("sum(requests) by (path)").unwrap(); + let series = evaluate_promql(&index, &parsed, None, None); + + assert_eq!(series.len(), 1); + assert_eq!(series[0].points, vec![(0, 1.0), (1, 1.0)]); + } + #[test] fn bare_rate_produces_a_windowed_point_per_timestamp_not_a_single_point() { let mut index = BTreeMap::new();