Add NoCompactMarkCheckAfter config

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

pkg/compactor/blocks_cleaner.go

@@ -45,6 +45,7 @@ type BlocksCleanerConfig struct {
 	ShardingStrategy                   string
 	CompactionStrategy                 string
 	BlockRanges                        []int64
+	NoCompactMarkCheckAfter            time.Duration
 }
 
 type BlocksCleaner struct {
@@ -777,7 +778,7 @@ func (c *BlocksCleaner) updateBucketMetrics(userID string, parquetEnabled bool,
 		c.tenantParquetBlocks.WithLabelValues(userID).Set(float64(len(idx.ParquetBlocks())))
 		remainingBlocksToConvert := 0
 		for _, b := range idx.NonParquetBlocks() {
-			if cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.cfg.BlockRanges, b.ID, noCompactMarkCheckFunc) {
+			if cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.cfg.NoCompactMarkCheckAfter.Milliseconds(), c.cfg.BlockRanges, b.ID, noCompactMarkCheckFunc) {
 				remainingBlocksToConvert++
 			}
 		}

pkg/compactor/compactor.go

@@ -304,6 +304,9 @@ type Config struct {
 	AcceptMalformedIndex        bool `yaml:"accept_malformed_index"`
 	CachingBucketEnabled        bool `yaml:"caching_bucket_enabled"`
 	CleanerCachingBucketEnabled bool `yaml:"cleaner_caching_bucket_enabled"`
+
+	// Injected internally
+	NoCompactMarkCheckAfter time.Duration `yaml:"-"`
 }
 
 // RegisterFlags registers the Compactor flags.
@@ -753,6 +756,7 @@ func (c *Compactor) starting(ctx context.Context) error {
 		ShardingStrategy:                   c.compactorCfg.ShardingStrategy,
 		CompactionStrategy:                 c.compactorCfg.CompactionStrategy,
 		BlockRanges:                        c.compactorCfg.BlockRanges.ToMilliseconds(),
+		NoCompactMarkCheckAfter:            c.compactorCfg.NoCompactMarkCheckAfter,
 	}, cleanerBucketClient, cleanerUsersScanner, c.compactorCfg.CompactionVisitMarkerTimeout, c.limits, c.parentLogger, cleanerRingLifecyclerID, c.registerer, c.compactorCfg.CleanerVisitMarkerTimeout, c.compactorCfg.CleanerVisitMarkerFileUpdateInterval,
 		c.compactorMetrics.syncerBlocksMarkedForDeletion, c.compactorMetrics.remainingPlannedCompactions)
 

pkg/cortex/modules.go

