Ruler ignore users (#4074)

* Extracted AllowedUsers from Compactor code.

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

* Added support for enabled/disabled tenants in ruler.

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

* CHANGELOG.md

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

* CHANGELOG.md and documentation

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

* Removed obsolete test.

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

* Make lint happy.

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

* Review feedback.

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

* Fix field names.

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

Author: pstibrany

CHANGELOG.md

@@ -38,6 +38,7 @@
 * [ENHANCEMENT] Add a metric `cortex_compactor_compaction_interval_seconds` for the compaction interval config value. #4040
 * [ENHANCEMENT] Ingester: added following per-ingester (instance) limits: max number of series in memory (`-ingester.instance-limits.max-series`), max number of users in memory (`-ingester.instance-limits.max-tenants`), max ingestion rate (`-ingester.instance-limits.max-ingestion-rate`), and max inflight requests (`-ingester.instance-limits.max-inflight-push-requests`). These limits are only used when using blocks storage. Limits can also be configured using runtime-config feature, and current values are exported as `cortex_ingester_instance_limits` metric. #3992.
 * [ENHANCEMENT] Cortex is now built with Go 1.16. #4062
+* [ENHANCEMENT] Ruler: Added `-ruler.enabled-tenants` and `-ruler.disabled-tenants` to explicitly enable or disable rules processing for specific tenants. #4074
 * [BUGFIX] Ruler-API: fix bug where `/api/v1/rules/<namespace>/<group_name>` endpoint return `400` instead of `404`. #4013
 * [BUGFIX] Distributor: reverted changes done to rate limiting in #3825. #3948
 * [BUGFIX] Ingester: Fix race condition when opening and closing tsdb concurrently. #3959

docs/configuration/config-file-reference.md

@@ -1584,6 +1584,18 @@ ring:
 # Enable the ruler api
 # CLI flag: -experimental.ruler.enable-api
 [enable_api: <boolean> | default = false]
+
+# Comma separated list of tenants whose rules this ruler can evaluate. If
+# specified, only these tenants will be handled by ruler, otherwise this ruler
+# can process rules from all tenants. Subject to sharding.
+# CLI flag: -ruler.enabled-tenants
+[enabled_tenants: <string> | default = ""]
+
+# Comma separated list of tenants whose rules this ruler cannot evaluate. If
+# specified, a ruler that would normally pick the specified tenant(s) for
+# processing will ignore them instead. Subject to sharding.
+# CLI flag: -ruler.disabled-tenants
+[disabled_tenants: <string> | default = ""]
 ```
 
 ### `ruler_storage_config`

pkg/compactor/compactor.go

@@ -172,18 +172,13 @@ type ConfigProvider interface {
 type Compactor struct {
 	services.Service
 
-	compactorCfg Config
-	storageCfg   cortex_tsdb.BlocksStorageConfig
-	cfgProvider  ConfigProvider
-	logger       log.Logger
-	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{}
+	compactorCfg   Config
+	storageCfg     cortex_tsdb.BlocksStorageConfig
+	cfgProvider    ConfigProvider
+	logger         log.Logger
+	parentLogger   log.Logger
+	registerer     prometheus.Registerer
+	allowedTenants *util.AllowedTenants
 
 	// Functions that creates bucket client, grouper, planner and compactor using the context.
 	// Useful for injecting mock objects from tests.
@@ -272,6 +267,7 @@ func newCompactor(
 		bucketClientFactory:    bucketClientFactory,
 		blocksGrouperFactory:   blocksGrouperFactory,
 		blocksCompactorFactory: blocksCompactorFactory,
+		allowedTenants:         util.NewAllowedTenants(compactorCfg.EnabledTenants, compactorCfg.DisabledTenants),
 
 		compactionRunsStarted: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_compactor_runs_started_total",
@@ -321,21 +317,10 @@ 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, ", "))
+		level.Info(c.logger).Log("msg", "compactor 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, ", "))
+		level.Info(c.logger).Log("msg", "compactor using disabled users", "disabled", strings.Join(compactorCfg.DisabledTenants, ", "))
 	}
 
 	c.Service = services.NewBasicService(c.starting, c.running, c.stopping)
@@ -711,7 +696,7 @@ func (c *Compactor) discoverUsers(ctx context.Context) ([]string, error) {
 }
 
 func (c *Compactor) ownUser(userID string) (bool, error) {
-	if !isAllowedUser(c.enabledUsers, c.disabledUsers, userID) {
+	if !c.allowedTenants.IsAllowed(userID) {
 		return false, nil
 	}
 
@@ -738,22 +723,6 @@ func (c *Compactor) ownUser(userID string) (bool, error) {
 	return rs.Instances[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
-}
-
 const compactorMetaPrefix = "compactor-meta-"
 
 // metaSyncDirForUser returns directory to store cached meta files.

pkg/compactor/compactor_test.go

@@ -1152,48 +1152,6 @@ 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))
-		})
-	}
-}
-
 func TestCompactor_DeleteLocalSyncFiles(t *testing.T) {
 	numUsers := 10
 

pkg/ruler/ruler.go

@@ -107,6 +107,9 @@ type Config struct {
 
 	EnableAPI bool `yaml:"enable_api"`
 
+	EnabledTenants  flagext.StringSliceCSV `yaml:"enabled_tenants"`
+	DisabledTenants flagext.StringSliceCSV `yaml:"disabled_tenants"`
+
 	RingCheckPeriod time.Duration `yaml:"-"`
 }
 
@@ -163,6 +166,9 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.ForGracePeriod, "ruler.for-grace-period", 10*time.Minute, `Minimum duration between alert and restored "for" state. This is maintained only for alerts with configured "for" time greater than grace period.`)
 	f.DurationVar(&cfg.ResendDelay, "ruler.resend-delay", time.Minute, `Minimum amount of time to wait before resending an alert to Alertmanager.`)
 
+	f.Var(&cfg.EnabledTenants, "ruler.enabled-tenants", "Comma separated list of tenants whose rules this ruler can evaluate. If specified, only these tenants will be handled by ruler, otherwise this ruler can process rules from all tenants. Subject to sharding.")
+	f.Var(&cfg.DisabledTenants, "ruler.disabled-tenants", "Comma separated list of tenants whose rules this ruler cannot evaluate. If specified, a ruler that would normally pick the specified tenant(s) for processing will ignore them instead. Subject to sharding.")
+
 	cfg.RingCheckPeriod = 5 * time.Second
 }
 
@@ -224,20 +230,23 @@ type Ruler struct {
 	ringCheckErrors prometheus.Counter
 	rulerSync       *prometheus.CounterVec
 
+	allowedTenants *util.AllowedTenants
+
 	registry prometheus.Registerer
 	logger   log.Logger
 }
 
 // NewRuler creates a new ruler from a distributor and chunk store.
 func NewRuler(cfg Config, manager MultiTenantManager, reg prometheus.Registerer, logger log.Logger, ruleStore rulestore.RuleStore, limits RulesLimits) (*Ruler, error) {
 	ruler := &Ruler{
-		cfg:         cfg,
-		store:       ruleStore,
-		manager:     manager,
-		registry:    reg,
-		logger:      logger,
-		limits:      limits,
-		clientsPool: newRulerClientPool(cfg.ClientTLSConfig, logger, reg),
+		cfg:            cfg,
+		store:          ruleStore,
+		manager:        manager,
+		registry:       reg,
+		logger:         logger,
+		limits:         limits,
+		clientsPool:    newRulerClientPool(cfg.ClientTLSConfig, logger, reg),
+		allowedTenants: util.NewAllowedTenants(cfg.EnabledTenants, cfg.DisabledTenants),
 
 		ringCheckErrors: promauto.With(reg).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_ruler_ring_check_errors_total",
@@ -250,6 +259,13 @@ func NewRuler(cfg Config, manager MultiTenantManager, reg prometheus.Registerer,
 		}, []string{"reason"}),
 	}
 
+	if len(cfg.EnabledTenants) > 0 {
+		level.Info(ruler.logger).Log("msg", "ruler using enabled users", "enabled", strings.Join(cfg.EnabledTenants, ", "))
+	}
+	if len(cfg.DisabledTenants) > 0 {
+		level.Info(ruler.logger).Log("msg", "ruler using disabled users", "disabled", strings.Join(cfg.DisabledTenants, ", "))
+	}
+
 	if cfg.EnableSharding {
 		ringStore, err := kv.NewClient(
 			cfg.Ring.KVStore,
@@ -472,20 +488,32 @@ func (r *Ruler) syncRules(ctx context.Context, reason string) {
 	r.manager.SyncRuleGroups(ctx, configs)
 }
 
-func (r *Ruler) listRules(ctx context.Context) (map[string]rulespb.RuleGroupList, error) {
+func (r *Ruler) listRules(ctx context.Context) (result map[string]rulespb.RuleGroupList, err error) {
 	switch {
 	case !r.cfg.EnableSharding:
-		return r.listRulesNoSharding(ctx)
+		result, err = r.listRulesNoSharding(ctx)
 
 	case r.cfg.ShardingStrategy == util.ShardingStrategyDefault:
-		return r.listRulesShardingDefault(ctx)
+		result, err = r.listRulesShardingDefault(ctx)
 
 	case r.cfg.ShardingStrategy == util.ShardingStrategyShuffle:
-		return r.listRulesShuffleSharding(ctx)
+		result, err = r.listRulesShuffleSharding(ctx)
 
 	default:
 		return nil, errors.New("invalid sharding configuration")
 	}
+
+	if err != nil {
+		return
+	}
+
+	for userID := range result {
+		if !r.allowedTenants.IsAllowed(userID) {
+			level.Debug(r.logger).Log("msg", "ignoring rule groups for user, not allowed", "user", userID)
+			delete(result, userID)
+		}
+	}
+	return
 }
 
 func (r *Ruler) listRulesNoSharding(ctx context.Context) (map[string]rulespb.RuleGroupList, error) {