diff --git a/src/engine.rs b/src/engine.rs index ad6a124..b768573 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -1267,14 +1267,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 @@ -1289,7 +1305,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 { @@ -1398,7 +1414,7 @@ fn evaluate_promql( groups .into_iter() .map(|(group_labels, sub_series)| { - let points = merge_series_exact(&sub_series, false); + let points = merge_series_windowed(&sub_series, false); PromqlSeriesDto { labels: group_labels, points } }) .collect() @@ -1426,7 +1442,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 } }) @@ -2081,6 +2097,32 @@ mod tests { .unwrap(); } + fn declare_requests_by_host_metric(index: &mut BTreeMap) { + declare_test_param( + index, + "path", + r"request path=(?P\S+) host=(?P\S+)", + "path", + MetricPrimitiveKind::String, + ); + declare_test_param( + index, + "host", + r"request path=(?P\S+) host=(?P\S+)", + "host", + MetricPrimitiveKind::String, + ); + declare_metric( + index, + metric_request( + "requests", + r"request path=(?P\S+) host=(?P\S+)", + vec!["path", "host"], + ), + ) + .unwrap(); + } + #[test] fn sum_over_string_kind_metric_is_rejected_with_an_actionable_error() { let mut index = BTreeMap::new(); @@ -2160,6 +2202,68 @@ mod tests { assert!(series.is_ok()); } + #[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).unwrap(); + + 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).unwrap(); + + 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).unwrap(); + + 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();