diff --git a/CHANGELOG.md b/CHANGELOG.md index 4fe16f3a8bd..7e5100d4cea 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/pkg/ingester/metrics.go b/pkg/ingester/metrics.go index 3ad21faad6d..54a573da1e6 100644 --- a/pkg/ingester/metrics.go +++ b/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.", diff --git a/pkg/ingester/metrics_test.go b/pkg/ingester/metrics_test.go index af09d07e2a3..5a43e420b93 100644 --- a/pkg/ingester/metrics_test.go +++ b/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)") +} diff --git a/pkg/querier/distributor_queryable.go b/pkg/querier/distributor_queryable.go index 42f5fb59d4a..19ceb6304b5 100644 --- a/pkg/querier/distributor_queryable.go +++ b/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 -} diff --git a/pkg/querier/distributor_queryable_test.go b/pkg/querier/distributor_queryable_test.go index f4dfa0d4dde..c6a28174b36 100644 --- a/pkg/querier/distributor_queryable_test.go +++ b/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 } diff --git a/pkg/util/request_tracker/request_extractor.go b/pkg/util/request_tracker/request_extractor.go index cbd9f31e8fe..7675d026f8b 100644 --- a/pkg/util/request_tracker/request_extractor.go +++ b/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 diff --git a/pkg/util/request_tracker/request_tracker_test.go b/pkg/util/request_tracker/request_tracker_test.go index 13ef0a802cd..9842b33ec2f 100644 --- a/pkg/util/request_tracker/request_tracker_test.go +++ b/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)) + }) +}