refactored with thanos changes

Signed-off-by: Alex Le <[email protected]>

Author: alexqyle

pkg/compactor/background_chunks_series_set.go

@@ -0,0 +1,61 @@
+package compactor
+
+import(
+	"context"
+
+	"github.com/prometheus/prometheus/storage"
+)
+
+type backgrounChunkSeriesSet struct {
+	nextSet chan storage.ChunkSeries
+	actual  storage.ChunkSeries
+	cs      storage.ChunkSeriesSet
+}
+
+func (b *backgrounChunkSeriesSet) Next() bool {
+	select {
+	case s, ok := <-b.nextSet:
+		b.actual = s
+		return ok
+	}
+}
+
+func (b *backgrounChunkSeriesSet) At() storage.ChunkSeries {
+	return b.actual
+}
+
+func (b *backgrounChunkSeriesSet) Err() error {
+	return b.cs.Err()
+}
+
+func (b *backgrounChunkSeriesSet) Warnings() storage.Warnings {
+	return b.cs.Warnings()
+}
+
+func (b *backgrounChunkSeriesSet) run(ctx context.Context) {
+	for {
+		if (!b.cs.Next()) {
+			close(b.nextSet)
+			return
+		}
+
+		select {
+		case b.nextSet <- b.cs.At():
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+func NewBackgroundChunkSeriesSet(ctx context.Context, cs storage.ChunkSeriesSet) storage.ChunkSeriesSet {
+	r :=  &backgrounChunkSeriesSet{
+		cs: cs,
+		nextSet: make(chan storage.ChunkSeries, 1000),
+	}
+
+	go func() {
+		r.run(ctx)
+	}()
+
+	return r
+}

pkg/compactor/block_visit_marker_test.go

@@ -153,7 +153,7 @@ func TestMarkBlockVisitedHeartBeat(t *testing.T) {
 			}
 			time.Sleep(2 * time.Second)
 			for _, meta := range blocks {
-				res, err := ReadBlockVisitMarker(ctx, objstore.WithNoopInstr(bkt), logger, meta.ULID.String(), partitionID, dummyCounter)
+				res, err := ReadBlockVisitMarker(context.Background(), objstore.WithNoopInstr(bkt), logger, meta.ULID.String(), partitionID, dummyCounter)
 				require.NoError(t, err)
 				require.Equal(t, tcase.expectedStatus, res.Status)
 			}

pkg/compactor/compactor.go

@@ -97,7 +97,8 @@ var (
 			blockVisitMarkerWriteFailed,
 			partitionedGroupInfoReadFailed,
 			partitionedGroupInfoWriteFailed,
-			noCompactionMarkFilter.NoCompactMarkedBlocks)
+			noCompactionMarkFilter.NoCompactMarkedBlocks,
+		)
 	}
 
 	DefaultBlocksCompactorFactory = func(ctx context.Context, cfg Config, logger log.Logger, reg prometheus.Registerer) (compact.Compactor, PlannerFactory, error) {
@@ -126,12 +127,12 @@ var (
 		return compactor, plannerFactory, nil
 	}
 
-	DefaultCompactionCompleteCheckerFactory = func(_ context.Context, _ objstore.InstrumentedBucket, _ log.Logger, _ prometheus.Counter, _ prometheus.Counter) compact.ExtraCompactionCompleteChecker {
-		return compact.DefaultCompactionCompleteChecker{}
+	DefaultBlockDeletableCheckerFactory = func(_ context.Context, _ objstore.InstrumentedBucket, _ log.Logger, _ prometheus.Counter, _ prometheus.Counter) compact.BlockDeletableChecker {
+		return compact.DefaultBlockDeletableChecker{}
 	}
 
-	PartitionCompactionCompleteCheckerFactory = func(ctx context.Context, bkt objstore.InstrumentedBucket, logger log.Logger, blockVisitMarkerReadFailed prometheus.Counter, partitionedGroupInfoWriteFailed prometheus.Counter) compact.ExtraCompactionCompleteChecker {
-		return NewPartitionCompactionCompleteChecker(ctx, bkt, logger, blockVisitMarkerReadFailed, partitionedGroupInfoWriteFailed)
+	PartitionCompactionBlockDeletableCheckerFactory = func(ctx context.Context, bkt objstore.InstrumentedBucket, logger log.Logger, blockVisitMarkerReadFailed prometheus.Counter, partitionedGroupInfoWriteFailed prometheus.Counter) compact.BlockDeletableChecker {
+		return NewPartitionCompactionBlockDeletableChecker(ctx, bkt, logger, blockVisitMarkerReadFailed, partitionedGroupInfoWriteFailed)
 	}
 )
 
@@ -176,13 +177,13 @@ type PlannerFactory func(
 	blockVisitMarkerWriteFailed prometheus.Counter,
 ) compact.Planner
 
-type CompactionCompleteCheckerFactory func(
+type BlockDeletableCheckerFactory func(
 	ctx context.Context,
 	bkt objstore.InstrumentedBucket,
 	logger log.Logger,
 	blockVisitMarkerReadFailed prometheus.Counter,
 	partitionedGroupInfoReadFailed prometheus.Counter,
-) compact.ExtraCompactionCompleteChecker
+) compact.BlockDeletableChecker
 
 // Limits defines limits used by the Compactor.
 type Limits interface {
@@ -334,7 +335,7 @@ type Compactor struct {
 
 	blocksPlannerFactory PlannerFactory
 
-	compactionCompleteCheckerFactory CompactionCompleteCheckerFactory
+	blockDeletableCheckerFactory BlockDeletableCheckerFactory
 
 	// Client used to run operations on the bucket storing blocks.
 	bucketClient objstore.Bucket
@@ -347,7 +348,6 @@ type Compactor struct {
 
 	// Metrics.
 	compactionRunsStarted           prometheus.Counter
-	compactionRunsInterrupted       prometheus.Counter
 	compactionRunsCompleted         prometheus.Counter
 	compactionRunsFailed            prometheus.Counter
 	compactionRunsLastSuccess       prometheus.Gauge
@@ -393,14 +393,14 @@ func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.BlocksStorageConfi
 		}
 	}
 
-	var compactionCompleteCheckerFactory CompactionCompleteCheckerFactory
+	var blockDeletableCheckerFactory BlockDeletableCheckerFactory
 	if compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle {
-		compactionCompleteCheckerFactory = PartitionCompactionCompleteCheckerFactory
+		blockDeletableCheckerFactory = PartitionCompactionBlockDeletableCheckerFactory
 	} else {
-		compactionCompleteCheckerFactory = DefaultCompactionCompleteCheckerFactory
+		blockDeletableCheckerFactory = DefaultBlockDeletableCheckerFactory
 	}
 
-	cortexCompactor, err := newCompactor(compactorCfg, storageCfg, cfgProvider, logger, registerer, bucketClientFactory, blocksGrouperFactory, blocksCompactorFactory, compactionCompleteCheckerFactory, limits)
+	cortexCompactor, err := newCompactor(compactorCfg, storageCfg, cfgProvider, logger, registerer, bucketClientFactory, blocksGrouperFactory, blocksCompactorFactory, blockDeletableCheckerFactory, limits)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create Cortex blocks compactor")
 	}
@@ -417,7 +417,7 @@ func newCompactor(
 	bucketClientFactory func(ctx context.Context) (objstore.Bucket, error),
 	blocksGrouperFactory BlocksGrouperFactory,
 	blocksCompactorFactory BlocksCompactorFactory,
-	compactionCompleteCheckerFactory CompactionCompleteCheckerFactory,
+	blockDeletableCheckerFactory BlockDeletableCheckerFactory,
 	limits Limits,
 ) (*Compactor, error) {
 	var remainingPlannedCompactions prometheus.Gauge
@@ -428,27 +428,23 @@ func newCompactor(
 		})
 	}
 	c := &Compactor{
-		compactorCfg:                     compactorCfg,
-		storageCfg:                       storageCfg,
-		cfgProvider:                      cfgProvider,
-		parentLogger:                     logger,
-		logger:                           log.With(logger, "component", "compactor"),
-		registerer:                       registerer,
-		syncerMetrics:                    newSyncerMetrics(registerer),
-		bucketClientFactory:              bucketClientFactory,
-		blocksGrouperFactory:             blocksGrouperFactory,
-		blocksCompactorFactory:           blocksCompactorFactory,
-		compactionCompleteCheckerFactory: compactionCompleteCheckerFactory,
-		allowedTenants:                   util.NewAllowedTenants(compactorCfg.EnabledTenants, compactorCfg.DisabledTenants),
+		compactorCfg:                 compactorCfg,
+		storageCfg:                   storageCfg,
+		cfgProvider:                  cfgProvider,
+		parentLogger:                 logger,
+		logger:                       log.With(logger, "component", "compactor"),
+		registerer:                   registerer,
+		syncerMetrics:                newSyncerMetrics(registerer),
+		bucketClientFactory:          bucketClientFactory,
+		blocksGrouperFactory:         blocksGrouperFactory,
+		blocksCompactorFactory:       blocksCompactorFactory,
+		blockDeletableCheckerFactory: blockDeletableCheckerFactory,
+		allowedTenants:               util.NewAllowedTenants(compactorCfg.EnabledTenants, compactorCfg.DisabledTenants),
 
 		compactionRunsStarted: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_compactor_runs_started_total",
 			Help: "Total number of compaction runs started.",
 		}),
-		compactionRunsInterrupted: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
-			Name: "cortex_compactor_runs_interrupted_total",
-			Help: "Total number of compaction runs interrupted.",
-		}),
 		compactionRunsCompleted: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_compactor_runs_completed_total",
 			Help: "Total number of compaction runs successfully completed.",
