feat bucket replicate: added flag to ignore blocks marked for deletion

Signed-off-by: Martin Chodur <m.chodur@seznam.cz>

CHANGELOG.md

@@ -25,6 +25,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#4974](https://github.com/thanos-io/thanos/pull/4974) Store: Support tls_config configuration for connecting with Azure storage.
 - [#4999](https://github.com/thanos-io/thanos/pull/4999) COS: Support `endpoint` configuration for vpc internal endpoint.
 - [#5059](https://github.com/thanos-io/thanos/pull/5059) Compactor: Adding minimum retention flag validation for downsampling retention.
+- [#5117](https://github.com/thanos-io/thanos/pull/5117) Bucket replicate: Added flag `--ignore-marked-for-deletion` to avoid replication of blocks with the deletion mark.
 
 ### Fixed
 

cmd/thanos/tools_bucket.go

@@ -695,6 +695,7 @@ func registerBucketReplicate(app extkingpin.AppClause, objStoreConfig *extflag.P
 	maxTime := model.TimeOrDuration(cmd.Flag("max-time", "End of time range limit to replicate. Thanos Replicate will replicate only metrics, which happened earlier than this value. Option can be a constant time in RFC3339 format or time duration relative to current time, such as -1d or 2h45m. Valid duration units are ms, s, m, h, d, w, y.").
 		Default("9999-12-31T23:59:59Z"))
 	ids := cmd.Flag("id", "Block to be replicated to the destination bucket. IDs will be used to match blocks and other matchers will be ignored. When specified, this command will be run only once after successful replication. Repeated field").Strings()
+	ignoreMarkedForDeletion := cmd.Flag("ignore-marked-for-deletion", "Do not replicate blocks that have deletion mark.").Bool()
 
 	cmd.Setup(func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ <-chan struct{}, _ bool) error {
 		matchers, err := replicate.ParseFlagMatchers(tbc.matcherStrs)
@@ -733,6 +734,7 @@ func registerBucketReplicate(app extkingpin.AppClause, objStoreConfig *extflag.P
 			minTime,
 			maxTime,
 			blockIDs,
+			*ignoreMarkedForDeletion,
 		)
 	})
 }

pkg/replicate/replicator.go

@@ -85,6 +85,7 @@ func RunReplicate(
 	singleRun bool,
 	minTime, maxTime *thanosmodel.TimeOrDurationValue,
 	blockIDs []ulid.ULID,
+	ignoreMarkedForDeletion bool,
 ) error {
 	logger = log.With(logger, "component", "replicate")
 
@@ -185,6 +186,7 @@ func RunReplicate(
 		resolutions,
 		compactions,
 		blockIDs,
+		ignoreMarkedForDeletion,
 	).Filter
 	metrics := newReplicationMetrics(reg)
 	ctx, cancel := context.WithCancel(context.Background())

pkg/replicate/scheme.go

@@ -29,11 +29,12 @@ import (
 
 // BlockFilter is block filter that filters out compacted and unselected blocks.
 type BlockFilter struct {
-	logger           log.Logger
-	labelSelector    labels.Selector
-	resolutionLevels map[compact.ResolutionLevel]struct{}
-	compactionLevels map[int]struct{}
-	blockIDs         []ulid.ULID
+	logger                  log.Logger
+	labelSelector           labels.Selector
+	resolutionLevels        map[compact.ResolutionLevel]struct{}
+	compactionLevels        map[int]struct{}
+	blockIDs                []ulid.ULID
+	ignoreMarkedForDeletion bool
 }
 
 // NewBlockFilter returns block filter.
@@ -43,6 +44,7 @@ func NewBlockFilter(
 	resolutionLevels []compact.ResolutionLevel,
 	compactionLevels []int,
 	blockIDs []ulid.ULID,
+	ignoreMarkedForDeletion bool,
 ) *BlockFilter {
 	allowedResolutions := make(map[compact.ResolutionLevel]struct{})
 	for _, resolutionLevel := range resolutionLevels {
@@ -54,16 +56,20 @@ func NewBlockFilter(
 	}
 
 	return &BlockFilter{
-		labelSelector:    labelSelector,
-		logger:           logger,
-		resolutionLevels: allowedResolutions,
-		compactionLevels: allowedCompactions,
-		blockIDs:         blockIDs,
+		labelSelector:           labelSelector,
+		logger:                  logger,
+		resolutionLevels:        allowedResolutions,
+		compactionLevels:        allowedCompactions,
+		blockIDs:                blockIDs,
+		ignoreMarkedForDeletion: ignoreMarkedForDeletion,
 	}
 }
 
 // Filter return true if block is non-compacted and matches selector.
-func (bf *BlockFilter) Filter(b *metadata.Meta) bool {
+func (bf *BlockFilter) Filter(b *metadata.Meta, markedForDeletion bool) bool {
+	if bf.ignoreMarkedForDeletion && markedForDeletion {
+		return false
+	}
 	if len(b.Thanos.Labels) == 0 {
 		level.Error(bf.logger).Log("msg", "filtering block", "reason", "labels should not be empty")
 		return false
@@ -115,7 +121,7 @@ func (bf *BlockFilter) Filter(b *metadata.Meta) bool {
 	return true
 }
 
-type blockFilterFunc func(b *metadata.Meta) bool
+type blockFilterFunc func(b *metadata.Meta, markedForDeletion bool) bool
 
 // TODO: Add filters field.
 type replicationScheme struct {
@@ -192,7 +198,8 @@ func (rs *replicationScheme) execute(ctx context.Context) error {
 	}
 
 	for id, meta := range metas {
-		if rs.blockFilter(meta) {
+		_, err := rs.fromBkt.ReaderWithExpectedErrs(rs.fromBkt.IsObjNotFoundErr).Get(ctx, path.Join(meta.ULID.String(), metadata.DeletionMarkFilename))
+		if rs.blockFilter(meta, !rs.fromBkt.IsObjNotFoundErr(err)) {
 			level.Info(rs.logger).Log("msg", "adding block to be replicated", "block_uuid", id.String())
 			availableBlocks = append(availableBlocks, meta)
 		}

pkg/replicate/scheme_test.go

@@ -62,6 +62,15 @@ func testMeta(ulid ulid.ULID) *metadata.Meta {
 	}
 }
 
+func testDeletionMark(ulid ulid.ULID) *metadata.DeletionMark {
+	return &metadata.DeletionMark{
+		ID:           ulid,
+		Version:      metadata.DeletionMarkVersion1,
+		Details:      "tests deletion mark",
+		DeletionTime: time.Time{}.Unix(),
+	}
+}
+
 func TestReplicationSchemeAll(t *testing.T) {
 	testBlockID := testULID(0)
 	var cases = []struct {
@@ -116,6 +125,28 @@ func TestReplicationSchemeAll(t *testing.T) {
 				}
 			},
 		},
+		{
+			name: "MarkedForDeletion",
+			prepare: func(ctx context.Context, t *testing.T, originBucket, targetBucket *objstore.InMemBucket) {
+				ulid := testULID(0)
+				meta := testMeta(ulid)
+				deletionMark := testDeletionMark(ulid)
+
+				b, err := json.Marshal(meta)
+				testutil.Ok(t, err)
+				d, err := json.Marshal(deletionMark)
+				testutil.Ok(t, err)
+				_ = originBucket.Upload(ctx, path.Join(ulid.String(), "meta.json"), bytes.NewReader(b))
+				_ = originBucket.Upload(ctx, path.Join(ulid.String(), "deletion-mark.json"), bytes.NewReader(d))
+				_ = originBucket.Upload(ctx, path.Join(ulid.String(), "chunks", "000001"), bytes.NewReader(nil))
+				_ = originBucket.Upload(ctx, path.Join(ulid.String(), "index"), bytes.NewReader(nil))
+			},
+			assert: func(ctx context.Context, t *testing.T, originBucket, targetBucket *objstore.InMemBucket) {
+				if len(targetBucket.Objects()) != 0 {
+					t.Fatal("TargetBucket should have been empty but is not.")
+				}
+			},
+		},
 		{
 			name: "PreviousPartialUpload",
 			prepare: func(ctx context.Context, t *testing.T, originBucket, targetBucket *objstore.InMemBucket) {
@@ -343,7 +374,7 @@ func TestReplicationSchemeAll(t *testing.T) {
 			selector = c.selector
 		}
 
-		filter := NewBlockFilter(logger, selector, []compact.ResolutionLevel{compact.ResolutionLevelRaw}, []int{1}, c.blockIDs).Filter
+		filter := NewBlockFilter(logger, selector, []compact.ResolutionLevel{compact.ResolutionLevelRaw}, []int{1}, c.blockIDs, true).Filter
 		fetcher, err := block.NewMetaFetcher(logger, 32, objstore.WithNoopInstr(originBucket), "", nil, nil)
 		testutil.Ok(t, err)