querier: drop redundant detachChunksFromBuffer copy

CHANGELOG.md

@@ -1,6 +1,8 @@
 # Changelog
 
 ## master / unreleased
+* [BUGFIX] Ingester: Fix `cortex_ingester_ingestion_delay_seconds` native histogram losing ~86% of observations by setting `NativeHistogramMinResetDuration` to 1h instead of the bare integer `1` (interpreted as 1ns). #7731
+* [BUGFIX] Querier: Remove the redundant `detachChunksFromBuffer` copy in the distributor queryable. gogo `Chunk.Unmarshal` already allocates chunk data separately, so the extra copy was a wasted allocation + memcpy that increased peak heap on the ingester-read path. #7732
 * [FEATURE] Engine: Add `-querier.selector-batch-size` and `-ruler.selector-batch-size` flags to configure series batching in the Thanos promQL engine. 0 disables batching. #7763
 * [CHANGE] Querier: Make query time range configurations per-tenant: `query_ingesters_within`, `query_store_after`, and `shuffle_sharding_ingesters_lookback_period`. Uses `model.Duration` instead of `time.Duration` to support serialization but has minimum unit of 1ms (nanoseconds/microseconds not supported). #7160
 * [CHANGE] Cache: Setting `-blocks-storage.bucket-store.metadata-cache.bucket-index-content-ttl` to 0 will disable the bucket-index cache. #7446

pkg/ingester/metrics.go

@@ -88,6 +88,20 @@ type ingesterMetrics struct {
 	unoptimizedRegexRejectedTotal    *prometheus.CounterVec
 }
 
+// ingestionDelaySecondsHistogramOpts defines the options for the
+// cortex_ingester_ingestion_delay_seconds native histogram. NativeHistogramMinResetDuration
+// must be a real duration (time.Hour): a bare integer literal is interpreted as
+// nanoseconds, which resets the native histogram on essentially every scrape and
+// discards the vast majority of observations (see cortexproject/cortex#7731).
+var ingestionDelaySecondsHistogramOpts = prometheus.HistogramOpts{
+	Name:                            "cortex_ingester_ingestion_delay_seconds",
+	Help:                            "Delay in seconds between sample ingestion time and sample timestamp.",
+	NativeHistogramBucketFactor:     1.1,
+	NativeHistogramMaxBucketNumber:  100,
+	NativeHistogramMinResetDuration: 1 * time.Hour,
+	Buckets:                         []float64{1, 5, 10, 30, 60, 120, 300, 600}, // 1s, 5s, 10s, 30s, 1m, 2m, 5m, 10m
+}
+
 func newIngesterMetrics(r prometheus.Registerer,
 	createMetricsConflictingWithTSDB bool,
 	activeSeriesEnabled bool,
@@ -151,14 +165,7 @@ func newIngesterMetrics(r prometheus.Registerer,
 			NativeHistogramMinResetDuration: 1 * time.Hour,
 			Buckets:                         prometheus.ExponentialBuckets(1, 2, 10), // 1 to 512 buckets
 		}, []string{"user"}),
