Per-tenant retention periods

CHANGELOG.md

@@ -7,6 +7,7 @@
 * [CHANGE] Query-frontend: removed `-querier.split-queries-by-day` (deprecated in Cortex 0.4.0). You should use `-querier.split-queries-by-interval` instead. #3813
 * [CHANGE] Store-gateway: the chunks pool controlled by `-blocks-storage.bucket-store.max-chunk-pool-bytes` is now shared across all tenants. #3830
 * [CHANGE] Ingester: return error code 400 instead of 429 when per-user/per-tenant series/metadata limits are reached. #3833
+* [CHANGE] Compactor: add `reason` label to `cortex_compactor_blocks_marked_for_deletion_total` metric. Source blocks marked for deletion by compactor are labelled as `compaction`, while blocks passing the retention period are labelled as `retention`. #3879
 * [FEATURE] Experimental Ruler Storage: Add a separate set of configuration options to configure the ruler storage backend under the `-ruler-storage.` flag prefix. All blocks storage bucket clients and the config service are currently supported. Clients using this implementation will only be enabled if the existing `-ruler.storage` flags are left unset. #3805 #3864
 * [FEATURE] Adds support to S3 server-side encryption using KMS. The S3 server-side encryption config can be overridden on a per-tenant basis for the blocks storage and ruler. Deprecated `-<prefix>.s3.sse-encryption`, you should use the following CLI flags that have been added. #3651 #3810 #3811 #3870
   - `-<prefix>.s3.sse.type`
@@ -26,6 +27,7 @@
     * `-alertmanager.alertmanager-client.tls-ca-path`
     * `-alertmanager.alertmanager-client.tls-server-name`
     * `-alertmanager.alertmanager-client.tls-insecure-skip-verify`
+* [FEATURE] Compactor: added blocks storage per-tenant retention support. This is configured via `-compactor.retention-period`, and can be overridden on a per-tenant basis. #3879
 * [ENHANCEMENT] Ruler: Add TLS and explicit basis authentication configuration options for the HTTP client the ruler uses to communicate with the alertmanager. #3752
   * `-ruler.alertmanager-client.basic-auth-username`: Configure the basic authentication username used by the client. Takes precedent over a URL configured username.
   * `-ruler.alertmanager-client.basic-auth-password`: Configure the basic authentication password used by the client. Takes precedent over a URL configured password.

docs/configuration/config-file-reference.md

@@ -3752,6 +3752,11 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s
 # CLI flag: -store-gateway.tenant-shard-size
 [store_gateway_tenant_shard_size: <int> | default = 0]
 
+# Delete blocks containing samples older than the specified retention period. 0
+# to disable.
+# CLI flag: -compactor.blocks-retention-period
+[compactor_blocks_retention_period: <duration> | default = 0s]
+
 # S3 server-side encryption type. Required to enable server-side encryption
 # overrides for a specific tenant. If not set, the default S3 client settings
 # are used.

pkg/compactor/blocks_cleaner.go