@@ -656,23 +652,16 @@ func (c *Compactor) running(ctx context.Context) error {
 }
 
 func (c *Compactor) compactUsers(ctx context.Context) {
-	failed := false
-	interrupted := false
+	succeeded := false
+	compactionErrorCount := 0
 
 	c.compactionRunsStarted.Inc()
 
 	defer func() {
-		// interruptions and successful runs are considered
-		// mutually exclusive but we consider a run failed if any
-		// tenant runs failed even if later runs are interrupted
-		if !interrupted && !failed {
+		if succeeded && compactionErrorCount == 0 {
 			c.compactionRunsCompleted.Inc()
 			c.compactionRunsLastSuccess.SetToCurrentTime()
-		}
-		if interrupted {
-			c.compactionRunsInterrupted.Inc()
-		}
-		if failed {
+		} else {
 			c.compactionRunsFailed.Inc()
 		}
 
@@ -686,7 +675,6 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 	level.Info(c.logger).Log("msg", "discovering users from bucket")
 	users, err := c.discoverUsersWithRetries(ctx)
 	if err != nil {
-		failed = true
 		level.Error(c.logger).Log("msg", "failed to discover users from bucket", "err", err)
 		return
 	}
@@ -706,8 +694,7 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 	for _, userID := range users {
 		// Ensure the context has not been canceled (ie. compactor shutdown has been triggered).
 		if ctx.Err() != nil {
-			interrupted = true
-			level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "user", userID)
+			level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "err", err)
 			return
 		}
 
@@ -737,15 +724,8 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 		level.Info(c.logger).Log("msg", "starting compaction of user blocks", "user", userID)
 
 		if err = c.compactUserWithRetries(ctx, userID); err != nil {
-			// TODO: patch thanos error types to support errors.Is(err, context.Canceled) here
-			if ctx.Err() != nil && ctx.Err() == context.Canceled {
-				interrupted = true
-				level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "user", userID)
-				return
-			}
-
 			c.compactionRunFailedTenants.Inc()
-			failed = true
+			compactionErrorCount++
 			level.Error(c.logger).Log("msg", "failed to compact user blocks", "user", userID, "err", err)
 			continue
 		}
@@ -780,6 +760,8 @@ func (c *Compactor) compactUsers(ctx context.Context) {
 			}
 		}
 	}
