fix bugs in logs results caching and its tests

CHANGELOG.md

@@ -31,6 +31,7 @@
 * [7906](https://github.com/grafana/loki/pull/7906) **kavirajk**: Add API endpoint that formats LogQL expressions and support new `fmt` subcommand in `logcli` to format LogQL query.
 * [7966](https://github.com/grafana/loki/pull/7966) **sandeepsukhani**: Fix query-frontend request load balancing when using k8s service.
 * [7988](https://github.com/grafana/loki/pull/7988) **ashwanthgoli** store: write overlapping chunks to multiple stores.
+* [7925](https://github.com/grafana/loki/pull/7925) **sandeepsukhani**: Fix bugs in logs results caching causing query-frontend to return logs outside of query window.
 
 ##### Changes
 

pkg/querier/queryrange/log_result_cache.go

@@ -180,8 +180,7 @@ func (l *logResultCache) handleHit(ctx context.Context, cacheKey string, cachedR
 	result := emptyResponse(cachedRequest)
 	// if the request is the same and cover the whole time range,
 	// we can just return the cached result.
-	if !lokiReq.GetStartTs().After(cachedRequest.GetStartTs()) && lokiReq.GetStartTs().Equal(cachedRequest.GetStartTs()) &&
-		!lokiReq.GetEndTs().Before(cachedRequest.GetEndTs()) && lokiReq.GetEndTs().Equal(cachedRequest.GetEndTs()) {
+	if cachedRequest.StartTs.UnixNano() <= lokiReq.StartTs.UnixNano() && cachedRequest.EndTs.UnixNano() >= lokiReq.EndTs.UnixNano() {
 		return result, nil
 	}
 	// we could be missing data at the start and the end.
@@ -240,7 +239,7 @@ func (l *logResultCache) handleHit(ctx context.Context, cacheKey string, cachedR
 			if startResp.Status != loghttp.QueryStatusSuccess {
 				return startResp, nil
 			}
-			result = mergeLokiResponse(startResp, result)
+			result = mergeLokiResponse(extractLokiResponse(lokiReq.GetStartTs(), lokiReq.GetEndTs(), startResp), result)
 		}
 	}
 
@@ -254,7 +253,7 @@ func (l *logResultCache) handleHit(ctx context.Context, cacheKey string, cachedR
 			if endResp.Status != loghttp.QueryStatusSuccess {
 				return endResp, nil
 			}
-			result = mergeLokiResponse(endResp, result)
+			result = mergeLokiResponse(extractLokiResponse(lokiReq.GetStartTs(), lokiReq.GetEndTs(), endResp), result)
 		}
 	}
 
@@ -274,6 +273,45 @@ func (l *logResultCache) handleHit(ctx context.Context, cacheKey string, cachedR
 	return result, nil
 }
 
+// extractLokiResponse extracts response with interval [start, end)
+func extractLokiResponse(start, end time.Time, r *LokiResponse) *LokiResponse {
+	extractedResp := LokiResponse{
+		Status:     r.Status,
+		Direction:  r.Direction,
+		Limit:      r.Limit,
+		Version:    r.Version,
+		ErrorType:  r.ErrorType,
+		Error:      r.Error,
+		Statistics: r.Statistics,
+		Data: LokiData{
+			ResultType: r.Data.ResultType,
+			Result:     []logproto.Stream{},
+		},
+	}
+	for _, stream := range r.Data.Result {
+		if stream.Entries[0].Timestamp.After(end) || stream.Entries[len(stream.Entries)-1].Timestamp.Before(start) {
+			continue
+		}
+
+		extractedStream := logproto.Stream{
+			Labels:  stream.Labels,
+			Entries: []logproto.Entry{},
+			Hash:    stream.Hash,
+		}
+		for _, entry := range stream.Entries {
+			if entry.Timestamp.Before(start) || entry.Timestamp.After(end) || entry.Timestamp.Equal(end) {
+				continue
+			}
+
+			extractedStream.Entries = append(extractedStream.Entries, entry)
+		}
+
+		extractedResp.Data.Result = append(extractedResp.Data.Result, extractedStream)
+	}
+
+	return &extractedResp
+}
+
 func isEmpty(lokiRes *LokiResponse) bool {
 	return lokiRes.Status == loghttp.QueryStatusSuccess && len(lokiRes.Data.Result) == 0
 }