Added option to enable/disable specific users for compaction. (#3385)

* Added option to enable/disable specific users for compaction.

Signed-off-by: Peter Štibraný <[email protected]>

* Log list of enabled/disabled users on startup.

Signed-off-by: Peter Štibraný <[email protected]>

* Always check if compactor owns the user.

Signed-off-by: Peter Štibraný <[email protected]>

* Make new flags visible in the documentation. Added CHANGELOG.

Signed-off-by: Peter Štibraný <[email protected]>

Author: pstibrany

CHANGELOG.md

@@ -87,6 +87,7 @@
 * [ENHANCEMENT] Added `cortex_alertmanager_config_last_reload_successful_seconds` metric to show timestamp of last successful AM config reload. #3289
 * [ENHANCEMENT] Blocks storage: reduced number of bucket listing operations to list block content (applies to newly created blocks only). #3363
 * [ENHANCEMENT] Ruler: Include the tenant ID on the notifier logs. #3372
+* [ENHANCEMENT] Blocks storage Compactor: Added `-compactor.enabled-tenants` and `-compactor.disabled-tenants` to explicitly enable or disable compaction of specific tenants. #3385
 * [BUGFIX] No-longer-needed ingester operations for queries triggered by queriers and rulers are now canceled. #3178
 * [BUGFIX] Ruler: directories in the configured `rules-path` will be removed on startup and shutdown in order to ensure they don't persist between runs. #3195
 * [BUGFIX] Handle hash-collisions in the query path. #3192

docs/blocks-storage/compactor.md

@@ -124,6 +124,18 @@ compactor:
   # CLI flag: -compactor.deletion-delay
   [deletion_delay: <duration> | default = 12h]
 
+  # Comma separated list of tenants that can be compacted. If specified, only
+  # these tenants will be compacted by compactor, otherwise all tenants can be
+  # compacted. Subject to sharding.
+  # CLI flag: -compactor.enabled-tenants
+  [enabled_tenants: <string> | default = ""]
+
+  # Comma separated list of tenants that cannot be compacted by this compactor.
+  # If specified, and compactor would normally pick given tenant for compaction
+  # (via -compactor.enabled-tenants or sharding), it will be ignored instead.
+  # CLI flag: -compactor.disabled-tenants
+  [disabled_tenants: <string> | default = ""]
+
   # 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

docs/configuration/config-file-reference.md

@@ -3681,6 +3681,18 @@ The `compactor_config` configures the compactor for the blocks storage.
 # CLI flag: -compactor.deletion-delay
 [deletion_delay: <duration> | default = 12h]
 
+# Comma separated list of tenants that can be compacted. If specified, only
+# these tenants will be compacted by compactor, otherwise all tenants can be
+# compacted. Subject to sharding.
+# CLI flag: -compactor.enabled-tenants
+[enabled_tenants: <string> | default = ""]
+
+# Comma separated list of tenants that cannot be compacted by this compactor. If
+# specified, and compactor would normally pick given tenant for compaction (via
+# -compactor.enabled-tenants or sharding), it will be ignored instead.
+# CLI flag: -compactor.disabled-tenants
+[disabled_tenants: <string> | default = ""]
+
 # 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

pkg/compactor/compactor.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	cortex_tsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/services"
 )
 
@@ -39,6 +40,9 @@ type Config struct {
 	CompactionConcurrency int                      `yaml:"compaction_concurrency"`
 	DeletionDelay         time.Duration            `yaml:"deletion_delay"`
 
+	EnabledTenants  flagext.StringSliceCSV `yaml:"enabled_tenants"`
+	DisabledTenants flagext.StringSliceCSV `yaml:"disabled_tenants"`
+
 	// Compactors sharding.
 	ShardingEnabled bool       `yaml:"sharding_enabled"`
 	ShardingRing    RingConfig `yaml:"sharding_ring"`
@@ -71,6 +75,9 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 		"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, "+
 		"if store gateway still has the block loaded, or compactor is ignoring the deletion because it's compacting the block at the same time.")
+
+	f.Var(&cfg.EnabledTenants, "compactor.enabled-tenants", "Comma separated list of tenants that can be compacted. If specified, only these tenants will be compacted by compactor, otherwise all tenants can be compacted. Subject to sharding.")
+	f.Var(&cfg.DisabledTenants, "compactor.disabled-tenants", "Comma separated list of tenants that cannot be compacted by this compactor. If specified, and compactor would normally pick given tenant for compaction (via -compactor.enabled-tenants or sharding), it will be ignored instead.")
 }
 
 // Compactor is a multi-tenant TSDB blocks compactor based on Thanos.