+
+	succeeded = true
 }
 
 func (c *Compactor) compactUserWithRetries(ctx context.Context, userID string) error {
@@ -868,8 +850,8 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 		c.blocksGrouperFactory(ctx, c.compactorCfg, bucket, ulogger, reg, c.blocksMarkedForDeletion, c.blocksMarkedForNoCompaction, c.garbageCollectedBlocks, c.remainingPlannedCompactions, c.blockVisitMarkerReadFailed, c.blockVisitMarkerWriteFailed, c.partitionedGroupInfoReadFailed, c.partitionedGroupInfoWriteFailed, c.ring, c.ringLifecycler, c.limits, userID, noCompactMarkerFilter),
 		c.blocksPlannerFactory(ctx, bucket, ulogger, c.compactorCfg, noCompactMarkerFilter, c.ringLifecycler, c.blockVisitMarkerReadFailed, c.blockVisitMarkerWriteFailed),
 		c.blocksCompactor,
-		c.compactionCompleteCheckerFactory(ctx, bucket, ulogger, c.blockVisitMarkerReadFailed, c.partitionedGroupInfoReadFailed),
-		compact.NoopCompactionLifecycleCallback{},
+		c.blockDeletableCheckerFactory(ctx, bucket, ulogger, c.blockVisitMarkerReadFailed, c.partitionedGroupInfoReadFailed),
+		ShardedCompactionLifecycleCallback{},
 		path.Join(c.compactorCfg.DataDir, "compact"),
 		bucket,
 		c.compactorCfg.CompactionConcurrency,

pkg/compactor/compactor_test.go

@@ -1614,14 +1614,14 @@ func prepare(t *testing.T, compactorCfg Config, bucketClient objstore.Bucket, li
 		}
 	}
 
-	var compactionCompleteCheckerFactory CompactionCompleteCheckerFactory
+	var blockDeletableCheckerFactory BlockDeletableCheckerFactory
 	if compactorCfg.ShardingStrategy == util.ShardingStrategyShuffle {
-		compactionCompleteCheckerFactory = PartitionCompactionCompleteCheckerFactory
+		blockDeletableCheckerFactory = PartitionCompactionBlockDeletableCheckerFactory
 	} else {
-		compactionCompleteCheckerFactory = DefaultCompactionCompleteCheckerFactory
+		blockDeletableCheckerFactory = DefaultBlockDeletableCheckerFactory
 	}
 
