diff --git a/CHANGELOG.md b/CHANGELOG.md index cf41c2821c..b0804457c4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re - [#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. - [#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 diff --git a/cmd/thanos/tools_bucket.go b/cmd/thanos/tools_bucket.go index 2c8755e4da..d9d612362e 100644 --- a/cmd/thanos/tools_bucket.go +++ b/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, ) }) } diff --git a/docs/components/tools.md b/docs/components/tools.md index 5562344e19..efa9dc475b 100644 --- a/docs/components/tools.md +++ b/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. diff --git a/pkg/replicate/replicator.go b/pkg/replicate/replicator.go index c2893b050a..8d5776110b 100644 --- a/pkg/replicate/replicator.go +++ b/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()) diff --git a/pkg/replicate/scheme.go b/pkg/replicate/scheme.go index f9b1067b8e..05d68991ad 100644 --- a/pkg/replicate/scheme.go +++ b/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) } diff --git a/pkg/replicate/scheme_test.go b/pkg/replicate/scheme_test.go index d58b114a2e..8d88f096f7 100644 --- a/pkg/replicate/scheme_test.go +++ b/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)