@@ -2,6 +2,7 @@ package compactor
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/go-kit/kit/log"
@@ -35,7 +36,7 @@ type BlocksCleaner struct {
 	services.Service
 
 	cfg          BlocksCleanerConfig
-	cfgProvider  bucket.TenantConfigProvider
+	cfgProvider  ConfigProvider
 	logger       log.Logger
 	bucketClient objstore.Bucket
 	usersScanner *cortex_tsdb.UsersScanner
@@ -50,13 +51,14 @@ type BlocksCleaner struct {
 	runsLastSuccess             prometheus.Gauge
 	blocksCleanedTotal          prometheus.Counter
 	blocksFailedTotal           prometheus.Counter
+	blocksMarkedForDeletion     prometheus.Counter
 	tenantBlocks                *prometheus.GaugeVec
 	tenantMarkedBlocks          *prometheus.GaugeVec
 	tenantPartialBlocks         *prometheus.GaugeVec
 	tenantBucketIndexLastUpdate *prometheus.GaugeVec
 }
 
-func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, usersScanner *cortex_tsdb.UsersScanner, cfgProvider bucket.TenantConfigProvider, logger log.Logger, reg prometheus.Registerer) *BlocksCleaner {
+func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, usersScanner *cortex_tsdb.UsersScanner, cfgProvider ConfigProvider, logger log.Logger, reg prometheus.Registerer) *BlocksCleaner {
 	c := &BlocksCleaner{
 		cfg:          cfg,
 		bucketClient: bucketClient,
@@ -87,6 +89,11 @@ func NewBlocksCleaner(cfg BlocksCleanerConfig, bucketClient objstore.Bucket, use
 			Name: "cortex_compactor_block_cleanup_failures_total",
 			Help: "Total number of blocks failed to be deleted.",
 		}),
+		blocksMarkedForDeletion: promauto.With(reg).NewCounter(prometheus.CounterOpts{
+			Name:        blocksMarkedForDeletionName,
+			Help:        blocksMarkedForDeletionHelp,
+			ConstLabels: prometheus.Labels{"reason": "retention"},
+		}),
 
 		// The following metrics don't have the "cortex_compactor" prefix because not strictly related to
 		// the compactor. They're just tracked by the compactor because it's the most logical place where these
@@ -310,6 +317,17 @@ func (c *BlocksCleaner) cleanUser(ctx context.Context, userID string, firstRun b
 		return err
 	}
 
+	// Mark blocks for future deletion based on the retention period for the user.
+	// Note doing this before UpdateIndex, so it reads in the deletion marks.
+	// The trade-off being that retention is not applied if the index has to be
+	// built, but this is rare.
+	if idx != nil {
+		// We do not want to stop the remaining work in the cleaner if an
+		// error occurs here. Errors are logged in the function.
+		retention := c.cfgProvider.CompactorBlocksRetentionPeriod(userID)
+		c.applyUserRetentionPeriod(ctx, idx, retention, userBucket, userLogger)
+	}
+
 	// Generate an updated in-memory version of the bucket index.
 	w := bucketindex.NewUpdater(c.bucketClient, userID, c.cfgProvider, c.logger)
 	idx, partials, err := w.UpdateIndex(ctx, idx)
@@ -391,3 +409,46 @@ func (c *BlocksCleaner) cleanUserPartialBlocks(ctx context.Context, partials map
 		level.Info(userLogger).Log("msg", "deleted partial block marked for deletion", "block", blockID)
 	}
 }
+
+// applyUserRetentionPeriod marks blocks for deletion which have aged past the retention period.
+func (c *BlocksCleaner) applyUserRetentionPeriod(ctx context.Context, idx *bucketindex.Index, retention time.Duration, userBucket *bucket.UserBucketClient, userLogger log.Logger) {
+	// The retention period of zero is a special value indicating to never delete.
+	if retention <= 0 {
+		return
+	}
+
+	level.Debug(userLogger).Log("msg", "applying retention", "retention", retention.String())
+	blocks := listBlocksOutsideRetentionPeriod(idx, time.Now().Add(-retention))
+
+	// Attempt to mark all blocks. It is not critical if a marking fails, as
+	// the cleaner will retry applying the retention in its next cycle.
+	for _, b := range blocks {
+		level.Info(userLogger).Log("msg", "applied retention: marking block for deletion", "block", b.ID, "maxTime", b.MaxTime)
+		if err := block.MarkForDeletion(ctx, userLogger, userBucket, b.ID, fmt.Sprintf("block exceeding retention of %v", retention), c.blocksMarkedForDeletion); err != nil {
+			level.Warn(userLogger).Log("msg", "failed to mark block for deletion", "block", b.ID, "err", err)
+		}
+	}
+}
+
+// listBlocksOutsideRetentionPeriod determines the blocks which have aged past
+// the specified retention period, and are not already marked for deletion.
+func listBlocksOutsideRetentionPeriod(idx *bucketindex.Index, threshold time.Time) (result bucketindex.Blocks) {
+	// Whilst re-marking a block is not harmful, it is wasteful and generates
+	// a warning log message. Use the block deletion marks already in-memory
+	// to prevent marking blocks already marked for deletion.
+	marked := make(map[ulid.ULID]struct{}, len(idx.BlockDeletionMarks))
+	for _, d := range idx.BlockDeletionMarks {
+		marked[d.ID] = struct{}{}
+	}
+
+	for _, b := range idx.Blocks {
+		maxTime := time.Unix(b.MaxTime/1000, 0)
+		if maxTime.Before(threshold) {
+			if _, isMarked := marked[b.ID]; !isMarked {
+				result = append(result, b)
+			}
+		}
+	}
+
+	return
+}