@@ -737,6 +737,7 @@ func (t *Cortex) initParquetConverter() (serv services.Service, err error) {
 func (t *Cortex) initCompactor() (serv services.Service, err error) {
 	t.Cfg.Compactor.ShardingRing.ListenPort = t.Cfg.Server.GRPCListenPort
 	ingestionReplicationFactor := t.Cfg.Ingester.LifecyclerConfig.RingConfig.ReplicationFactor
+	t.Cfg.Compactor.NoCompactMarkCheckAfter = t.Cfg.ParquetConverter.NoCompactMarkCheckAfter
 
 	t.Compactor, err = compactor.NewCompactor(t.Cfg.Compactor, t.Cfg.BlocksStorage, util_log.Logger, prometheus.DefaultRegisterer, t.Overrides, ingestionReplicationFactor)
 	if err != nil {

pkg/parquetconverter/converter.go

@@ -54,10 +54,11 @@ var RingOp = ring.NewOp([]ring.InstanceState{ring.ACTIVE}, nil)
 type NoCompactMarkCheckFunc = func(bId ulid.ULID) bool
 
 type Config struct {
-	MetaSyncConcurrency int           `yaml:"meta_sync_concurrency"`
-	ConversionInterval  time.Duration `yaml:"conversion_interval"`
-	MaxRowsPerRowGroup  int           `yaml:"max_rows_per_row_group"`
-	FileBufferEnabled   bool          `yaml:"file_buffer_enabled"`
+	MetaSyncConcurrency     int           `yaml:"meta_sync_concurrency"`
+	ConversionInterval      time.Duration `yaml:"conversion_interval"`
+	MaxRowsPerRowGroup      int           `yaml:"max_rows_per_row_group"`
+	FileBufferEnabled       bool          `yaml:"file_buffer_enabled"`
+	NoCompactMarkCheckAfter time.Duration `yaml:"no_compact_mark_check_after"`
 
 	DataDir string `yaml:"data_dir"`
 
@@ -109,6 +110,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&cfg.MaxRowsPerRowGroup, "parquet-converter.max-rows-per-row-group", 1e6, "Max number of rows per parquet row group.")
 	f.DurationVar(&cfg.ConversionInterval, "parquet-converter.conversion-interval", time.Minute, "The frequency at which the conversion job runs.")
 	f.BoolVar(&cfg.FileBufferEnabled, "parquet-converter.file-buffer-enabled", true, "Whether to enable buffering the writes in disk to reduce memory utilization.")
+	f.DurationVar(&cfg.NoCompactMarkCheckAfter, "parquet-converter.no-compact-mark-check-after", time.Hour*13, "The time after which a `no-compact-mark.json` file should be checked.")
 }
 
 func NewConverter(cfg Config, storageCfg cortex_tsdb.BlocksStorageConfig, blockRanges []int64, logger log.Logger, registerer prometheus.Registerer, limits *validation.Overrides) (*Converter, error) {
@@ -392,7 +394,7 @@ func (c *Converter) convertUser(ctx context.Context, logger log.Logger, ring rin
 			return cortex_parquet.ExistBlockNoCompact(ctx, uBucket, logger, b.ULID)
 		}
 
-		if !cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.blockRanges, b.ULID, noCompactMarkCheckFunc) {
+		if !cortex_parquet.ShouldConvertBlockToParquet(b.MinTime, b.MaxTime, c.cfg.NoCompactMarkCheckAfter.Milliseconds(), c.blockRanges, b.ULID, noCompactMarkCheckFunc) {
 			continue
 		}
 

pkg/storage/parquet/util.go

@@ -6,15 +6,21 @@ import (
 
 type NoCompactMarkCheckFunc = func(bId ulid.ULID) bool
 
-func ShouldConvertBlockToParquet(mint, maxt int64, timeRanges []int64, bId ulid.ULID, checkFunc NoCompactMarkCheckFunc) bool {
+func ShouldConvertBlockToParquet(mint, maxt, noCompactMarkCheckAfter int64, timeRanges []int64, bId ulid.ULID, checkFunc NoCompactMarkCheckFunc) bool {
 	// We assume timeRanges[0] is the TSDB block duration (2h), and we don't convert them.
 	blockTimeRange := getBlockTimeRange(mint, maxt, timeRanges)
 	if blockTimeRange > timeRanges[0] {
 		return true
 	}
 
-	if blockTimeRange == timeRanges[0] && checkFunc(bId) {
-		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
+	// after the noCompactMarkCheckAfter to reduce calls to `checkFunc`.
+	if mint >= noCompactMarkCheckAfter {
+		if blockTimeRange == timeRanges[0] && checkFunc(bId) {
+			return true
+		}
 	}
 	return false
 }

pkg/storage/parquet/util_test.go

@@ -14,11 +14,12 @@ import (
 
 func TestShouldConvertBlockToParquet(t *testing.T) {
 	for _, tc := range []struct {
-		name       string
-		mint, maxt int64
-		durations  tsdb.DurationList
-		expected   bool
-		checkFunc  NoCompactMarkCheckFunc
+		name                    string
+		mint, maxt              int64
+		noCompactMarkCheckAfter int64
+		durations               tsdb.DurationList
+		expected                bool
+		checkFunc               NoCompactMarkCheckFunc
 	}{
 		{
 			name:      "2h block. Don't convert",
@@ -50,6 +51,17 @@ func TestShouldConvertBlockToParquet(t *testing.T) {
 				return true
 			},
 		},
+		{
+			name:                    "2h block. Exist NoCompactMark. noCompactMarkCheckAfter is one hour. Not Convert",
+			mint:                    0,
+			maxt:                    2 * time.Hour.Milliseconds(),
+			noCompactMarkCheckAfter: 1 * time.Hour.Milliseconds(),
+			durations:               tsdb.DurationList{2 * time.Hour, 12 * time.Hour},
+			expected:                false,
+			checkFunc: func(bId ulid.ULID) bool {
+				return true
+			},
+		},
 		{
 			name:      "1h block. Exist NoCompactMark. Convert",
 			mint:      0,
@@ -104,7 +116,7 @@ func TestShouldConvertBlockToParquet(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			id, err := ulid.New(ulid.Now(), rand.Reader)
 			require.NoError(t, err)
-			res := ShouldConvertBlockToParquet(tc.mint, tc.maxt, (&tc.durations).ToMilliseconds(), id, tc.checkFunc)
+			res := ShouldConvertBlockToParquet(tc.mint, tc.maxt, tc.noCompactMarkCheckAfter, (&tc.durations).ToMilliseconds(), id, tc.checkFunc)
 			require.Equal(t, tc.expected, res)
 		})
 	}