Merge branch 'main' into srikanth/cache-flux-straddle

Assisted-by: Claude Fable 5.1

pkg/querier/bucket_cache.go

@@ -151,8 +151,10 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 			slog.Uint64("cachable_end", cachableEndMs))
 	}
 
+	stepMs := uint64(step.Milliseconds())
+
 	// Trim the result to exclude data within flux interval
-	trimmedResult := bc.trimResultToFluxBoundary(fresh, cachableEndMs)
+	trimmedResult := bc.trimResultToFluxBoundary(fresh, q, cachableEndMs, stepMs)
 	if trimmedResult == nil {
 		// Result type is not cacheable (raw or scalar)
 		return
@@ -160,7 +162,6 @@ func (bc *bucketCache) Put(ctx context.Context, orgID valuer.UUID, q qbtypes.Que
 
 	// Adjust start and end times to only cache complete intervals
 	cachableStartMs := startMs
-	stepMs := uint64(step.Milliseconds())
 
 	// If we have a step interval, adjust boundaries to only cache complete intervals
 	if stepMs > 0 {
@@ -760,13 +761,21 @@ func (bc *bucketCache) deduplicateWarnings(warnings []string) []string {
 }
 
 // trimResultToFluxBoundary trims the result to exclude data points beyond the flux boundary.
-func (bc *bucketCache) trimResultToFluxBoundary(result *qbtypes.Result, fluxBoundary uint64) *qbtypes.Result {
+func (bc *bucketCache) trimResultToFluxBoundary(result *qbtypes.Result, q qbtypes.Query, fluxBoundary uint64, stepMs uint64) *qbtypes.Result {
 	trimmedResult := &qbtypes.Result{
 		Type:     result.Type,
 		Stats:    result.Stats,
 		Warnings: result.Warnings,
 	}
 
+	// A promql value at T is the query evaluated at T, so it is complete once
+	// T is at the boundary. Every other value at T aggregates [T, T+step) and
+	// is still filling until the whole step is behind the boundary.
+	lastCachableTimestampMs := fluxBoundary
+	if _, isPromQL := q.(*promqlQuery); !isPromQL && stepMs > 0 && fluxBoundary > stepMs {
+		lastCachableTimestampMs = fluxBoundary - stepMs
+	}
+
 	switch {
 	case result.Type.IsBuiltFromTimeSeries():
 		// Trim time series data
@@ -794,7 +803,7 @@ func (bc *bucketCache) trimResultToFluxBoundary(result *qbtypes.Result, fluxBoun
 						if value.Partial {
 							continue
 						}
-						if uint64(value.Timestamp) <= fluxBoundary {
+						if uint64(value.Timestamp) <= lastCachableTimestampMs {
 							trimmedSeries.Values = append(trimmedSeries.Values, value)
 						}
 					}

pkg/querier/bucket_cache_test.go

@@ -293,6 +293,31 @@ func TestBucketCache_FluxInterval(t *testing.T) {
 	assert.Len(t, missing, 1)
 }
 
+func TestBucketCache_Put_DropsStepCrossingFluxBoundary(t *testing.T) {
+	bc := createTestBucketCache(t)
+	ctx := context.Background()
+	orgID := valuer.UUID{}
+	step := qbtypes.Step{Duration: time.Minute}
+	stepMs := uint64(time.Minute.Milliseconds())
+
+	fluxBoundary := uint64(time.Now().UnixMilli()) - uint64(defaultFluxInterval.Milliseconds())
+	// the step that contains the boundary is still filling
+	straddlingStepMs := fluxBoundary / stepMs * stepMs
+	query := &mockQuery{fingerprint: "flux-straddle", startMs: straddlingStepMs - 10*stepMs, endMs: straddlingStepMs + stepMs}
+	bc.Put(ctx, orgID, query, step, &qbtypes.Result{
+		Type:  qbtypes.RequestTypeTimeSeries,
+		Value: createTestTimeSeries("A", query.startMs, query.endMs, stepMs),
+	})
+
+	cached, missing := bc.GetMissRanges(ctx, orgID, query, step)
+	require.NotNil(t, cached)
+	for _, value := range cached.Value.(*qbtypes.TimeSeriesData).Aggregations[0].Series[0].Values {
+		assert.Less(t, uint64(value.Timestamp), straddlingStepMs)
+	}
+	require.NotEmpty(t, missing)
+	assert.LessOrEqual(t, missing[0].From, straddlingStepMs)
+}
+
 func TestBucketCache_MergeTimeSeriesResults(t *testing.T) {
 	memCache := createTestCache(t)
 	bc := NewBucketCache(instrumentationtest.New().ToProviderSettings(), memCache, cacheTTL, defaultFluxInterval)

pkg/querier/heatmap_test.go

@@ -71,7 +71,7 @@ func TestTrimResultToFluxBoundaryKeepsTheHeatmapAxis(t *testing.T) {
 		},
 	}
 
-	trimmed := cache.trimResultToFluxBoundary(result, 1710000060000)
+	trimmed := cache.trimResultToFluxBoundary(result, &mockQuery{fingerprint: "heatmap"}, 1710000060000, 60000)
 
 	tsData, ok := trimmed.Value.(*qbtypes.TimeSeriesData)
 	require.True(t, ok)

tests/fixtures/time.py

@@ -1,4 +1,5 @@
 import datetime
+import time
 from typing import Any
 
 import isodate
@@ -19,3 +20,8 @@ def parse_duration(duration: Any) -> datetime.timedelta:
     if isinstance(duration, datetime.timedelta):
         return duration
     return datetime.timedelta(seconds=duration)
+
+
+def wait_until_second_of_minute(low: int, high: int) -> None:
+    while not low <= datetime.datetime.now(tz=datetime.UTC).second <= high:
+        time.sleep(1)

tests/integration/tests/queriercommon/11_cache_flux_boundary.py

@@ -0,0 +1,98 @@
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+from http import HTTPStatus
+from uuid import uuid4
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logs import Logs
+from fixtures.querier import (
+    build_aggregation,
+    build_group_by_field,
+    build_scalar_query,
+    get_all_series,
+    index_series_by_label,
+    make_query_request,
+)
+from fixtures.time import wait_until_second_of_minute
+
+MINUTE = timedelta(minutes=1)
+
+
+def test_flux_boundary_interval_is_refreshed_with_late_data(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+    insert_logs: Callable[[list[Logs]], None],
+) -> None:
+    """
+    The minute that holds the flux boundary is still filling, so its cached
+    count must not outlive the logs that arrive after the first request.
+    """
+    # the boundary must fall after the early logs (second 4) and before the
+    # late ones (second 50)
+    wait_until_second_of_minute(12, 40)
+    now = datetime.now(tz=UTC)
+    # the cache outlives the run, so a fixed filter would serve the previous
+    # run's points back to this one
+    run = uuid4().hex[:8]
+    interval_start = (now - 5 * MINUTE).replace(second=0, microsecond=0)
+    window_start = interval_start - 10 * MINUTE
+    insert_logs(
+        [
+            Logs(
+                timestamp=window_start + minute * MINUTE + timedelta(seconds=1 + i),
+                resources={"service.name": "svc-a"},
+                attributes={"run": run},
+                body=f"svc-a {minute} {i}",
+            )
+            for minute in range(10)
+            for i in range(2)
+        ]
+        + [
+            Logs(
+                timestamp=interval_start + timedelta(seconds=2 + i),
+                resources={"service.name": "svc-a"},
+                attributes={"run": run},
+                body=f"early {i}",
+            )
+            for i in range(3)
+        ]
+    )
+    token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    query = build_scalar_query(
+        name="A",
+        signal="logs",
+        aggregations=[build_aggregation("count()")],
+        group_by=[build_group_by_field("service.name", "string", "resource")],
+        filter_expression=f"run = '{run}'",
+        step_interval=60,
+    )
+    window_start_ms = int(window_start.timestamp() * 1000)
+
+    warm = make_query_request(signoz, token, window_start_ms, int(datetime.now(tz=UTC).timestamp() * 1000), [query], no_cache=False)
+    assert warm.status_code == HTTPStatus.OK, warm.text
+
+    insert_logs(
+        [
+            Logs(
+                timestamp=interval_start + timedelta(seconds=50, milliseconds=i),
+                resources={"service.name": "svc-a"},
+                attributes={"run": run},
+                body=f"late {i}",
+            )
+            for i in range(4)
+        ]
+    )
+
+    later_ms = int(datetime.now(tz=UTC).timestamp() * 1000)
+    cached = make_query_request(signoz, token, window_start_ms, later_ms, [query], no_cache=False)
+    assert cached.status_code == HTTPStatus.OK, cached.text
+    fresh = make_query_request(signoz, token, window_start_ms, later_ms, [query], no_cache=True)
+    assert fresh.status_code == HTTPStatus.OK, fresh.text
+
+    interval_ms = int(interval_start.timestamp() * 1000)
+    fresh_points = {point["timestamp"]: point["value"] for point in index_series_by_label(get_all_series(fresh.json(), "A"), "service.name")["svc-a"]["values"]}
+    assert fresh_points[interval_ms] == 7, fresh_points
+    cached_points = {point["timestamp"]: point["value"] for point in index_series_by_label(get_all_series(cached.json(), "A"), "service.name")["svc-a"]["values"]}
+    assert cached_points[interval_ms] == fresh_points[interval_ms], f"cache served the stale count {cached_points[interval_ms]} for the interval that contains the flux boundary"