diff --git a/cmd/thanos/compact.go b/cmd/thanos/compact.go index b0ea2f346cd..3ca5c6463a8 100644 --- a/cmd/thanos/compact.go +++ b/cmd/thanos/compact.go @@ -19,6 +19,7 @@ import ( "github.com/opentracing/opentracing-go" "github.com/pkg/errors" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/prometheus/tsdb" "github.com/thanos-io/thanos/pkg/block" "github.com/thanos-io/thanos/pkg/block/indexheader" @@ -121,6 +122,9 @@ func registerCompact(m map[string]setupFunc, app *kingpin.Application) { compactionConcurrency := cmd.Flag("compact.concurrency", "Number of goroutines to use when compacting groups."). Default("1").Int() + deleteDelay := modelDuration(cmd.Flag("delete-delay", fmt.Sprintf("Time before a block marked for deletion is deleted from bucket")). + Default("48h")) + selectorRelabelConf := regSelectorRelabelFlags(cmd) m[component.Compact.String()] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ <-chan struct{}, _ bool) error { @@ -130,6 +134,7 @@ func registerCompact(m map[string]setupFunc, app *kingpin.Application) { *dataDir, objStoreConfig, time.Duration(*consistencyDelay), + time.Duration(*deleteDelay), *haltOnError, *acceptMalformedIndex, *wait, @@ -158,6 +163,7 @@ func runCompact( dataDir string, objStoreConfig *extflag.PathOrContent, consistencyDelay time.Duration, + deleteDelay time.Duration, haltOnError bool, acceptMalformedIndex bool, wait bool, @@ -187,6 +193,12 @@ func runCompact( Name: "thanos_compactor_aborted_partial_uploads_deletion_attempts_total", Help: "Total number of started deletions of blocks that are assumed aborted and only partially uploaded.", }) + _ = promauto.With(reg).NewGaugeFunc(prometheus.GaugeOpts{ + Name: "thanos_delete_delay_seconds", + Help: "Configured delete delay in seconds.", + }, func() float64 { + return deleteDelay.Seconds() + }) reg.MustRegister(halted, retried, iterations, partialUploadDeleteAttempts) downsampleMetrics := newDownsampleMetrics(reg) @@ -285,6 +297,7 @@ func runCompact( return errors.Wrap(err, "clean working downsample directory") } + blocksCleaner := compact.NewBlocksCleaner(logger, bkt, deleteDelay) compactor, err := compact.NewBucketCompactor(logger, sy, comp, compactDir, bkt, concurrency) if err != nil { cancel() @@ -334,6 +347,22 @@ func runCompact( return nil } + g.Add(func() error { + if !wait { + return blocksCleaner.DeleteMarkedBlocks(ctx) + } + + // --wait=true is specified. + return runutil.Repeat(deleteDelay, ctx.Done(), func() error { + if err := blocksCleaner.DeleteMarkedBlocks(ctx); err != nil { + return errors.Wrap(err, "error cleaning blocks") + } + return nil + }) + }, func(error) { + cancel() + }) + g.Add(func() error { defer runutil.CloseWithLogOnErr(logger, bkt, "bucket client") diff --git a/docs/components/compact.md b/docs/components/compact.md index edd71eb7b11..74136195d4a 100644 --- a/docs/components/compact.md +++ b/docs/components/compact.md @@ -11,8 +11,8 @@ It is generally not semantically concurrency safe and must be deployed as a sing It is also responsible for downsampling of data: -* creating 5m downsampling for blocks larger than **40 hours** (2d, 2w) -* creating 1h downsampling for blocks larger than **10 days** (2w). +- creating 5m downsampling for blocks larger than **40 hours** (2d, 2w) +- creating 1h downsampling for blocks larger than **10 days** (2w). Example: @@ -35,9 +35,9 @@ On-disk data is safe to delete between restarts and should be the first attempt Resolution - distance between data points on your graphs. E.g. -* raw - the same as scrape interval at the moment of data ingestion -* 5m - data point is every 5 minutes -* 1h - data point is every 1h +- raw - the same as scrape interval at the moment of data ingestion +- 5m - data point is every 5 minutes +- 1h - data point is every 1h Keep in mind, that the initial goal of downsampling is not saving disk space (Read further for elaboration on storage space consumption). The goal of downsampling is providing an opportunity to get fast results for range queries of big time intervals like months or years. In other words, if you set `--retention.resolution-raw` less then `--retention.resolution-5m` and `--retention.resolution-1h` - you might run into a problem of not being able to "zoom in" to your historical data. @@ -55,8 +55,8 @@ In fact, downsampling doesn't save you any space but instead it adds 2 more bloc ## Groups -The compactor groups blocks using the external_labels added by the Prometheus who produced the block. -The labels must be both _unique_ and _persistent_ across different Prometheus instances. +The compactor groups blocks using the external*labels added by the Prometheus who produced the block. +The labels must be both \_unique* and _persistent_ across different Prometheus instances. By _unique_, we mean that the set of labels in a Prometheus instance must be different from all other sets of labels of your Prometheus instances, so that the compactor will be able to group blocks by Prometheus instance. @@ -66,7 +66,8 @@ compacting blocks from an instance even when a Prometheus instance goes down for ## Flags -[embedmd]:# (flags/compact.txt $) +[embedmd]: # "flags/compact.txt $" + ```$ usage: thanos compact [] @@ -144,5 +145,5 @@ Flags: selecting blocks. It follows native Prometheus relabel-config syntax. See format details: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config - + --delete-delay=48h Time before a block marked for deletion is deleted from bucket. ``` diff --git a/pkg/block/block.go b/pkg/block/block.go index 740b3d63060..197884deef6 100644 --- a/pkg/block/block.go +++ b/pkg/block/block.go @@ -6,6 +6,7 @@ package block import ( + "bytes" "context" "encoding/json" "fmt" @@ -14,6 +15,7 @@ import ( "path" "path/filepath" "strings" + "time" "github.com/go-kit/kit/log" "github.com/go-kit/kit/log/level" @@ -126,6 +128,32 @@ func cleanUp(logger log.Logger, bkt objstore.Bucket, id ulid.ULID, err error) er return err } +// 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) error { + deletionMarkExists, err := objstore.Exists(ctx, bkt, path.Join(id.String(), metadata.DeletionMarkFilename)) + if err != nil { + return errors.Wrapf(err, "check compactor meta for id %s in bucket", id.String()) + } + if deletionMarkExists { + return nil + } + + deletionMark, err := json.Marshal(metadata.DeletionMark{ + ID: id, + DeletionTime: time.Now().Unix(), + }) + if err != nil { + return errors.Wrap(err, "compactor meta json marshalling") + } + + if err := bkt.Upload(ctx, path.Join(id.String(), metadata.DeletionMarkFilename), bytes.NewReader(deletionMark)); err != nil { + return errors.Wrap(err, "upload compactor-meta file to bucket") + } + + level.Info(logger).Log("msg", "block", id, "has been marked for deletion") + return nil +} + // Delete removes directory that is meant to be block directory. // NOTE: Always prefer this method for deleting blocks. // * We have to delete block's files in the certain order (meta.json first) diff --git a/pkg/block/metadata/deletionmark.go b/pkg/block/metadata/deletionmark.go new file mode 100644 index 00000000000..d4112006a21 --- /dev/null +++ b/pkg/block/metadata/deletionmark.go @@ -0,0 +1,71 @@ +// Copyright (c) The Thanos Authors. +// Licensed under the Apache License 2.0. + +package metadata + +import ( + "context" + "encoding/json" + "io/ioutil" + "path" + + "github.com/go-kit/kit/log" + "github.com/oklog/ulid" + "github.com/pkg/errors" + "github.com/thanos-io/thanos/pkg/objstore" + "github.com/thanos-io/thanos/pkg/runutil" +) + +const ( + // DeletionMarkFilename is the known json filename to store details about when block is marked for deletion. + DeletionMarkFilename = "deletion-mark.json" + + // DeletionMarkVersion1 is a enumeration of deletion-mark versions supported by Thanos. + DeletionMarkVersion1 = iota + 1 +) + +// ErrorDeletionMarkNotFound is the error when deletion-mark.json file is not found. +var ErrorDeletionMarkNotFound = errors.New("deletion-mark.json not found") + +// DeletionMark stores block id and when block was marked for deletion. +type DeletionMark struct { + // ID of the tsdb block. + ID ulid.ULID `json:"id"` + + // DeletionTime is a unix timestamp of when the block was marked to be deleted. + DeletionTime int64 `json:"deletion_time"` + + // Version of the file. + Version int `json:"version"` +} + +// ReadDeletionMark reads the given compactor meta from /deletion-mark.json in bucket. +func ReadDeletionMark(ctx context.Context, bkt objstore.Bucket, logger log.Logger, dir string) (*DeletionMark, error) { + deletionMarkFile := path.Join(dir, DeletionMarkFilename) + + r, err := bkt.Get(ctx, deletionMarkFile) + if bkt.IsObjNotFoundErr(err) { + return nil, ErrorDeletionMarkNotFound + } + if err != nil { + return nil, errors.Wrapf(err, "get file: %v", deletionMarkFile) + } + + defer runutil.CloseWithLogOnErr(logger, r, "close bkt deletion-mark reader") + + metaContent, err := ioutil.ReadAll(r) + if err != nil { + return nil, errors.Wrapf(err, "read file: %v", deletionMarkFile) + } + + deletionMark := DeletionMark{} + if err := json.Unmarshal(metaContent, &deletionMark); err != nil { + return nil, errors.Wrap(err, "unmarshal compactor meta") + } + + if deletionMark.Version != DeletionMarkVersion1 { + return nil, errors.Errorf("unexpected deletion-mark file version %d", deletionMark.Version) + } + + return &deletionMark, nil +} diff --git a/pkg/compact/blocks_cleaner.go b/pkg/compact/blocks_cleaner.go new file mode 100644 index 00000000000..4d09a3312d2 --- /dev/null +++ b/pkg/compact/blocks_cleaner.go @@ -0,0 +1,60 @@ +// Copyright (c) The Thanos Authors. +// Licensed under the Apache License 2.0. + +package compact + +import ( + "context" + "time" + + "github.com/go-kit/kit/log" + "github.com/go-kit/kit/log/level" + "github.com/pkg/errors" + "github.com/thanos-io/thanos/pkg/block" + "github.com/thanos-io/thanos/pkg/block/metadata" + "github.com/thanos-io/thanos/pkg/objstore" +) + +// BlocksCleaner is a struct that deletes blocks from bucket +// which are marked for deletion. +type BlocksCleaner struct { + logger log.Logger + deleteDelay time.Duration + bkt objstore.Bucket +} + +// NewBlocksCleaner creates a new BlocksCleaner. +func NewBlocksCleaner(logger log.Logger, bkt objstore.Bucket, deleteDelay time.Duration) *BlocksCleaner { + return &BlocksCleaner{ + logger: logger, + deleteDelay: deleteDelay, + bkt: bkt, + } +} + +// DeleteMarkedBlocks reads compactor-meta.json file in block to check when the block was marked for deletion and +// deletes the block deleteDelay duration after block is marked for deletion. +func (s *BlocksCleaner) DeleteMarkedBlocks(ctx context.Context) error { + return s.bkt.Iter(ctx, "", func(name string) error { + id, ok := block.IsBlockDir(name) + if !ok { + return nil + } + + deletionMark, err := metadata.ReadDeletionMark(ctx, s.bkt, s.logger, id.String()) + if err != nil { + if errors.Cause(err) != metadata.ErrorDeletionMarkNotFound { + return errors.Wrap(err, "read file: %s") + } + } + + if time.Since(time.Unix(deletionMark.DeletionTime, 0)) > s.deleteDelay { + if err := block.Delete(ctx, s.logger, s.bkt, deletionMark.ID); err != nil { + return errors.Wrap(err, "delete block") + } + level.Info(s.logger).Log("msg", "blocks cleaner: deleting block", "id", id) + } + + return nil + }) +} diff --git a/pkg/compact/clean.go b/pkg/compact/clean.go index af681a940e4..215b951854f 100644 --- a/pkg/compact/clean.go +++ b/pkg/compact/clean.go @@ -41,7 +41,7 @@ func BestEffortCleanAbortedPartialUploads(ctx context.Context, logger log.Logger } deleteAttempts.Inc() - if err := block.Delete(ctx, logger, bkt, id); err != nil { + if err := block.MarkForDeletion(ctx, logger, bkt, id); err != nil { level.Warn(logger).Log("msg", "failed to delete aborted partial upload; skipping", "block", id, "thresholdAge", PartialUploadThresholdAge, "err", err) return } diff --git a/pkg/compact/clean_test.go b/pkg/compact/clean_test.go index 85654f8c8a3..8239d28a3d9 100644 --- a/pkg/compact/clean_test.go +++ b/pkg/compact/clean_test.go @@ -63,7 +63,11 @@ func TestBestEffortCleanAbortedPartialUploads(t *testing.T) { exists, err := bkt.Exists(ctx, path.Join(shouldDeleteID.String(), "chunks", "000001")) testutil.Ok(t, err) - testutil.Equals(t, false, exists) + testutil.Equals(t, true, exists) + + exists, err = bkt.Exists(ctx, path.Join(shouldDeleteID.String(), metadata.DeletionMarkFilename)) + testutil.Ok(t, err) + testutil.Equals(t, true, exists) exists, err = bkt.Exists(ctx, path.Join(shouldIgnoreID1.String(), "chunks", "000001")) testutil.Ok(t, err) diff --git a/pkg/compact/compact.go b/pkg/compact/compact.go index b268724e108..1ccde9bb964 100644 --- a/pkg/compact/compact.go +++ b/pkg/compact/compact.go @@ -243,9 +243,9 @@ func (s *Syncer) GarbageCollect(ctx context.Context) error { // Spawn a new context so we always delete a block in full on shutdown. delCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) - level.Info(s.logger).Log("msg", "deleting outdated block", "block", id) + level.Info(s.logger).Log("msg", "marking outdated block for deletion", "block", id) - err := block.Delete(delCtx, s.logger, s.bkt, id) + err := block.MarkForDeletion(delCtx, s.logger, s.bkt, id) cancel() if err != nil { s.metrics.garbageCollectionFailures.Inc() @@ -559,7 +559,7 @@ func RepairIssue347(ctx context.Context, logger log.Logger, bkt objstore.Bucket, defer cancel() // TODO(bplotka): Issue with this will introduce overlap that will halt compactor. Automate that (fix duplicate overlaps caused by this). - if err := block.Delete(delCtx, logger, bkt, ie.id); err != nil { + if err := block.MarkForDeletion(delCtx, logger, bkt, ie.id); err != nil { return errors.Wrapf(err, "deleting old block %s failed. You need to delete this block manually", ie.id) } @@ -760,8 +760,8 @@ func (cg *Group) deleteBlock(b string) error { // Spawn a new context so we always delete a block in full on shutdown. delCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) defer cancel() - level.Info(cg.logger).Log("msg", "deleting compacted block", "old_block", id) - if err := block.Delete(delCtx, cg.logger, cg.bkt, id); err != nil { + level.Info(cg.logger).Log("msg", "marking compacted block for deletion", "old_block", id) + if err := block.MarkForDeletion(delCtx, cg.logger, cg.bkt, id); err != nil { return errors.Wrapf(err, "delete block %s from bucket", id) } return nil diff --git a/pkg/compact/compact_e2e_test.go b/pkg/compact/compact_e2e_test.go index 43ce2974341..f88133cb4b4 100644 --- a/pkg/compact/compact_e2e_test.go +++ b/pkg/compact/compact_e2e_test.go @@ -103,7 +103,16 @@ func TestSyncer_GarbageCollect_e2e(t *testing.T) { var rem []ulid.ULID err = bkt.Iter(ctx, "", func(n string) error { - rem = append(rem, ulid.MustParse(n[:len(n)-1])) + id := ulid.MustParse(n[:len(n)-1]) + deletionMarkFile := path.Join(id.String(), metadata.DeletionMarkFilename) + + exists, err := bkt.Exists(ctx, deletionMarkFile) + if err != nil { + return err + } + if !exists { + rem = append(rem, id) + } return nil }) testutil.Ok(t, err) diff --git a/pkg/compact/retention.go b/pkg/compact/retention.go index 47f61f0cfbf..4fa77de19a0 100644 --- a/pkg/compact/retention.go +++ b/pkg/compact/retention.go @@ -31,8 +31,8 @@ func ApplyRetentionPolicyByResolution(ctx context.Context, logger log.Logger, bk maxTime := time.Unix(m.MaxTime/1000, 0) if time.Now().After(maxTime.Add(retentionDuration)) { - level.Info(logger).Log("msg", "applying retention: deleting block", "id", id, "maxTime", maxTime.String()) - if err := block.Delete(ctx, logger, bkt, id); err != nil { + level.Info(logger).Log("msg", "applying retention: marking block for deletion", "id", id, "maxTime", maxTime.String()) + if err := block.MarkForDeletion(ctx, logger, bkt, id); err != nil { return errors.Wrap(err, "delete block") } } diff --git a/pkg/compact/retention_test.go b/pkg/compact/retention_test.go index 371d3c4152a..dab35e69812 100644 --- a/pkg/compact/retention_test.go +++ b/pkg/compact/retention_test.go @@ -7,6 +7,7 @@ import ( "bytes" "context" "encoding/json" + "path/filepath" "strings" "testing" "time" @@ -250,7 +251,13 @@ func TestApplyRetentionPolicyByResolution(t *testing.T) { got := []string{} testutil.Ok(t, bkt.Iter(context.TODO(), "", func(name string) error { - got = append(got, name) + exists, err := bkt.Exists(ctx, filepath.Join(name, metadata.DeletionMarkFilename)) + if err != nil { + return err + } + if !exists { + got = append(got, name) + } return nil }))