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

thanos-io#5117)

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

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

* cr: fix docs and test

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

* cr: add debug log on failed deletion mark read

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

* cr: return error instead of logging issues with reading deletion marks

Signed-off-by: Martin Chodur <m.chodur@seznam.cz>
Signed-off-by: Nicholaswang <wzhever@gmail.com>

CHANGELOG.md

@@ -27,6 +27,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#5059](https://github.com/thanos-io/thanos/pull/5059) Compactor: Adding minimum retention flag validation for downsampling retention.
 - [#4667](https://github.com/thanos-io/thanos/pull/4667) Add a pure aws-sdk auth for s3 storage.
 - [#5111](https://github.com/thanos-io/thanos/pull/5111) Add matcher support to Query Rules endpoint.
+- [#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,
  )
  })
 }

docs/components/tools.md

@@ -522,6 +522,9 @@ Flags:
  other matchers will be ignored. When specified,
  this command will be run only once after
  successful replication. Repeated field
+ --ignore-marked-for-deletion
+ Do not replicate blocks that have deletion
+ mark.
  --log.format=logfmt Log format to use. Possible options: logfmt or
  json.
  --log.level=info Log filtering level.

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,11 @@ 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 err != nil && !rs.fromBkt.IsObjNotFoundErr(err) {
+ return errors.Wrapf(err, "failed to read deletion mark from origin bucket block %s", meta.ULID.String())
+ }
+ 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,14 +62,24 @@ 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 {
- name string
- selector labels.Selector
- blockIDs []ulid.ULID
- prepare func(ctx context.Context, t *testing.T, originBucket, targetBucket *objstore.InMemBucket)
- assert func(ctx context.Context, t *testing.T, originBucket, targetBucket *objstore.InMemBucket)
+ name string
+ selector labels.Selector
+ blockIDs []ulid.ULID
+ ignoreMarkedForDeletion bool
+ prepare func(ctx context.Context, t *testing.T, originBucket, targetBucket *objstore.InMemBucket)
+ assert func(ctx context.Context, t *testing.T, originBucket, targetBucket *objstore.InMemBucket)
  }{
  {
  name: "EmptyOrigin",
@@ -116,6 +126,27 @@ func TestReplicationSchemeAll(t *testing.T) {
  }
  },
  },
+ {
+ name: "MarkedForDeletion",
+ ignoreMarkedForDeletion: true,
+ 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) {
+ testutil.Equals(t, map[string][]byte{}, targetBucket.Objects())
+ },
+ },
  {
  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, c.ignoreMarkedForDeletion).Filter
  fetcher, err := block.NewMetaFetcher(logger, 32, objstore.WithNoopInstr(originBucket), "", nil, nil)
  testutil.Ok(t, err)