Added negative offset check for caching queries

Signed-off-by: pawarpranav83 <[email protected]>

Author: pawarpranav83

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## master / unreleased
+* [BUGFIX] Query Frontend: queries with negative offset should check whether it is cacheable or not. #5719
 * [CHANGE] Azure Storage: Upgraded objstore dependency and support Azure Workload Identity Authentication. Added `connection_string` to support authenticating via SAS token. Marked `msi_resource` config as deprecating. #5645
 * [CHANGE] Store Gateway: Add a new fastcache based inmemory index cache. #5619
 * [CHANGE] Index Cache: Multi level cache backfilling operation becomes async. Added `-blocks-storage.bucket-store.index-cache.multilevel.max-async-concurrency` and `-blocks-storage.bucket-store.index-cache.multilevel.max-async-buffer-size` configs and metric `cortex_store_multilevel_index_cache_backfill_dropped_items_total` for number of dropped items. #5661

pkg/querier/tripperware/queryrange/results_cache.go

@@ -265,12 +265,13 @@ func (s resultsCache) shouldCacheResponse(ctx context.Context, req tripperware.R
 	if !s.isAtModifierCachable(ctx, req, maxCacheTime) {
 		return false
 	}
+	if !s.isOffsetCachable(ctx, req) {
+		return false
+	}
 
 	return true
 }
 
-var errAtModifierAfterEnd = errors.New("at modifier after end")
-
 // isAtModifierCachable returns true if the @ modifier result
 // is safe to cache.
 func (s resultsCache) isAtModifierCachable(ctx context.Context, r tripperware.Request, maxCacheTime int64) bool {
@@ -280,6 +281,7 @@ func (s resultsCache) isAtModifierCachable(ctx context.Context, r tripperware.Re
 	//      below maxCacheTime. In such cases if any tenant is intentionally
 	//      playing with old data, we could cache empty result if we look
 	//      beyond query end.
+	var errAtModifierAfterEnd = errors.New("at modifier after end")
 	query := r.GetQuery()
 	if !strings.Contains(query, "@") {
 		return true
@@ -321,6 +323,46 @@ func (s resultsCache) isAtModifierCachable(ctx context.Context, r tripperware.Re
 	return atModCachable
 }
 
+// isOffsetCachable returns true if the offset is positive, result is safe to cache.
+// and false when offset is negative, result is not cached.
+func (s resultsCache) isOffsetCachable(ctx context.Context, r tripperware.Request) bool {
+	var errNegativeOffset = errors.New("negative offset")
+	query := r.GetQuery()
+	if !strings.Contains(query, "offset") {
+		return true
+	}
+	expr, err := parser.ParseExpr(query)
+	if err != nil {
+		level.Warn(util_log.WithContext(ctx, s.logger)).Log("msg", "failed to parse query, considering offset as not cachable", "query", query, "err", err)
+		return false
+	}
+
+	offsetCachable := true
+	parser.Inspect(expr, func(n parser.Node, _ []parser.Node) error {
+		switch e := n.(type) {
+		case *parser.VectorSelector:
+			if e.OriginalOffset < 0 {
+				offsetCachable = false
+				return errNegativeOffset
+			}
+		case *parser.MatrixSelector:
+			offset := e.VectorSelector.(*parser.VectorSelector).OriginalOffset
+			if offset < 0 {
+				offsetCachable = false
+				return errNegativeOffset
+			}
+		case *parser.SubqueryExpr:
+			if e.OriginalOffset < 0 {
+				offsetCachable = false
+				return errNegativeOffset
+			}
+		}
+		return nil
+	})
+
+	return offsetCachable
+}
+
 func getHeaderValuesWithName(r tripperware.Response, headerName string) (headerValues []string) {
 	for name, hv := range r.HTTPHeaders() {
 		if name != headerName {

pkg/querier/tripperware/queryrange/results_cache_test.go

@@ -416,6 +416,45 @@ func TestShouldCache(t *testing.T) {
 			input:    tripperware.Response(&PrometheusResponse{}),
 			expected: false,
 		},
+		// offset on vector selectors.
+		{
+			name:     "positive offset on vector selector",
+			request:  &PrometheusRequest{Query: "metric offset 10ms", End: 125000},
+			input:    tripperware.Response(&PrometheusResponse{}),
+			expected: true,
+		},
+		{
+			name:     "negative offset on vector selector",
+			request:  &PrometheusRequest{Query: "metric offset -10ms", End: 125000},
+			input:    tripperware.Response(&PrometheusResponse{}),
+			expected: false,
+		},
+		// offset on matrix selectors.
+		{
+			name:     "positive offset on matrix selector",
+			request:  &PrometheusRequest{Query: "rate(metric[5m] offset 10ms)", End: 125000},
+			input:    tripperware.Response(&PrometheusResponse{}),
+			expected: true,
+		},
+		{
+			name:     "negative offset on matrix selector",
+			request:  &PrometheusRequest{Query: "rate(metric[5m] offset -10ms)", End: 125000},
+			input:    tripperware.Response(&PrometheusResponse{}),
+			expected: false,
+		},
+		// offset on subqueries.
+		{
+			name:     "positive offset on subqueries",
+			request:  &PrometheusRequest{Query: "sum_over_time(rate(metric[1m])[10m:1m] offset 10ms)", End: 125000},
+			input:    tripperware.Response(&PrometheusResponse{}),
+			expected: true,
+		},
+		{
+			name:     "negative offset on subqueries",
+			request:  &PrometheusRequest{Query: "sum_over_time(rate(metric[1m])[10m:1m] offset -10ms)", End: 125000},
+			input:    tripperware.Response(&PrometheusResponse{}),
+			expected: false,
+		},
 	} {
 		{
 			t.Run(tc.name, func(t *testing.T) {