@@ -83,6 +90,12 @@ type Compactor struct {
 	parentLogger log.Logger
 	registerer   prometheus.Registerer
 
+	// If empty, all users are enabled. If not empty, only users in the map are enabled (possibly owned by compactor, also subject to sharding configuration).
+	enabledUsers map[string]struct{}
+
+	// If empty, no users are disabled. If not empty, users in the map are disabled (not owned by this compactor).
+	disabledUsers map[string]struct{}
+
 	// 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)
@@ -179,6 +192,24 @@ func newCompactor(
 		}),
 	}
 
+	if len(compactorCfg.EnabledTenants) > 0 {
+		c.enabledUsers = map[string]struct{}{}
+		for _, u := range compactorCfg.EnabledTenants {
+			c.enabledUsers[u] = struct{}{}
+		}
+
+		level.Info(c.logger).Log("msg", "using enabled users", "enabled", strings.Join(compactorCfg.EnabledTenants, ", "))
+	}
+
+	if len(compactorCfg.DisabledTenants) > 0 {
+		c.disabledUsers = map[string]struct{}{}
+		for _, u := range compactorCfg.DisabledTenants {
+			c.disabledUsers[u] = struct{}{}
+		}
+
+		level.Info(c.logger).Log("msg", "using disabled users", "disabled", strings.Join(compactorCfg.DisabledTenants, ", "))
+	}
+
 	c.Service = services.NewBasicService(c.starting, c.running, c.stopping)
 
 	return c, nil
@@ -322,15 +353,13 @@ func (c *Compactor) compactUsers(ctx context.Context) error {
 			return ctx.Err()
 		}
 
-		// If sharding is enabled, ensure the user ID belongs to our shard.
-		if c.compactorCfg.ShardingEnabled {
-			if owned, err := c.ownUser(userID); err != nil {
-				level.Warn(c.logger).Log("msg", "unable to check if user is owned by this shard", "user", userID, "err", err)
-				continue
-			} else if !owned {
-				level.Debug(c.logger).Log("msg", "skipping user because not owned by this shard", "user", userID)
-				continue
-			}
+		// Ensure the user ID belongs to our shard.
+		if owned, err := c.ownUser(userID); err != nil {
+			level.Warn(c.logger).Log("msg", "unable to check if user is owned by this shard", "user", userID, "err", err)
+			continue
+		} else if !owned {
+			level.Debug(c.logger).Log("msg", "skipping user because not owned by this shard", "user", userID)
+			continue
 		}
 
 		level.Info(c.logger).Log("msg", "starting compaction of user blocks", "user", userID)
@@ -444,6 +473,10 @@ func (c *Compactor) discoverUsers(ctx context.Context) ([]string, error) {
 }
 
 func (c *Compactor) ownUser(userID string) (bool, error) {
+	if !isAllowedUser(c.enabledUsers, c.disabledUsers, userID) {
+		return false, nil
+	}
+
 	// Always owned if sharding is disabled.
 	if !c.compactorCfg.ShardingEnabled {
 		return true, nil
@@ -466,3 +499,19 @@ func (c *Compactor) ownUser(userID string) (bool, error) {
 
 	return rs.Ingesters[0].Addr == c.ringLifecycler.Addr, nil
 }
+
+func isAllowedUser(enabledUsers, disabledUsers map[string]struct{}, userID string) bool {
+	if len(enabledUsers) > 0 {
+		if _, ok := enabledUsers[userID]; !ok {
+			return false
+		}
+	}
+
+	if len(disabledUsers) > 0 {
+		if _, ok := disabledUsers[userID]; ok {
+			return false
+		}
+	}
+
+	return true
+}

pkg/compactor/compactor_test.go

@@ -937,3 +937,45 @@ func mockDeletionMarkJSON(id string, deletionTime time.Time) string {
 
 	return string(content)
 }
+
+func TestAllowedUser(t *testing.T) {
+	testCases := map[string]struct {
+		enabled, disabled map[string]struct{}
+		user              string
+		expected          bool
+	}{
+		"no enabled or disabled": {
+			user:     "test",
+			expected: true,
+		},
+
+		"only enabled, enabled": {
+			enabled:  map[string]struct{}{"user": {}},
+			user:     "user",
+			expected: true,
+		},
+
+		"only enabled, disabled": {
+			enabled:  map[string]struct{}{"user": {}},
+			user:     "not user",
+			expected: false,
+		},
+
+		"only disabled, disabled": {
+			disabled: map[string]struct{}{"user": {}},
+			user:     "user",
+			expected: false,
+		},
+
+		"only disabled, enabled": {
+			disabled: map[string]struct{}{"user": {}},
+			user:     "not user",
+			expected: true,
+		},
+	}
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			require.Equal(t, tc.expected, isAllowedUser(tc.enabled, tc.disabled, tc.user))
+		})
+	}
+}