-		ingestionDelaySeconds: promauto.With(r).NewHistogramVec(prometheus.HistogramOpts{
-			Name:                            "cortex_ingester_ingestion_delay_seconds",
-			Help:                            "Delay in seconds between sample ingestion time and sample timestamp.",
-			NativeHistogramBucketFactor:     1.1,
-			NativeHistogramMaxBucketNumber:  100,
-			NativeHistogramMinResetDuration: 1,
-			Buckets:                         []float64{1, 5, 10, 30, 60, 120, 300, 600}, // 1s, 5s, 10s, 30s, 1m, 2m, 5m, 10m
-		}, []string{"user"}),
+		ingestionDelaySeconds: promauto.With(r).NewHistogramVec(ingestionDelaySecondsHistogramOpts, []string{"user"}),
 		oooLabelsTotal: promauto.With(r).NewCounterVec(prometheus.CounterOpts{
 			Name: "cortex_ingester_out_of_order_labels_total",
 			Help: "The total number of out of order label found per user.",

pkg/ingester/metrics_test.go

@@ -3,6 +3,7 @@ package ingester
 import (
 	"bytes"
 	"testing"
+	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
@@ -1298,3 +1299,14 @@ func populateTSDBMetrics(base float64) *prometheus.Registry {
 
 	return r
 }
+
+// TestIngestionDelaySecondsHistogramResetDuration is a regression test for
+// cortexproject/cortex#7731: ingestionDelaySeconds was registered with
+// NativeHistogramMinResetDuration: 1, where 1 is interpreted as 1 nanosecond
+// (the field is a time.Duration). That reset the native histogram on essentially
+// every scrape and discarded ~86% of observations. The reset duration must be a
+// real hour-long window.
+func TestIngestionDelaySecondsHistogramResetDuration(t *testing.T) {
+	require.Equal(t, time.Hour, ingestionDelaySecondsHistogramOpts.NativeHistogramMinResetDuration,
+		"NativeHistogramMinResetDuration must be time.Hour, not a bare integer (interpreted as nanoseconds)")
+}

pkg/querier/distributor_queryable.go

@@ -186,10 +186,13 @@ func (q *distributorQuerier) streamingSelect(ctx context.Context, sortSeries, pa
 			continue
 		}
 
-		// Detach label and chunk data from gRPC unmarshal buffers so the Go GC
-		// can reclaim receive buffers and reduce heap usage.
+		// Detach label data from the gRPC unmarshal buffer so the Go GC can
+		// reclaim receive buffers and reduce heap usage. Chunk data does NOT
+		// need detaching: the gogo-generated Chunk.Unmarshal allocates a fresh
+		// slice per Recv, so chunk Data never aliases the wire buffer (see
+		// cortexproject/cortex#7732).
 		ls := cortexpb.FromLabelAdaptersToLabelsWithCopy(result.Labels)
-		chunks, err := chunkcompat.FromChunks(ls, detachChunksFromBuffer(result.Chunks))
+		chunks, err := chunkcompat.FromChunks(ls, result.Chunks)
 		if err != nil {
 			return storage.ErrSeriesSet(err)
 		}
@@ -454,13 +457,3 @@ func labelHintsToSelectHints(hints *storage.LabelHints) *storage.SelectHints {
 // detachChunksFromBuffer returns a copy of the chunks slice with data byte
 // slices re-allocated so that the series no longer references the gRPC
 // unmarshal buffer, allowing the Go GC to reclaim receive buffers.
-func detachChunksFromBuffer(chunks []client.Chunk) []client.Chunk {
-	copied := make([]client.Chunk, len(chunks))
-	for i, c := range chunks {
-		copied[i] = c
-		if len(c.Data) > 0 {
-			copied[i].Data = append([]byte(nil), c.Data...)
-		}
-	}
-	return copied
-}

pkg/querier/distributor_queryable_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"testing"
 	"time"
+	"unsafe"
 
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/labels"
@@ -689,24 +690,58 @@ func BenchmarkIngesterStreamingSelect(b *testing.B) {
 		return &client.QueryStreamResponse{Chunkseries: series}
 	}
 
-	b.Run("with_detach", func(b *testing.B) {
+	b.Run("without_detach", func(b *testing.B) {
 		b.ReportAllocs()
 		for i := 0; i < b.N; i++ {
 			resp := buildResponse()
 			for _, result := range resp.Chunkseries {
 				_ = cortexpb.FromLabelAdaptersToLabelsWithCopy(result.Labels)
-				_ = detachChunksFromBuffer(result.Chunks)
 			}
 		}
 	})
+}
 
-	b.Run("without_detach", func(b *testing.B) {
-		b.ReportAllocs()
-		for i := 0; i < b.N; i++ {
-			resp := buildResponse()
-			for _, result := range resp.Chunkseries {
-				_ = cortexpb.FromLabelAdaptersToLabels(result.Labels)
-			}
-		}
-	})
+// TestChunksDoNotAliasWireBuffer is a regression test for
+// cortexproject/cortex#7732: the gogo-generated Chunk.Unmarshal allocates a
+// fresh slice for each chunk's Data, so chunk data never aliases the gRPC wire
+// buffer. The distributor queryable must therefore pass result.Chunks directly
+// to chunkcompat.FromChunks rather than copying it first (the copy was a wasted
+// allocation + memcpy that, because the response stays live for the whole
+// streamingSelect body, actually increased peak heap).
+func TestChunksDoNotAliasWireBuffer(t *testing.T) {
+	chunkData := []byte("some-chunk-bytes-that-should-not-alias")
+	resp := &client.QueryStreamResponse{
+		Chunkseries: []client.TimeSeriesChunk{
+			{
+				Labels: []cortexpb.LabelAdapter{{Name: "foo", Value: "bar"}},
+				Chunks: []client.Chunk{
+					{StartTimestampMs: 0, EndTimestampMs: 1000, Data: chunkData},
+				},
+			},
+		},
+	}
+
+	marshaled, err := resp.Marshal()
+	require.NoError(t, err)
+
+	var unmarshaled client.QueryStreamResponse
+	require.NoError(t, unmarshaled.Unmarshal(marshaled))
+
+	got := unmarshaled.Chunkseries[0].Chunks[0].Data
+	require.Equal(t, chunkData, got)
+
+	// The unmarshaled chunk Data must be a distinct allocation from the wire
+	// buffer, proving the redundant detach copy in #7670 was unnecessary.
+	require.False(t, aliases(got, marshaled),
+		"chunk Data aliases the wire buffer; the redundant detach copy may be needed after all")
+}
+
+// aliases reports whether b's backing array overlaps a's backing array.
+func aliases(a, b []byte) bool {
+	if len(a) == 0 || len(b) == 0 {
+		return false
+	}
+	a0, a1 := uintptr(unsafe.Pointer(&a[0])), uintptr(unsafe.Pointer(&a[0]))+uintptr(len(a))
+	b0, b1 := uintptr(unsafe.Pointer(&b[0])), uintptr(unsafe.Pointer(&b[0]))+uintptr(len(b))
+	return a0 < b1 && b0 < a1
 }

pkg/util/request_tracker/request_extractor.go

@@ -83,7 +83,10 @@ func trimStringByBytes(str string, size int) string {
 	bytesStr := []byte(str)
 	trimIndex := len(bytesStr)
 	if size < len(bytesStr) {
-		for !utf8.RuneStart(bytesStr[size]) {
+		// Scan backwards to a rune boundary. Bound the scan at size > 0: if the
+		// string has no rune start (e.g. only UTF-8 continuation bytes) the loop
+		// must not underflow past zero, which would panic on bytesStr[-1].
+		for size > 0 && !utf8.RuneStart(bytesStr[size]) {
 			size--
 		}
 		trimIndex = size

pkg/util/request_tracker/request_tracker_test.go

@@ -191,3 +191,20 @@ func TestRangedQueryExtractorMultiByteTruncation(t *testing.T) {
 		assert.True(t, utf8.Valid(entry), "entry should be valid UTF-8")
 	})
 }
+
+// TestTrimForJsonMarshalContinuationBytes reproduces cortexproject/cortex#7729:
+// when the string consists only of UTF-8 continuation bytes (0x80-0xBF) there is
+// no rune start to scan back to, so the backwards scan in trimStringByBytes
+// underflows past zero and panics with index out of range [-1]. The fix bounds
+// the scan at size > 0.
+func TestTrimForJsonMarshalContinuationBytes(t *testing.T) {
+	// 1200 continuation bytes (0x80) with a truncation size below the length.
+	continuation := strings.Repeat("\x80", 1200)
+
+	require.NotPanics(t, func() {
+		out := trimForJsonMarshal(continuation, 900)
+		// Result must always be valid UTF-8 (no partial runes).
+		assert.True(t, utf8.ValidString(out), "result should be valid UTF-8")
+		assert.LessOrEqual(t, len(out), len(continuation))
+	})
+}