Reuse grpc buffer in querier's store-gateway stream (#7519)

* reuse grpc buffer in querier's store-gate
way stream

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>

* lint

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>

* Scope down pr to just detach

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>

* update changelog

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>

* Disable intern

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>

* Use series copy

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>

* Copy labels

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>

---------

Signed-off-by: Essam Eldaly <eeldaly@amazon.com>

Author: eeldaly

CHANGELOG.md

@@ -27,6 +27,7 @@
 * [ENHANCEMENT] Distributor: Add HMAC-SHA256 stream authentication for `PushStream` via `-distributor.sign-write-requests-keys`. #7475
 * [ENHANCEMENT] Instrument Ingester CPU profile with source for read APIs. #7494
 * [ENHANCEMENT] Ingester: Convert expanded postings cache from FIFO to LRU eviction to retain frequently-queried entries under memory pressure. #7510
+* [ENHANCEMENT] Querier: Detach series label and chunk data from gRPC unmarshal buffers in store-gateway streaming path, allowing the Go GC to reclaim receive buffers. #7519
 * [BUGFIX] Querier: Fix queryWithRetry and labelsWithRetry returning (nil, nil) on cancelled context by propagating ctx.Err(). #7370
 * [BUGFIX] Metrics Helper: Fix non-deterministic bucket order in merged histograms by sorting buckets after map iteration, matching Prometheus client library behavior. #7380
 * [BUGFIX] Distributor: Return HTTP 401 Unauthorized when tenant ID resolution fails in the Prometheus Remote Write 2.0 path. #7389

pkg/querier/blocks_store_queryable.go

@@ -29,6 +29,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/pool"
 	thanosquery "github.com/thanos-io/thanos/pkg/query"
 	"github.com/thanos-io/thanos/pkg/store/hintspb"
+	"github.com/thanos-io/thanos/pkg/store/labelpb"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"github.com/thanos-io/thanos/pkg/strutil"
 	"go.uber.org/atomic"
@@ -675,7 +676,11 @@ func (q *blocksStoreQuerier) fetchSeriesFromStores(
 			myQueriedBlocks := []ulid.ULID(nil)
 
 			processSeries := func(s *storepb.Series) error {
-				mySeries = append(mySeries, s)
+				// Detach series data from the gRPC unmarshal buffer so that it can be freed.
+				sCopy := *s
+				sCopy.Labels = append([]labelpb.ZLabel(nil), s.Labels...)
+				detachSeriesFromBuffer(&sCopy)
+				mySeries = append(mySeries, &sCopy)
 
 				// Add series fingerprint to query limiter; will return error if we are over the limit
 				limitErr := queryLimiter.AddSeries(cortexpb.FromLabelsToLabelAdapters(s.PromLabels()))
@@ -1189,6 +1194,17 @@ func convertBlockHintsToULIDs(hints []hintspb.Block) ([]ulid.ULID, error) {
 	return res, nil
 }
 
+// detachSeriesFromBuffer re-allocates label strings and chunk data byte slices
+// so that the series no longer references the gRPC unmarshal buffer.
+func detachSeriesFromBuffer(s *storepb.Series) {
+	labelpb.ReAllocZLabelsStrings(&s.Labels, false)
+	for i := range s.Chunks {
+		if s.Chunks[i].Raw != nil && len(s.Chunks[i].Raw.Data) > 0 {
+			s.Chunks[i].Raw.Data = append([]byte(nil), s.Chunks[i].Raw.Data...)
+		}
+	}
+}
+
 // countChunkBytes returns the size of the chunks making up the provided series in bytes
 func countChunkBytes(series ...*storepb.Series) (count int) {
 	for _, s := range series {