Introduced time-based concurrent compaction

Signed-off-by: Marco Pracucci <[email protected]>

Author: pracucci

pkg/compactor/blocks_sharding_filter.go

@@ -15,6 +15,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"github.com/prometheus/prometheus/tsdb"
+	tsdb_errors "github.com/prometheus/prometheus/tsdb/errors"
 	"github.com/thanos-io/thanos/pkg/block"
 	"github.com/thanos-io/thanos/pkg/compact"
 	"github.com/thanos-io/thanos/pkg/compact/downsample"
@@ -28,20 +29,19 @@ import (
 
 // Config holds the Compactor config.
 type Config struct {
-	BlockRanges          cortex_tsdb.DurationList `yaml:"block_ranges"`
-	BlockSyncConcurrency int                      `yaml:"block_sync_concurrency"`
-	MetaSyncConcurrency  int                      `yaml:"meta_sync_concurrency"`
-	ConsistencyDelay     time.Duration            `yaml:"consistency_delay"`
-	DataDir              string                   `yaml:"data_dir"`
-	CompactionInterval   time.Duration            `yaml:"compaction_interval"`
-	CompactionRetries    int                      `yaml:"compaction_retries"`
-	DeletionDelay        time.Duration            `yaml:"deletion_delay"`
+	BlockRanges           cortex_tsdb.DurationList `yaml:"block_ranges"`
+	BlockSyncConcurrency  int                      `yaml:"block_sync_concurrency"`
+	MetaSyncConcurrency   int                      `yaml:"meta_sync_concurrency"`
+	ConsistencyDelay      time.Duration            `yaml:"consistency_delay"`
+	DataDir               string                   `yaml:"data_dir"`
+	CompactionInterval    time.Duration            `yaml:"compaction_interval"`
+	CompactionRetries     int                      `yaml:"compaction_retries"`
+	CompactionConcurrency int                      `yaml:"compaction_concurrency"`
+	DeletionDelay         time.Duration            `yaml:"deletion_delay"`
 
 	// Compactors sharding.
-	ShardingEnabled            bool       `yaml:"sharding_enabled"`
-	ShardingRing               RingConfig `yaml:"sharding_ring"`
-	PerTenantNumShards         uint       `yaml:"per_tenant_num_shards"`
-	PerTenantShardsConcurrency int        `yaml:"per_tenant_shards_concurrency"`
+	ShardingEnabled bool       `yaml:"sharding_enabled"`
+	ShardingRing    RingConfig `yaml:"sharding_ring"`
 
 	// No need to add options to customize the retry backoff,
 	// given the defaults should be fine, but allow to override
@@ -59,15 +59,14 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.retryMaxBackoff = time.Minute
 
 	f.Var(&cfg.BlockRanges, "compactor.block-ranges", "List of compaction time ranges.")
-	f.DurationVar(&cfg.ConsistencyDelay, "compactor.consistency-delay", 30*time.Minute, fmt.Sprintf("Minimum age of fresh (non-compacted) blocks before they are being processed. Malformed blocks older than the maximum of consistency-delay and %s will be removed.", compact.PartialUploadThresholdAge))
+	f.DurationVar(&cfg.ConsistencyDelay, "compactor.consistency-delay", 0, fmt.Sprintf("Minimum age of fresh (non-compacted) blocks before they are being processed. Malformed blocks older than the maximum of consistency-delay and %s will be removed.", compact.PartialUploadThresholdAge))
 	f.IntVar(&cfg.BlockSyncConcurrency, "compactor.block-sync-concurrency", 20, "Number of Go routines to use when syncing block index and chunks files from the long term storage.")
 	f.IntVar(&cfg.MetaSyncConcurrency, "compactor.meta-sync-concurrency", 20, "Number of Go routines to use when syncing block meta files from the long term storage.")
+	f.IntVar(&cfg.CompactionConcurrency, "compactor.compaction-concurrency", 1, "Max number of concurrent compactions running.")
 	f.StringVar(&cfg.DataDir, "compactor.data-dir", "./data", "Data directory in which to cache blocks and process compactions")
 	f.DurationVar(&cfg.CompactionInterval, "compactor.compaction-interval", time.Hour, "The frequency at which the compaction runs")
 	f.IntVar(&cfg.CompactionRetries, "compactor.compaction-retries", 3, "How many times to retry a failed compaction during a single compaction interval")
 	f.BoolVar(&cfg.ShardingEnabled, "compactor.sharding-enabled", false, "Shard tenants across multiple compactor instances. Sharding is required if you run multiple compactor instances, in order to coordinate compactions and avoid race conditions leading to the same tenant blocks simultaneously compacted by different instances.")
-	f.UintVar(&cfg.PerTenantNumShards, "compactor.per-tenant-num-shards", 1, "Number of shards a single tenant blocks should be grouped into (0 or 1 means per-tenant blocks sharding is disabled).")
-	f.IntVar(&cfg.PerTenantShardsConcurrency, "compactor.per-tenant-shards-concurrency", 1, "Number of concurrent shards compacted for a single tenant.")
 	f.DurationVar(&cfg.DeletionDelay, "compactor.deletion-delay", 12*time.Hour, "Time before a block marked for deletion is deleted from bucket. "+
 		"If not 0, blocks will be marked for deletion and compactor component will delete blocks marked for deletion from the bucket. "+
 		"If delete-delay is 0, blocks will be deleted straight away. Note that deleting blocks immediately can cause query failures, "+
@@ -82,7 +81,7 @@ type Compactor struct {
 	storageCfg   cortex_tsdb.Config
 	logger       log.Logger
 
-	// function that creates bucket client and TSDB compactor using the context.
+	// Function that creates bucket client and TSDB compactor using the context.
 	// Useful for injecting mock objects from tests.
 	createBucketClientAndTsdbCompactor func(ctx context.Context) (objstore.Bucket, tsdb.Compactor, error)
 
@@ -271,10 +270,12 @@ func (c *Compactor) compactUsersWithRetries(ctx context.Context) {
 	c.compactionRunsStarted.Inc()
 
 	for retries.Ongoing() {
-		if success := c.compactUsers(ctx); success {
+		if err := c.compactUsers(ctx); err == nil {
 			c.compactionRunsCompleted.Inc()
 			c.compactionRunsLastSuccess.SetToCurrentTime()
 			return
+		} else if errors.Is(err, context.Canceled) {
+			return
 		}
 
 		retries.Wait()
@@ -283,20 +284,22 @@ func (c *Compactor) compactUsersWithRetries(ctx context.Context) {
 	c.compactionRunsFailed.Inc()
 }
 
-func (c *Compactor) compactUsers(ctx context.Context) bool {
+func (c *Compactor) compactUsers(ctx context.Context) error {
 	level.Info(c.logger).Log("msg", "discovering users from bucket")
 	users, err := c.discoverUsers(ctx)
 	if err != nil {
 		level.Error(c.logger).Log("msg", "failed to discover users from bucket", "err", err)
-		return false
+		return errors.Wrap(err, "failed to discover users from bucket")
 	}
 	level.Info(c.logger).Log("msg", "discovered users from bucket", "users", len(users))
 
+	errs := tsdb_errors.MultiError{}
+
 	for _, userID := range users {
 		// Ensure the context has not been canceled (ie. compactor shutdown has been triggered).
 		if ctx.Err() != nil {
 			level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "err", err)
-			return false
+			return ctx.Err()
 		}
 
 		// If sharding is enabled, ensure the user ID belongs to our shard.
@@ -314,13 +317,14 @@ func (c *Compactor) compactUsers(ctx context.Context) bool {
 
 		if err = c.compactUser(ctx, userID); err != nil {
 			level.Error(c.logger).Log("msg", "failed to compact user blocks", "user", userID, "err", err)
+			errs.Add(errors.Wrapf(err, "failed to compact user blocks (user: %s)", userID))
 			continue
 		}
 
 		level.Info(c.logger).Log("msg", "successfully compacted user blocks", "user", userID)
 	}
 
-	return true
+	return errs.Err()
 }
 
 func (c *Compactor) compactUser(ctx context.Context, userID string) error {
@@ -348,9 +352,11 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 		// the directory used by the Thanos Syncer, whatever is the user ID.
 		path.Join(c.compactorCfg.DataDir, "meta-"+userID),
 		reg,
+		// List of filters to apply (order matters).
 		[]block.MetadataFilter{
-			// List of filters to apply (order matters).
-			NewBlocksShardingFilter(uint32(c.compactorCfg.PerTenantNumShards)),
+			// Remove the ingester ID because we don't shard blocks anymore, while still
+			// honoring the shard ID if sharding was already done.
+			NewLabelRemoverFilter([]string{cortex_tsdb.IngesterIDExternalLabel}),
 			block.NewConsistencyDelayMetaFilter(ulogger, c.compactorCfg.ConsistencyDelay, reg),
 			ignoreDeletionMarkFilter,
 			deduplicateBlocksFilter,
@@ -380,10 +386,11 @@ func (c *Compactor) compactUser(ctx context.Context, userID string) error {
 	compactor, err := compact.NewBucketCompactor(
 		ulogger,
 		syncer,
-		c.tsdbCompactor,
+		NewTimeShardingGrouper(bucket, c.compactorCfg.BlockRanges.ToMilliseconds(), ulogger),
+		NewTimeShardingCompactor(c.tsdbCompactor, c.compactorCfg.BlockRanges.ToMilliseconds()),
 		path.Join(c.compactorCfg.DataDir, "compact"),
 		bucket,
-		c.compactorCfg.PerTenantShardsConcurrency,
+		c.compactorCfg.CompactionConcurrency,
 	)
 	if err != nil {
 		return errors.Wrap(err, "failed to create bucket compactor")