-	c, err := newCompactor(compactorCfg, storageCfg, overrides, logger, registry, bucketClientFactory, blocksGrouperFactory, blocksCompactorFactory, compactionCompleteCheckerFactory, overrides)
+	c, err := newCompactor(compactorCfg, storageCfg, overrides, logger, registry, bucketClientFactory, blocksGrouperFactory, blocksCompactorFactory, blockDeletableCheckerFactory, overrides)
 	require.NoError(t, err)
 
 	return c, tsdbCompactor, tsdbPlanner, logs, registry
@@ -1646,8 +1646,8 @@ func (m *tsdbCompactorMock) Compact(dest string, dirs []string, open []*tsdb.Blo
 	return args.Get(0).(ulid.ULID), args.Error(1)
 }
 
-func (m *tsdbCompactorMock) CompactWithAdditionalPostings(dest string, dirs []string, open []*tsdb.Block, additionalPostingsProvider tsdb.AdditionalPostingsFunc) (ulid.ULID, error) {
-	args := m.Called(dest, dirs, open, additionalPostingsProvider)
+func (m *tsdbCompactorMock) CompactWithBlockPopulator(dest string, dirs []string, open []*tsdb.Block, blockPopulator tsdb.BlockPopulator) (ulid.ULID, error) {
+	args := m.Called(dest, dirs, open, blockPopulator)
 	return args.Get(0).(ulid.ULID), args.Error(1)
 }
 

pkg/compactor/partition_compaction_complete_checker.go

@@ -11,22 +11,22 @@ import (
 	"github.com/thanos-io/thanos/pkg/compact"
 )
 
-type PartitionCompactionCompleteChecker struct {
+type PartitionCompactionBlockDeletableChecker struct {
 	ctx                            context.Context
 	bkt                            objstore.InstrumentedBucket
 	logger                         log.Logger
 	blockVisitMarkerReadFailed     prometheus.Counter
 	partitionedGroupInfoReadFailed prometheus.Counter
 }
 
-func NewPartitionCompactionCompleteChecker(
+func NewPartitionCompactionBlockDeletableChecker(
 	ctx context.Context,
 	bkt objstore.InstrumentedBucket,
 	logger log.Logger,
 	blockVisitMarkerReadFailed prometheus.Counter,
 	partitionedGroupInfoReadFailed prometheus.Counter,
-) *PartitionCompactionCompleteChecker {
-	return &PartitionCompactionCompleteChecker{
+) *PartitionCompactionBlockDeletableChecker {
+	return &PartitionCompactionBlockDeletableChecker{
 		ctx:                            ctx,
 		bkt:                            bkt,
 		logger:                         logger,
@@ -35,9 +35,12 @@ func NewPartitionCompactionCompleteChecker(
 	}
 }
 
-func (p *PartitionCompactionCompleteChecker) IsComplete(group *compact.Group, blockID ulid.ULID) bool {
-	partitionedGroupID := group.PartitionedGroupID()
-	currentPartitionID := group.PartitionID()
+func (p *PartitionCompactionBlockDeletableChecker) CanDelete(group *compact.Group, blockID ulid.ULID) bool {
+	if group.PartitionedInfo() == nil {
+		return true
+	}
+	partitionedGroupID := group.PartitionedInfo().PartitionedGroupID
+	currentPartitionID := group.PartitionedInfo().PartitionID
 	partitionedGroupInfo, err := ReadPartitionedGroupInfo(p.ctx, p.bkt, p.logger, partitionedGroupID, p.partitionedGroupInfoReadFailed)
 	if err != nil {
 		level.Warn(p.logger).Log("msg", "unable to read partitioned group info", "partitioned_group_id", partitionedGroupID, "block_id", blockID, "err", err)
@@ -46,7 +49,7 @@ func (p *PartitionCompactionCompleteChecker) IsComplete(group *compact.Group, bl
 	return p.IsPartitionedBlockComplete(partitionedGroupInfo, currentPartitionID, blockID)
 }
 
-func (p *PartitionCompactionCompleteChecker) IsPartitionedBlockComplete(partitionedGroupInfo *PartitionedGroupInfo, currentPartitionID int, blockID ulid.ULID) bool {
+func (p *PartitionCompactionBlockDeletableChecker) IsPartitionedBlockComplete(partitionedGroupInfo *PartitionedGroupInfo, currentPartitionID int, blockID ulid.ULID) bool {
 	partitionedGroupID := partitionedGroupInfo.PartitionedGroupID
 	for _, partitionID := range partitionedGroupInfo.getPartitionIDsByBlock(blockID) {
 		// Skip current partition ID since current one is completed