compact: Added support for no-compact markers in planner.

The planner algo was adapted to avoid unnecessary changes to
blocks caused by excluded blocks, so we can quickly switch to
different planning logic in next iteration.

Fixes: #1424

Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>

cmd/thanos/compact.go

@@ -220,6 +220,7 @@ func runCompact(
 		component,
 	)
 	api := blocksAPI.NewBlocksAPI(logger, conf.label, flagsMap)
+	noCompactMarkerFilter := compact.NewGatherNoCompactionMarkFilter(logger, bkt)
 	var sy *compact.Syncer
 	{
 		// Make sure all compactor meta syncs are done through Syncer.SyncMeta for readability.
@@ -229,6 +230,7 @@ func runCompact(
 				block.NewConsistencyDelayMetaFilter(logger, conf.consistencyDelay, extprom.WrapRegistererWithPrefix("thanos_", reg)),
 				ignoreDeletionMarkFilter,
 				duplicateBlocksFilter,
+				noCompactMarkerFilter,
 			}, []block.MetadataModifier{block.NewReplicaLabelRemover(logger, conf.dedupReplicaLabels)},
 		)
 		cf.UpdateOnChange(func(blocks []metadata.Meta, err error) {
@@ -280,7 +282,7 @@ func runCompact(
 
 	grouper := compact.NewDefaultGrouper(logger, bkt, conf.acceptMalformedIndex, enableVerticalCompaction, reg, blocksMarkedForDeletion, garbageCollectedBlocks)
 	blocksCleaner := compact.NewBlocksCleaner(logger, bkt, ignoreDeletionMarkFilter, deleteDelay, blocksCleaned, blockCleanupFailures)
-	compactor, err := compact.NewBucketCompactor(logger, sy, grouper, compact.NewTSDBBasedPlanner(logger, levels), comp, compactDir, bkt, conf.compactionConcurrency)
+	compactor, err := compact.NewBucketCompactor(logger, sy, grouper, compact.NewTSDBBasedPlanner(logger, levels, noCompactMarkerFilter), comp, compactDir, bkt, conf.compactionConcurrency)
 	if err != nil {
 		cancel()
 		return errors.Wrap(err, "create bucket compactor")

pkg/block/block.go

@@ -144,7 +144,7 @@ func MarkForDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket
 		return nil
 	}
 
-	deletionMark, err := json.Marshal(metadata.DeletionMark{
+	deletionMark, err := json.Marshal(metadata.DeletionMarker{
 		ID:           id,
 		DeletionTime: time.Now().Unix(),
 		Version:      metadata.DeletionMarkVersion1,

pkg/block/block_test.go

@@ -278,7 +278,7 @@ func TestMarkForDeletion(t *testing.T) {
 		{
 			name: "block with deletion mark already, expected log and no metric increment",
 			preUpload: func(t testing.TB, id ulid.ULID, bkt objstore.Bucket) {
-				deletionMark, err := json.Marshal(metadata.DeletionMark{
+				deletionMark, err := json.Marshal(metadata.DeletionMarker{
 					ID:           id,
 					DeletionTime: time.Now().Unix(),
 					Version:      metadata.DeletionMarkVersion1,

pkg/block/fetcher.go

@@ -71,6 +71,9 @@ const (
 	// but don't have a replacement block yet.
 	markedForDeletionMeta = "marked-for-deletion"
 
+	// MarkedForNoCompactionMeta is label for blocks which are loaded but also marked for no compaction. This label is also counted in `loaded` label metric.
+	MarkedForNoCompactionMeta = "marked-for-no-compact"
+
 	// Modified label values.
 	replicaRemovedMeta = "replica-label-removed"
 )
@@ -111,6 +114,7 @@ func newFetcherMetrics(reg prometheus.Registerer) *fetcherMetrics {
 		[]string{timeExcludedMeta},
 		[]string{duplicateMeta},
 		[]string{markedForDeletionMeta},
+		[]string{MarkedForNoCompactionMeta},
 	)
 	m.modified = extprom.NewTxGaugeVec(
 		reg,
@@ -759,7 +763,7 @@ type IgnoreDeletionMarkFilter struct {
 	logger          log.Logger
 	delay           time.Duration
 	bkt             objstore.InstrumentedBucketReader
-	deletionMarkMap map[ulid.ULID]*metadata.DeletionMark
+	deletionMarkMap map[ulid.ULID]*metadata.DeletionMarker
 }
 
 // NewIgnoreDeletionMarkFilter creates IgnoreDeletionMarkFilter.
@@ -772,29 +776,30 @@ func NewIgnoreDeletionMarkFilter(logger log.Logger, bkt objstore.InstrumentedBuc
 }
 
 // DeletionMarkBlocks returns block ids that were marked for deletion.
-func (f *IgnoreDeletionMarkFilter) DeletionMarkBlocks() map[ulid.ULID]*metadata.DeletionMark {
+func (f *IgnoreDeletionMarkFilter) DeletionMarkBlocks() map[ulid.ULID]*metadata.DeletionMarker {
 	return f.deletionMarkMap
 }
 
 // Filter filters out blocks that are marked for deletion after a given delay.
 // It also returns the blocks that can be deleted since they were uploaded delay duration before current time.
 func (f *IgnoreDeletionMarkFilter) Filter(ctx context.Context, metas map[ulid.ULID]*metadata.Meta, synced *extprom.TxGaugeVec) error {
-	f.deletionMarkMap = make(map[ulid.ULID]*metadata.DeletionMark)
+	f.deletionMarkMap = make(map[ulid.ULID]*metadata.DeletionMarker)
 
 	for id := range metas {
-		deletionMark, err := metadata.ReadDeletionMark(ctx, f.bkt, f.logger, id.String())
-		if err == metadata.ErrorDeletionMarkNotFound {
-			continue
-		}
-		if errors.Cause(err) == metadata.ErrorUnmarshalDeletionMark {
-			level.Warn(f.logger).Log("msg", "found partial deletion-mark.json; if we will see it happening often for the same block, consider manually deleting deletion-mark.json from the object storage", "block", id, "err", err)
-			continue
-		}
-		if err != nil {
+		m := &metadata.DeletionMarker{}
+		if err := metadata.ReadMarker(ctx, f.logger, f.bkt, id.String(), m); err != nil {
+			if errors.Cause(err) == metadata.ErrorMarkerNotFound {
+				continue
+			}
+			if errors.Cause(err) == metadata.ErrorUnmarshalMarker {
+				level.Warn(f.logger).Log("msg", "found partial deletion-mark.json; if we will see it happening often for the same block, consider manually deleting deletion-mark.json from the object storage", "block", id, "err", err)
+				continue
+			}
 			return err
 		}
-		f.deletionMarkMap[id] = deletionMark
-		if time.Since(time.Unix(deletionMark.DeletionTime, 0)).Seconds() > f.delay.Seconds() {
+
+		f.deletionMarkMap[id] = m
+		if time.Since(time.Unix(m.DeletionTime, 0)).Seconds() > f.delay.Seconds() {
 			synced.WithLabelValues(markedForDeletionMeta).Inc()
 			delete(metas, id)
 		}

pkg/block/fetcher_test.go

@@ -1084,13 +1084,13 @@ func TestIgnoreDeletionMarkFilter_Filter(t *testing.T) {
 			delay:  48 * time.Hour,
 		}
 
-		shouldFetch := &metadata.DeletionMark{
+		shouldFetch := &metadata.DeletionMarker{
 			ID:           ULID(1),
 			DeletionTime: now.Add(-15 * time.Hour).Unix(),
 			Version:      1,
 		}
 
-		shouldIgnore := &metadata.DeletionMark{
+		shouldIgnore := &metadata.DeletionMarker{
 			ID:           ULID(2),
 			DeletionTime: now.Add(-60 * time.Hour).Unix(),
 			Version:      1,