From 51b3154b5d4c6e5638e38f84e7c67a94334ea5b1 Mon Sep 17 00:00:00 2001 From: Sungjin1212 Date: Fri, 6 Nov 2026 18:43:31 +0900 Subject: [PATCH] Allow 2h block conversion if it has no compact mark Signed-off-by: Sungjin1212 --- CHANGELOG.md | 1 + pkg/compactor/blocks_cleaner.go | 13 ++- pkg/compactor/blocks_cleaner_test.go | 8 +- pkg/parquetconverter/converter.go | 6 +- pkg/parquetconverter/converter_test.go | 119 +++++++++++++++++++--- pkg/storage/parquet/converter_marker.go | 11 +++ pkg/storage/parquet/util.go | 24 ++++- pkg/storage/parquet/util_test.go | 126 +++++++++++++++++++++++- 8 files changed, 282 insertions(+), 26 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6bfc9ac8532..314058076da 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ * [ENHANCEMENT] gRPC: When `-auth.enabled=true`, a gRPC request without the `X-Scope-OrgID` metadata is now rejected with the `UNAUTHENTICATED` status code instead of `UNKNOWN`. #7873 * [ENHANCEMENT] Upgrade promql-engine and Thanos to latest. #7889 * [ENHANCEMENT] Ruler: Retry a notification the Alertmanager failed to accept with a transport error or a 5xx response, instead of dropping it and leaving its alert undelivered until the next evaluation of the rule group. Enable the retries with `-ruler.notification.backoff-retries` (`0`, no retry, by default), and tune the delay between them with `-ruler.notification.backoff-min-period` and `-ruler.notification.backoff-max-period`. The retries are bounded by `-ruler.notification-timeout` and counted by the new `cortex_ruler_notification_retries_total` metric. #7909 +* [ENHANCEMENT] Parquet Converter: Allow 2h blocks conversion if the block has a no-compact-mark.json file. #6865 * [BUGFIX] Update prometheus alertmanager version to v0.34.1 to apply upstream inhibit rule bug fixes. #7877 * [BUGFIX] Tenant Federation: Fix regex tenant federation not working in single binary mode, and failing with `multiple org IDs present` when a matched tenant ID contains regex metacharacters and also matches other tenants as a regex. #7869 * [BUGFIX] Query Frontend: Fix the data selection time window computed for an instant query sent without a `time` parameter. The default evaluation time was passed to `util.ParseTimeParam` in milliseconds, but that helper formats the default and parses it as seconds, so the value overflowed `UnixNano()` and wrapped to a timestamp decades away from `now`. This made the `data_select_min_time` and `data_select_max_time` query stats wrong, and stopped the `time_window` query attribute of query priority and query rejection from matching such queries correctly. #7880 diff --git a/pkg/compactor/blocks_cleaner.go b/pkg/compactor/blocks_cleaner.go index 240696028a3..5e906a2e80c 100644 --- a/pkg/compactor/blocks_cleaner.go +++ b/pkg/compactor/blocks_cleaner.go @@ -781,11 +781,16 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userLogger log.Logger, us } level.Info(userLogger).Log("msg", "finish writing new index", "duration", time.Since(begin), "duration_ms", time.Since(begin).Milliseconds()) } - c.updateBucketMetrics(userID, parquetEnabled, idx, float64(len(partials)), float64(totalBlocksBlocksMarkedForNoCompaction)) + + noCompactMarkCheckFunc := func(blockID ulid.ULID) bool { + return cortex_parquet.ExistBlockNoCompact(ctx, userBucket, userLogger, blockID) + } + + c.updateBucketMetrics(userID, parquetEnabled, idx, float64(len(partials)), float64(totalBlocksBlocksMarkedForNoCompaction), noCompactMarkCheckFunc) return nil } -func (c *BlocksCleaner) updateBucketMetrics(userID string, parquetEnabled bool, idx *bucketindex.Index, partials, totalBlocksBlocksMarkedForNoCompaction float64) { +func (c *BlocksCleaner) updateBucketMetrics(userID string, parquetEnabled bool, idx *bucketindex.Index, partials, totalBlocksBlocksMarkedForNoCompaction float64, noCompactMarkCheckFunc func(ulid.ULID) bool) { c.tenantBlocks.WithLabelValues(userID).Set(float64(len(idx.Blocks))) c.tenantBlocksMarkedForDelete.WithLabelValues(userID).Set(float64(len(idx.BlockDeletionMarks))) c.tenantBlocksMarkedForNoCompaction.WithLabelValues(userID).Set(totalBlocksBlocksMarkedForNoCompaction) @@ -794,8 +799,10 @@ func (c *BlocksCleaner) updateBucketMetrics(userID string, parquetEnabled bool, if parquetEnabled { c.tenantParquetBlocks.WithLabelValues(userID).Set(float64(len(idx.ParquetBlocks()))) remainingBlocksToConvert := 0 + now := time.Now().UnixMilli() for _, b := range idx.NonParquetBlocks() { - if cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.cfg.BlockRanges) { + hasNoCompactMark := func() bool { return noCompactMarkCheckFunc(b.ID) } + if cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, now, c.cfg.BlockRanges, hasNoCompactMark) { remainingBlocksToConvert++ } } diff --git a/pkg/compactor/blocks_cleaner_test.go b/pkg/compactor/blocks_cleaner_test.go index db9e88989eb..4b371377061 100644 --- a/pkg/compactor/blocks_cleaner_test.go +++ b/pkg/compactor/blocks_cleaner_test.go @@ -366,7 +366,7 @@ func testBlocksCleanerWithOptions(t *testing.T, options testBlocksCleanerOptions cortex_bucket_parquet_blocks_count{user="user-6"} 1 # HELP cortex_bucket_parquet_unconverted_blocks_count Total number of unconverted parquet blocks in the bucket. Blocks marked for deletion are included. # TYPE cortex_bucket_parquet_unconverted_blocks_count gauge - cortex_bucket_parquet_unconverted_blocks_count{user="user-5"} 0 + cortex_bucket_parquet_unconverted_blocks_count{user="user-5"} 1 cortex_bucket_parquet_unconverted_blocks_count{user="user-6"} 0 `), "cortex_bucket_blocks_count", @@ -1225,8 +1225,12 @@ func TestBlocksCleaner_ParquetMetrics(t *testing.T) { }, } + mockNoCompactMarkCheckFunc := func(blockID ulid.ULID) bool { + return false + } + // Update metrics - cleaner.updateBucketMetrics("user1", true, idx, 0, 0) + cleaner.updateBucketMetrics("user1", true, idx, 0, 0, mockNoCompactMarkCheckFunc) // Verify metrics require.NoError(t, prom_testutil.CollectAndCompare(cleaner.tenantParquetBlocks, strings.NewReader(` diff --git a/pkg/parquetconverter/converter.go b/pkg/parquetconverter/converter.go index 06e46442562..466c614de98 100644 --- a/pkg/parquetconverter/converter.go +++ b/pkg/parquetconverter/converter.go @@ -423,7 +423,11 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin continue } - if !cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.blockRanges) { + hasNoCompactMark := func() bool { + return cortex_parquet.ExistBlockNoCompact(ctx, uBucket, logger, b.ULID) + } + + if !cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, time.Now().UnixMilli(), c.blockRanges, hasNoCompactMark) { continue } diff --git a/pkg/parquetconverter/converter_test.go b/pkg/parquetconverter/converter_test.go index 77ad7cfcc60..afb993e4ff4 100644 --- a/pkg/parquetconverter/converter_test.go +++ b/pkg/parquetconverter/converter_test.go @@ -19,10 +19,12 @@ import ( parquetgo "github.com/parquet-go/parquet-go" "github.com/prometheus-community/parquet-common/convert" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/client_golang/prometheus/testutil" "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/tsdb" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" "github.com/thanos-io/objstore" "github.com/thanos-io/objstore/providers/filesystem" @@ -327,6 +329,87 @@ func TestConverter_CleanupMetricsForNotOwnedUser(t *testing.T) { assert.Equal(t, 0.0, testutil.ToFloat64(converter.metrics.convertBlockFailures.WithLabelValues(userID))) } +func TestConverter_ShouldConvertNoCompactMarkBlocks(t *testing.T) { + cfg := prepareConfig() + userID := "user" + ringStore, closer := consul.NewInMemoryClient(ring.GetCodec(), log.NewNopLogger(), nil) + t.Cleanup(func() { assert.NoError(t, closer.Close()) }) + dir := t.TempDir() + + cfg.Ring.InstanceID = "parquet-converter-1" + cfg.Ring.InstanceAddr = "1.2.3.4" + cfg.Ring.KVStore.Mock = ringStore + bucketClient, err := filesystem.NewBucket(t.TempDir()) + require.NoError(t, err) + userBucket := bucket.NewPrefixedBucketClient(bucketClient, userID) + limits := &validation.Limits{} + flagext.DefaultValues(limits) + limits.ParquetConverterEnabled = true + + c, logger, _ := prepare(t, cfg, objstore.WithNoopInstr(bucketClient), limits, nil) + c.ringLifecycler = &ring.Lifecycler{ + Addr: "1.2.3.4", + } + + ctx := context.Background() + + lbls := labels.Labels{labels.Label{ + Name: "__name__", + Value: "test", + }} + + blocks := []ulid.ULID{} + numBlocks := 2 + mint := time.Hour * 13 + maxt := time.Hour * 15 + // Create 2h blocks + for range numBlocks { + rnd := rand.New(rand.NewSource(time.Now().Unix())) + id, err := e2e.CreateBlock(ctx, rnd, dir, []labels.Labels{lbls}, 2, mint.Milliseconds(), maxt.Milliseconds(), time.Minute.Milliseconds(), 10) + require.NoError(t, err) + blocks = append(blocks, id) + } + + // Create a 2h block that is too young to have its no-compact mark checked. + recentMaxt := time.Now().Truncate(2 * time.Hour) + recentMint := recentMaxt.Add(-2 * time.Hour) + recentBlock, err := e2e.CreateBlock(ctx, rand.New(rand.NewSource(time.Now().Unix())), dir, []labels.Labels{lbls}, 2, recentMint.UnixMilli(), recentMaxt.UnixMilli(), time.Minute.Milliseconds(), 10) + require.NoError(t, err) + + for _, bId := range append(blocks, recentBlock) { + blockDir := fmt.Sprintf("%s/%s", dir, bId.String()) + b, err := tsdb.OpenBlock(nil, blockDir, nil, nil) + require.NoError(t, err) + // upload block + err = block.Upload(ctx, logger, userBucket, b.Dir(), metadata.NoneFunc) + require.NoError(t, err) + + // upload no-compact-mark.json + err = block.MarkForNoCompact(ctx, logger, userBucket, bId, metadata.ManualNoCompactReason, "", promauto.With(nil).NewCounter(prometheus.CounterOpts{})) + require.NoError(t, err) + } + + ringMock := &ring.RingMock{} + ringMock.On("Get", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ring.ReplicationSet{ + Instances: []ring.InstanceDesc{ + { + Addr: "1.2.3.4", + }, + }, + }, nil) + + err = c.convertUser(ctx, log.NewNopLogger(), ringMock, userID) + require.NoError(t, err) + + // Verify the converted metric was incremented only for the old blocks + assert.Equal(t, float64(numBlocks), testutil.ToFloat64(c.metrics.convertedBlocks.WithLabelValues(userID))) + + // Verify the recent block was not converted, even though it has a no-compact mark + mark, err := parquet.ReadConverterMark(ctx, recentBlock, objstore.WithNoopInstr(userBucket), logger) + require.NoError(t, err) + assert.False(t, parquet.ValidConverterMarkVersion(mark.Version)) +} + func TestConverter_BlockConversionFailure(t *testing.T) { // Create a new registry for testing reg := prometheus.NewRegistry() @@ -376,7 +459,16 @@ func TestConverter_BlockConversionFailure(t *testing.T) { Addr: "1.2.3.4", } - err = converter.convertUser(context.Background(), logger, &RingMock{ReadRing: &ring.Ring{}}, userID) + ringMock := &ring.RingMock{} + ringMock.On("Get", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ring.ReplicationSet{ + Instances: []ring.InstanceDesc{ + { + Addr: "1.2.3.4", + }, + }, + }, nil) + + err = converter.convertUser(context.Background(), logger, ringMock, userID) require.NoError(t, err) // Verify the failure metric was incremented @@ -443,7 +535,16 @@ func TestConverter_ShouldNotFailOnAccessDenyError(t *testing.T) { Addr: "1.2.3.4", } - err = converter.convertUser(context.Background(), logger, &RingMock{ReadRing: &ring.Ring{}}, userID) + ringMock := &ring.RingMock{} + ringMock.On("Get", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(ring.ReplicationSet{ + Instances: []ring.InstanceDesc{ + { + Addr: "1.2.3.4", + }, + }, + }, nil) + + err = converter.convertUser(context.Background(), logger, ringMock, userID) require.Error(t, err) // Verify the failure metric was not incremented @@ -471,20 +572,6 @@ func (m *mockBucket) Get(ctx context.Context, name string) (io.ReadCloser, error return m.Bucket.Get(ctx, name) } -type RingMock struct { - ring.ReadRing -} - -func (r *RingMock) Get(key uint32, op ring.Operation, bufDescs []ring.InstanceDesc, bufHosts []string, bufZones map[string]int) (ring.ReplicationSet, error) { - return ring.ReplicationSet{ - Instances: []ring.InstanceDesc{ - { - Addr: "1.2.3.4", - }, - }, - }, nil -} - // mockTenantLimits implements the validation.TenantLimits interface for testing type mockTenantLimits struct { limits map[string]*validation.Limits diff --git a/pkg/storage/parquet/converter_marker.go b/pkg/storage/parquet/converter_marker.go index 55d85c45b3c..b46463b4520 100644 --- a/pkg/storage/parquet/converter_marker.go +++ b/pkg/storage/parquet/converter_marker.go @@ -9,8 +9,10 @@ import ( "github.com/efficientgo/core/errors" "github.com/go-kit/log" + "github.com/go-kit/log/level" "github.com/oklog/ulid/v2" "github.com/thanos-io/objstore" + "github.com/thanos-io/thanos/pkg/block/metadata" "github.com/thanos-io/thanos/pkg/runutil" "github.com/cortexproject/cortex/pkg/storage/bucket" @@ -34,6 +36,15 @@ type ConverterMark struct { Shards int `json:"shards,omitempty"` } +func ExistBlockNoCompact(ctx context.Context, userBkt objstore.InstrumentedBucket, logger log.Logger, blockID ulid.ULID) bool { + noCompactMarkerExists, err := userBkt.Exists(ctx, path.Join(blockID.String(), metadata.NoCompactMarkFilename)) + if err != nil { + level.Warn(logger).Log("msg", "unable to get stats of no-compact-mark.json for block", "block", blockID.String(), "err", err) + return false + } + return noCompactMarkerExists +} + func ReadConverterMark(ctx context.Context, id ulid.ULID, userBkt objstore.InstrumentedBucket, logger log.Logger) (*ConverterMark, error) { markerPath := path.Join(id.String(), ConverterMarkerFileName) reader, err := userBkt.WithExpectedErrs(bucket.IsOneOfTheExpectedErrors(userBkt.IsAccessDeniedErr, userBkt.IsObjNotFoundErr)).Get(ctx, markerPath) diff --git a/pkg/storage/parquet/util.go b/pkg/storage/parquet/util.go index 1e08ab301fa..7564fa0d715 100644 --- a/pkg/storage/parquet/util.go +++ b/pkg/storage/parquet/util.go @@ -1,8 +1,26 @@ package parquet -func ShouldConvertBlockToParquet(mint, maxt int64, timeRanges []int64) bool { - // We assume timeRanges[0] is the TSDB block duration (2h), and we don't convert them. - return getBlockTimeRange(mint, maxt, timeRanges) > timeRanges[0] +// ShouldConvertBlockToParquet returns whether the block should be converted to parquet. +// mint, maxt and now are timestamps in milliseconds. +func ShouldConvertBlockToParquet(mint, maxt, now int64, timeRanges []int64, hasNoCompactMark func() bool) bool { + // We assume timeRanges[0] is the TSDB block duration (2h). Bigger blocks are always converted. + if getBlockTimeRange(mint, maxt, timeRanges) > timeRanges[0] { + return true + } + + // We should check if 2h blocks have a `no-compact-mark.json` file + // since these will never be compacted to 12h block. + // We check if the `no-compact-mark.json` file exists only for blocks older + // than the next block range to reduce object storage calls, since younger + // blocks are still expected to be compacted. + return now-maxt >= noCompactMarkCheckAge(timeRanges) && hasNoCompactMark() +} + +func noCompactMarkCheckAge(timeRanges []int64) int64 { + if len(timeRanges) > 1 { + return timeRanges[1] + } + return timeRanges[0] } func getBlockTimeRange(mint, maxt int64, timeRanges []int64) int64 { diff --git a/pkg/storage/parquet/util_test.go b/pkg/storage/parquet/util_test.go index db0b3423366..9b900585cf0 100644 --- a/pkg/storage/parquet/util_test.go +++ b/pkg/storage/parquet/util_test.go @@ -15,6 +15,7 @@ func TestShouldConvertBlockToParquet(t *testing.T) { mint, maxt int64 durations tsdb.DurationList expected bool + checkFunc func() bool }{ { name: "2h block. Don't convert", @@ -22,6 +23,9 @@ func TestShouldConvertBlockToParquet(t *testing.T) { maxt: 2 * time.Hour.Milliseconds(), durations: tsdb.DurationList{2 * time.Hour, 12 * time.Hour}, expected: false, + checkFunc: func() bool { + return false + }, }, { name: "1h block. Don't convert", @@ -29,6 +33,29 @@ func TestShouldConvertBlockToParquet(t *testing.T) { maxt: 1 * time.Hour.Milliseconds(), durations: tsdb.DurationList{2 * time.Hour, 12 * time.Hour}, expected: false, + checkFunc: func() bool { + return false + }, + }, + { + name: "2h block. Exist NoCompactMark. Convert", + mint: 0, + maxt: 2 * time.Hour.Milliseconds(), + durations: tsdb.DurationList{2 * time.Hour, 12 * time.Hour}, + expected: true, + checkFunc: func() bool { + return true + }, + }, + { + name: "1h block. Exist NoCompactMark. Convert", + mint: 0, + maxt: 1 * time.Hour.Milliseconds(), + durations: tsdb.DurationList{2 * time.Hour, 12 * time.Hour}, + expected: true, + checkFunc: func() bool { + return true + }, }, { name: "3h block. Convert", @@ -36,6 +63,9 @@ func TestShouldConvertBlockToParquet(t *testing.T) { maxt: 3 * time.Hour.Milliseconds(), durations: tsdb.DurationList{2 * time.Hour, 12 * time.Hour}, expected: true, + checkFunc: func() bool { + return false + }, }, { name: "12h block. Convert", @@ -43,6 +73,9 @@ func TestShouldConvertBlockToParquet(t *testing.T) { maxt: 12 * time.Hour.Milliseconds(), durations: tsdb.DurationList{2 * time.Hour, 12 * time.Hour}, expected: true, + checkFunc: func() bool { + return false + }, }, { name: "12h block with 1h offset. Convert", @@ -50,6 +83,9 @@ func TestShouldConvertBlockToParquet(t *testing.T) { maxt: 13 * time.Hour.Milliseconds(), durations: tsdb.DurationList{2 * time.Hour, 12 * time.Hour}, expected: true, + checkFunc: func() bool { + return false + }, }, { name: "24h block. Convert", @@ -57,11 +93,99 @@ func TestShouldConvertBlockToParquet(t *testing.T) { maxt: 24 * time.Hour.Milliseconds(), durations: tsdb.DurationList{2 * time.Hour, 12 * time.Hour}, expected: true, + checkFunc: func() bool { + return false + }, }, } { t.Run(tc.name, func(t *testing.T) { - res := ShouldConvertBlockToParquet(tc.mint, tc.maxt, (&tc.durations).ToMilliseconds()) + now := tc.maxt + (24 * time.Hour).Milliseconds() + res := ShouldConvertBlockToParquet(tc.mint, tc.maxt, now, (&tc.durations).ToMilliseconds(), tc.checkFunc) + require.Equal(t, tc.expected, res) + }) + } +} + +func TestShouldConvertBlockToParquet_NoCompactMarkCheckAge(t *testing.T) { + const checkAfter = 12 * time.Hour + + // Real timestamps are used on purpose: the block's age has to be computed + // from the current time, not by comparing a timestamp with a duration. + maxt := time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC) + mint := maxt.Add(-2 * time.Hour) + + for _, tc := range []struct { + name string + durations tsdb.DurationList + age time.Duration + hasMark bool + expectedCheck bool + expected bool + }{ + { + name: "block ended one hour ago. Don't check the marker", + age: time.Hour, + hasMark: true, + }, + { + name: "block is just younger than the threshold. Don't check the marker", + age: checkAfter - time.Millisecond, + hasMark: true, + }, + { + name: "block is exactly as old as the threshold. Convert", + age: checkAfter, + hasMark: true, + expectedCheck: true, + expected: true, + }, + { + name: "block is older than the threshold. Convert", + age: 48 * time.Hour, + hasMark: true, + expectedCheck: true, + expected: true, + }, + { + name: "block is older than the threshold without marker. Don't convert", + age: 48 * time.Hour, + expectedCheck: true, + }, + { + name: "block ends in the future. Don't check the marker", + age: -time.Hour, + hasMark: true, + }, + { + name: "threshold follows the configured block ranges. Don't check the marker", + durations: tsdb.DurationList{2 * time.Hour, 24 * time.Hour}, + age: checkAfter, + hasMark: true, + }, + { + name: "single block range uses that range as threshold. Convert", + durations: tsdb.DurationList{2 * time.Hour}, + age: 2 * time.Hour, + hasMark: true, + expectedCheck: true, + expected: true, + }, + } { + t.Run(tc.name, func(t *testing.T) { + checked := false + checkFunc := func() bool { + checked = true + return tc.hasMark + } + + durations := tc.durations + if durations == nil { + durations = tsdb.DurationList{2 * time.Hour, checkAfter} + } + + res := ShouldConvertBlockToParquet(mint.UnixMilli(), maxt.UnixMilli(), maxt.Add(tc.age).UnixMilli(), (&durations).ToMilliseconds(), checkFunc) require.Equal(t, tc.expected, res) + require.Equal(t, tc.expectedCheck, checked) }) } }