Add posibility to delete blocks older than max time

Signed-off-by: Kevin Hellemun <17928966+OGKevin@users.noreply.github.com>

cmd/thanos/tools_bucket.go

@@ -444,6 +444,7 @@ func registerBucketReplicate(m map[string]setupFunc, root *kingpin.CmdClause, na
 		Default("0000-01-01T00:00:00Z"))
 	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"))
+	deleteOldBlocks := cmd.Flag("delete-old-blocks", "Delete blocks that are older then max-time.").Default("false").Bool()
 
 	m[name+" replicate"] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ <-chan struct{}, _ bool) error {
 		matchers, err := replicate.ParseFlagMatchers(*matcherStrs)
@@ -471,6 +472,7 @@ func registerBucketReplicate(m map[string]setupFunc, root *kingpin.CmdClause, na
 			*singleRun,
 			minTime,
 			maxTime,
+			*deleteOldBlocks,
 		)
 	}
 

docs/components/tools.md

@@ -473,6 +473,7 @@ Flags:
                                  duration relative to current time, such as -1d
                                  or 2h45m. Valid duration units are ms, s, m, h,
                                  d, w, y.
+      --delete-old-blocks        Delete blocks that are older then max-time.
 
 ```
 

pkg/block/block.go

@@ -123,6 +123,20 @@ func cleanUp(logger log.Logger, bkt objstore.Bucket, id ulid.ULID, err error) er
 
 // MarkForDeletion creates a file which stores information about when the block was marked for deletion.
 func MarkForDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid.ULID, markedForDeletion prometheus.Counter) error {
+	return markForDeletion(ctx, logger, bkt, id, time.Now(), markedForDeletion)
+}
+
+// MarkForFutureDeletion creates a file which stores information about when the block should be deleted in the future.
+func MarkForFutureDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid.ULID, futureDeletionTime time.Time, markedForDeletion prometheus.Counter) error {
+	if time.Now().Before(futureDeletionTime) {
+		return errors.New(fmt.Sprintf("deletion time %s is not in the future", futureDeletionTime.Format(time.RFC3339)))
+	}
+
+	return markForDeletion(ctx, logger, bkt, id, futureDeletionTime, markedForDeletion)
+}
+
+// MarkForDeletion creates a file which stores information about when the block should be marked for deletion.
+func markForDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket, id ulid.ULID, deletionTime time.Time, markedForDeletion prometheus.Counter) error {
 	deletionMarkFile := path.Join(id.String(), metadata.DeletionMarkFilename)
 	deletionMarkExists, err := bkt.Exists(ctx, deletionMarkFile)
 	if err != nil {
@@ -135,7 +149,7 @@ func MarkForDeletion(ctx context.Context, logger log.Logger, bkt objstore.Bucket
 
 	deletionMark, err := json.Marshal(metadata.DeletionMark{
 		ID:           id,
-		DeletionTime: time.Now().Unix(),
+		DeletionTime: deletionTime.Unix(),
 		Version:      metadata.DeletionMarkVersion1,
 	})
 	if err != nil {

pkg/block/metadata/deletionmark.go

@@ -32,7 +32,7 @@ var ErrorDeletionMarkNotFound = errors.New("deletion-mark.json not found")
 // or the deletion-mark.json file is not a valid json file.
 var ErrorUnmarshalDeletionMark = errors.New("unmarshal deletion-mark.json")
 
-// DeletionMark stores block id and when block was marked for deletion.
+// DeletionMark stores block id| and when block was marked for deletion.
 type DeletionMark struct {
 	// ID of the tsdb block.
 	ID ulid.ULID `json:"id"`

pkg/replicate/replicator.go

@@ -83,6 +83,7 @@ func RunReplicate(
 	toObjStoreConfig *extflag.PathOrContent,
 	singleRun bool,
 	maxTime, minTime *thanosmodel.TimeOrDurationValue,
+	deleteOldBlocks bool,
 ) error {
 	logger = log.With(logger, "component", "replicate")
 
@@ -163,6 +164,14 @@ func RunReplicate(
 	}, []string{"result"})
 	replicationRunDuration.WithLabelValues(labelSuccess)
 	replicationRunDuration.WithLabelValues(labelError)
+	blocksCleaned := promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name: "thanos_replicate_blocks_cleaned_total",
+		Help: "Total number of blocks deleted in replicator.",
+	})
+	blockCleanupFailures := promauto.With(reg).NewCounter(prometheus.CounterOpts{
+		Name: "thanos_replicate_block_cleanup_failures_total",
+		Help: "Failures encountered while deleting blocks in replicator.",
+	})
 
 	fetcher, err := thanosblock.NewMetaFetcher(
 		logger,
@@ -186,6 +195,8 @@ func RunReplicate(
 	metrics := newReplicationMetrics(reg)
 	ctx, cancel := context.WithCancel(context.Background())
 
+	blocksCleaner := compact.NewBlocksCleaner(logger, toBkt, thanosblock.NewIgnoreDeletionMarkFilter(logger, toBkt, time.Hour), time.Hour, blocksCleaned, blockCleanupFailures)
+
 	replicateFn := func() error {
 		timestamp := time.Now()
 		entropy := ulid.Monotonic(rand.New(rand.NewSource(timestamp.UnixNano())), 0)
@@ -198,10 +209,16 @@ func RunReplicate(
 		logger := log.With(logger, "replication-run-id", ulid.String())
 		level.Info(logger).Log("msg", "running replication attempt")
 
-		if err := newReplicationScheme(logger, metrics, blockFilter, fetcher, fromBkt, toBkt, reg).execute(ctx); err != nil {
+		if err := newReplicationScheme(logger, metrics, blockFilter, fetcher, fromBkt, toBkt, reg, maxTime, deleteOldBlocks).execute(ctx); err != nil {
 			return errors.Wrap(err, "replication execute")
 		}
 
+		if deleteOldBlocks {
+			if err := blocksCleaner.DeleteMarkedBlocks(ctx); err != nil {
+				return errors.Wrap(err, "failed to delete old blocks")
+			}
+		}
+
 		return nil
 	}
 

pkg/replicate/scheme.go

@@ -11,6 +11,9 @@ import (
 	"io/ioutil"
 	"path"
 	"sort"
+	"time"
+
+	"github.com/thanos-io/thanos/pkg/model"
 
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
@@ -114,6 +117,9 @@ type replicationScheme struct {
 	metrics *replicationMetrics
 
 	reg prometheus.Registerer
+
+	maxTime                     *model.TimeOrDurationValue
+	markBlocksForFutureDeletion bool
 }
 
 type replicationMetrics struct {
@@ -124,6 +130,8 @@ type replicationMetrics struct {
 	blocksAlreadyReplicated prometheus.Counter
 	blocksReplicated        prometheus.Counter
 	objectsReplicated       prometheus.Counter
+
+	blocksMarkedForDeletion prometheus.Counter
 }
 
 func newReplicationMetrics(reg prometheus.Registerer) *replicationMetrics {
@@ -152,31 +160,29 @@ func newReplicationMetrics(reg prometheus.Registerer) *replicationMetrics {
 			Name: "thanos_replicate_objects_replicated_total",
 			Help: "Total number of objects replicated.",
 		}),
+		blocksMarkedForDeletion: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name: "thanos_replicate _blocks_marked_for_deletion_total",
+			Help: "Total number of blocks marked for deletion in compactor.",
+		}),
 	}
 	return m
 }
 
-func newReplicationScheme(
-	logger log.Logger,
-	metrics *replicationMetrics,
-	blockFilter blockFilterFunc,
-	fetcher thanosblock.MetadataFetcher,
-	from objstore.InstrumentedBucketReader,
-	to objstore.Bucket,
-	reg prometheus.Registerer,
-) *replicationScheme {
+func newReplicationScheme(logger log.Logger, metrics *replicationMetrics, blockFilter blockFilterFunc, fetcher thanosblock.MetadataFetcher, from objstore.InstrumentedBucketReader, to objstore.Bucket, reg prometheus.Registerer, maxTime *model.TimeOrDurationValue, markFoFutureDeletion bool) *replicationScheme {
 	if logger == nil {
 		logger = log.NewNopLogger()
 	}
 
 	return &replicationScheme{
-		logger:      logger,
-		blockFilter: blockFilter,
-		fetcher:     fetcher,
-		fromBkt:     from,
-		toBkt:       to,
-		metrics:     metrics,
-		reg:         reg,
+		logger:                      logger,
+		blockFilter:                 blockFilter,
+		fetcher:                     fetcher,
+		fromBkt:                     from,
+		toBkt:                       to,
+		metrics:                     metrics,
+		reg:                         reg,
+		maxTime:                     maxTime,
+		markBlocksForFutureDeletion: markFoFutureDeletion,
 	}
 }
 
@@ -231,7 +237,7 @@ func (rs *replicationScheme) execute(ctx context.Context) error {
 	})
 
 	for _, b := range availableBlocks {
-		if err := rs.ensureBlockIsReplicated(ctx, b.BlockMeta.ULID); err != nil {
+		if err := rs.ensureBlockIsReplicated(ctx, b); err != nil {
 			return errors.Wrapf(err, "ensure block %v is replicated", b.BlockMeta.ULID.String())
 		}
 	}
@@ -241,8 +247,8 @@ func (rs *replicationScheme) execute(ctx context.Context) error {
 
 // ensureBlockIsReplicated ensures that a block present in the origin bucket is
 // present in the target bucket.
-func (rs *replicationScheme) ensureBlockIsReplicated(ctx context.Context, id ulid.ULID) error {
-	blockID := id.String()
+func (rs *replicationScheme) ensureBlockIsReplicated(ctx context.Context, meta *metadata.Meta) error {
+	blockID := meta.ULID.String()
 	chunksDir := path.Join(blockID, thanosblock.ChunksDirname)
 	indexFile := path.Join(blockID, thanosblock.IndexFilename)
 	metaFile := path.Join(blockID, thanosblock.MetaFilename)
@@ -281,7 +287,7 @@ func (rs *replicationScheme) ensureBlockIsReplicated(ctx context.Context, id uli
 			// If the origin meta file content and target meta file content is
 			// equal, we know we have already successfully replicated
 			// previously.
-			level.Debug(rs.logger).Log("msg", "skipping block as already replicated", "block_uuid", id.String())
+			level.Debug(rs.logger).Log("msg", "skipping block as already replicated", "block_uuid", meta.ULID.String())
 			rs.metrics.blocksAlreadyReplicated.Inc()
 
 			return nil
@@ -309,6 +315,14 @@ func (rs *replicationScheme) ensureBlockIsReplicated(ctx context.Context, id uli
 		return errors.Wrap(err, "upload meta file")
 	}
 
+	if rs.markBlocksForFutureDeletion {
+		deletionTime := time.Unix(meta.MaxTime/1000, 0).Add(time.Duration(*rs.maxTime.Dur))
+		if err := thanosblock.MarkForFutureDeletion(ctx, rs.logger, rs.toBkt, meta.ULID, deletionTime, nil); err != nil {
+			return errors.Wrap(err, "failed to mark block for future deletion")
+		}
+		rs.metrics.blocksMarkedForDeletion.Inc()
+	}
+
 	rs.metrics.blocksReplicated.Inc()
 
 	return nil

pkg/replicate/scheme_test.go

@@ -315,15 +315,7 @@ func TestReplicationSchemeAll(t *testing.T) {
 		fetcher, err := block.NewMetaFetcher(logger, 32, objstore.WithNoopInstr(originBucket), "", nil, nil, nil)
 		testutil.Ok(t, err)
 
-		r := newReplicationScheme(
-			logger,
-			newReplicationMetrics(nil),
-			filter,
-			fetcher,
-			objstore.WithNoopInstr(originBucket),
-			targetBucket,
-			nil,
-		)
+		r := newReplicationScheme(logger, newReplicationMetrics(nil), filter, fetcher, objstore.WithNoopInstr(originBucket), targetBucket, nil, nil, false)
 
 		err = r.execute(ctx)
 		testutil.Ok(t, err)