Introduce user scanner to Ruler/Alertmanager

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

CHANGELOG.md

@@ -4,6 +4,7 @@
 * [CHANGE] StoreGateway/Alertmanager: Add default 5s connection timeout on client. #6603
 * [CHANGE] Ingester: Remove EnableNativeHistograms config flag and instead gate keep through new per-tenant limit at ingestion. #6718
 * [CHANGE] Validate a tenantID when to use a single tenant resolver. #6727
+* [FEATURE] Alertmanager/Ruler: Introduce a user scanner to reduce the number of list calls to object storage. #6999 
 * [FEATURE] Distributor: Add an experimental `-distributor.otlp.enable-type-and-unit-labels` flag to add `__type__` and `__unit__` labels for OTLP metrics. #6969
 * [FEATURE] Distributor: Add an experimental `-distributor.otlp.allow-delta-temporality` flag to ingest delta temporality otlp metrics. #6934
 * [FEATURE] Query Frontend: Add dynamic interval size for query splitting. This is enabled by configuring experimental flags `querier.max-shards-per-query` and/or `querier.max-fetched-data-duration-per-query`. The split interval size is dynamically increased to maintain a number of shards and total duration fetched below the configured values. #6458

docs/configuration/config-file-reference.md

@@ -981,6 +981,21 @@ local:
   # Path at which alertmanager configurations are stored.
   # CLI flag: -alertmanager-storage.local.path
   [path: <string> | default = ""]
+
+users_scanner:
+  # Strategy to use to scan users. Supported values are: list, user_index.
+  # CLI flag: -alertmanager-storage.users-scanner.strategy
+  [strategy: <string> | default = "list"]
+
+  # Maximum period of time to consider the user index as stale. Fall back to the
+  # base scanner if stale. Only valid when strategy is user_index.
+  # CLI flag: -alertmanager-storage.users-scanner.user-index.max-stale-period
+  [max_stale_period: <duration> | default = 1h]
+
+  # TTL of the cached users. 0 disables caching and relies on caching at bucket
+  # client level.
+  # CLI flag: -alertmanager-storage.users-scanner.cache-ttl
+  [cache_ttl: <duration> | default = 0s]
 ```
 
 ### `blocks_storage_config`
@@ -5772,6 +5787,21 @@ local:
   # Directory to scan for rules
   # CLI flag: -ruler-storage.local.directory
   [directory: <string> | default = ""]
+
+users_scanner:
+  # Strategy to use to scan users. Supported values are: list, user_index.
+  # CLI flag: -ruler-storage.users-scanner.strategy
+  [strategy: <string> | default = "list"]
+
+  # Maximum period of time to consider the user index as stale. Fall back to the
+  # base scanner if stale. Only valid when strategy is user_index.
+  # CLI flag: -ruler-storage.users-scanner.user-index.max-stale-period
+  [max_stale_period: <duration> | default = 1h]
+
+  # TTL of the cached users. 0 disables caching and relies on caching at bucket
+  # client level.
+  # CLI flag: -ruler-storage.users-scanner.cache-ttl
+  [cache_ttl: <duration> | default = 0s]
 ```
 
 ### `runtime_configuration_storage_config`

pkg/alertmanager/alertstore/bucketclient/bucket_client.go

@@ -3,19 +3,22 @@ package bucketclient
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"strings"
 	"sync"
 
 	"github.com/go-kit/log"
 	"github.com/gogo/protobuf/proto"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/thanos-io/objstore"
-
-	"github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/thanos-io/thanos/pkg/extprom"
 
 	"github.com/cortexproject/cortex/pkg/alertmanager/alertspb"
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb/users"
 	"github.com/cortexproject/cortex/pkg/util/concurrency"
 	"github.com/cortexproject/cortex/pkg/util/runutil"
 )
@@ -45,27 +48,36 @@ type BucketAlertStore struct {
 	amBucket     objstore.Bucket
 	cfgProvider  bucket.TenantConfigProvider
 	logger       log.Logger
+
+	usersScanner users.Scanner
 }
 
-func NewBucketAlertStore(bkt objstore.Bucket, cfgProvider bucket.TenantConfigProvider, logger log.Logger) *BucketAlertStore {
+func NewBucketAlertStore(bkt objstore.InstrumentedBucket, userScannerCfg tsdb.UsersScannerConfig, cfgProvider bucket.TenantConfigProvider, logger log.Logger, reg prometheus.Registerer) (*BucketAlertStore, error) {
+	alertBucket := bucket.NewPrefixedBucketClient(bkt, alertsPrefix)
+
+	usersScanner, err := users.NewScanner(userScannerCfg, alertBucket, logger, extprom.WrapRegistererWith(prometheus.Labels{"component": "alertmanager"}, reg))
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to initialize alertmanager users scanner")
+	}
 	return &BucketAlertStore{
-		alertsBucket: bucket.NewPrefixedBucketClient(bkt, alertsPrefix),
+		usersScanner: usersScanner,
+		alertsBucket: alertBucket,
 		amBucket:     bucket.NewPrefixedBucketClient(bkt, alertmanagerPrefix),
 		cfgProvider:  cfgProvider,
 		logger:       logger,
-	}
+	}, nil
 }
 
 // ListAllUsers implements alertstore.AlertStore.
 func (s *BucketAlertStore) ListAllUsers(ctx context.Context) ([]string, error) {
-	var userIDs []string
-
-	err := s.alertsBucket.Iter(ctx, "", func(key string) error {
-		userIDs = append(userIDs, key)
-		return nil
-	})
-
-	return userIDs, err
+	active, deleting, _, err := s.usersScanner.ScanUsers(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("unable to list users in alertmanager store bucket: %w", err)
+	}
+	userIDs := make([]string, 0, len(active)+len(deleting))
+	userIDs = append(userIDs, active...)
+	userIDs = append(userIDs, deleting...)
+	return userIDs, nil
 }
 
 // GetAlertConfigs implements alertstore.AlertStore.

pkg/alertmanager/alertstore/config.go

@@ -7,13 +7,15 @@ import (
 	"github.com/cortexproject/cortex/pkg/alertmanager/alertstore/local"
 	"github.com/cortexproject/cortex/pkg/configs/client"
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 )
 
-// Config configures a the alertmanager storage backend.
+// Config configures the alertmanager storage backend.
 type Config struct {
 	bucket.Config `yaml:",inline"`
-	ConfigDB      client.Config     `yaml:"configdb"`
-	Local         local.StoreConfig `yaml:"local"`
+	ConfigDB      client.Config           `yaml:"configdb"`
+	Local         local.StoreConfig       `yaml:"local"`
+	UsersScanner  tsdb.UsersScannerConfig `yaml:"users_scanner"`
 }
 
 // RegisterFlags registers the backend storage config.
@@ -24,6 +26,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.ConfigDB.RegisterFlagsWithPrefix(prefix, f)
 	cfg.Local.RegisterFlagsWithPrefix(prefix, f)
 	cfg.RegisterFlagsWithPrefix(prefix, f)
+	cfg.UsersScanner.RegisterFlagsWithPrefix(prefix, f)
 }
 
 // IsFullStateSupported returns if the given configuration supports access to FullState objects.

pkg/alertmanager/alertstore/store.go

@@ -67,5 +67,5 @@ func NewAlertStore(ctx context.Context, cfg Config, cfgProvider bucket.TenantCon
 		return nil, err
 	}
 
-	return bucketclient.NewBucketAlertStore(bucketClient, cfgProvider, logger), nil
+	return bucketclient.NewBucketAlertStore(bucketClient, cfg.UsersScanner, cfgProvider, logger, reg)
 }

pkg/alertmanager/alertstore/store_test.go

@@ -8,12 +8,14 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/prometheus/alertmanager/cluster/clusterpb"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/thanos-io/objstore"
 
 	"github.com/cortexproject/cortex/pkg/alertmanager/alertspb"
 	"github.com/cortexproject/cortex/pkg/alertmanager/alertstore/bucketclient"
+	cortextsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 )
 
 var (
@@ -42,6 +44,15 @@ func TestAlertStore_ListAllUsers(t *testing.T) {
 			require.NoError(t, err)
 			assert.ElementsMatch(t, []string{"user-1", "user-2"}, users)
 		}
+
+		{
+			// delete user-1 alertmanager config
+			require.NoError(t, store.DeleteAlertConfig(ctx, "user-1"))
+
+			users, err := store.ListAllUsers(ctx)
+			require.NoError(t, err)
+			assert.ElementsMatch(t, []string{"user-2"}, users)
+		}
 	})
 }
 
@@ -172,7 +183,10 @@ func TestAlertStore_DeleteAlertConfig(t *testing.T) {
 func runForEachAlertStore(t *testing.T, testFn func(t *testing.T, store AlertStore, b *mockBucket, client interface{})) {
 	bucketClient := objstore.NewInMemBucket()
 	mBucketClient := &mockBucket{Bucket: bucketClient}
-	bucketStore := bucketclient.NewBucketAlertStore(mBucketClient, nil, log.NewNopLogger())
+	usersScannerConfig := cortextsdb.UsersScannerConfig{Strategy: cortextsdb.UserScanStrategyList}
+	reg := prometheus.NewPedanticRegistry()
+	bucketStore, err := bucketclient.NewBucketAlertStore(mBucketClient, usersScannerConfig, nil, log.NewNopLogger(), reg)
+	assert.NoError(t, err)
 
 	stores := map[string]struct {
 		store  AlertStore
@@ -212,7 +226,10 @@ func makeTestFullState(content string) alertspb.FullStateDesc {
 func TestBucketAlertStore_GetSetDeleteFullState(t *testing.T) {
 	bucket := objstore.NewInMemBucket()
 	mBucketClient := &mockBucket{Bucket: bucket}
-	store := bucketclient.NewBucketAlertStore(mBucketClient, nil, log.NewNopLogger())
+	usersScannerConfig := cortextsdb.UsersScannerConfig{Strategy: cortextsdb.UserScanStrategyList}
+	reg := prometheus.NewPedanticRegistry()
+	store, err := bucketclient.NewBucketAlertStore(mBucketClient, usersScannerConfig, nil, log.NewNopLogger(), reg)
+	assert.NoError(t, err)
 	ctx := context.Background()
 
 	state1 := makeTestFullState("one")
@@ -297,6 +314,14 @@ type mockBucket struct {
 	err error
 }
 
+func (m *mockBucket) WithExpectedErrs(expectedFunc objstore.IsOpFailureExpectedFunc) objstore.Bucket {
+	return m
+}
+
+func (m *mockBucket) ReaderWithExpectedErrs(expectedFunc objstore.IsOpFailureExpectedFunc) objstore.BucketReader {
+	return m
+}
+
 func (m *mockBucket) Get(ctx context.Context, name string) (io.ReadCloser, error) {
 	if m.err != nil {
 		return nil, m.err

pkg/ruler/rulestore/bucketclient/bucket_client.go

@@ -13,12 +13,16 @@ import (
 	"github.com/go-kit/log/level"
 	"github.com/gogo/protobuf/proto"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/thanos-io/objstore"
+	"github.com/thanos-io/thanos/pkg/extprom"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/cortexproject/cortex/pkg/ruler/rulespb"
 	"github.com/cortexproject/cortex/pkg/ruler/rulestore"
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb/users"
 	"github.com/cortexproject/cortex/pkg/util/multierror"
 )
 
@@ -42,14 +46,24 @@ type BucketRuleStore struct {
 	bucket      objstore.Bucket
 	cfgProvider bucket.TenantConfigProvider
 	logger      log.Logger
+
+	usersScanner users.Scanner
 }
 
-func NewBucketRuleStore(bkt objstore.Bucket, cfgProvider bucket.TenantConfigProvider, logger log.Logger) *BucketRuleStore {
-	return &BucketRuleStore{
-		bucket:      bucket.NewPrefixedBucketClient(bkt, rulesPrefix),
-		cfgProvider: cfgProvider,
-		logger:      logger,
+func NewBucketRuleStore(bkt objstore.Bucket, userScannerCfg tsdb.UsersScannerConfig, cfgProvider bucket.TenantConfigProvider, logger log.Logger, reg prometheus.Registerer) (*BucketRuleStore, error) {
+	rulesBucket := bucket.NewPrefixedBucketClient(bkt, rulesPrefix)
+
+	usersScanner, err := users.NewScanner(userScannerCfg, rulesBucket, logger, extprom.WrapRegistererWith(prometheus.Labels{"component": "ruler"}, reg))
+	if err != nil {
+		return nil, errors.Wrap(err, "unable to initialize ruler users scanner")
 	}
+
+	return &BucketRuleStore{
+		bucket:       rulesBucket,
+		cfgProvider:  cfgProvider,
+		logger:       logger,
+		usersScanner: usersScanner,
+	}, nil
 }
 
 // getRuleGroup loads and return a rules group. If existing rule group is supplied, it is Reset and reused. If nil, new RuleGroupDesc is allocated.
@@ -94,16 +108,14 @@ func (b *BucketRuleStore) getRuleGroup(ctx context.Context, userID, namespace, g
 
 // ListAllUsers implements rules.RuleStore.
 func (b *BucketRuleStore) ListAllUsers(ctx context.Context) ([]string, error) {
-	var users []string
-	err := b.bucket.Iter(ctx, "", func(user string) error {
-		users = append(users, strings.TrimSuffix(user, objstore.DirDelim))
-		return nil
-	})
+	active, deleting, _, err := b.usersScanner.ScanUsers(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("unable to list users in rule store bucket: %w", err)
 	}
-
-	return users, nil
+	userIDs := make([]string, 0, len(active)+len(deleting))
+	userIDs = append(userIDs, active...)
+	userIDs = append(userIDs, deleting...)
+	return userIDs, nil
 }
 
 // ListAllRuleGroups implements rules.RuleStore.

pkg/ruler/rulestore/bucketclient/bucket_client_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/pkg/errors"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/rulefmt"
 	"github.com/stretchr/testify/assert"
@@ -19,6 +20,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/ruler/rulespb"
 	"github.com/cortexproject/cortex/pkg/ruler/rulestore"
+	cortextsdb "github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/storage/tsdb/testutil"
 )
 
@@ -105,7 +107,10 @@ func TestListRules(t *testing.T) {
 func TestLoadPartialRules(t *testing.T) {
 	bucketClient := objstore.NewInMemBucket()
 	mockedBucketClient := &testutil.MockBucketFailure{Bucket: bucketClient, GetFailures: map[string]error{}}
-	bucketStore := NewBucketRuleStore(mockedBucketClient, nil, log.NewNopLogger())
+	usersScannerConfig := cortextsdb.UsersScannerConfig{Strategy: cortextsdb.UserScanStrategyList}
+	reg := prometheus.NewPedanticRegistry()
+	bucketStore, err := NewBucketRuleStore(mockedBucketClient, usersScannerConfig, nil, log.NewNopLogger(), reg)
+	require.NoError(t, err)
 
 	groups := []testGroup{
 		{user: "user1", namespace: "hello", ruleGroup: rulefmt.RuleGroup{Name: "second testGroup", Interval: model.Duration(2 * time.Minute)}},
@@ -263,7 +268,10 @@ func TestDelete(t *testing.T) {
 
 func runForEachRuleStore(t *testing.T, testFn func(t *testing.T, store rulestore.RuleStore, bucketClient interface{})) {
 	bucketClient := objstore.NewInMemBucket()
-	bucketStore := NewBucketRuleStore(bucketClient, nil, log.NewNopLogger())
+	reg := prometheus.NewPedanticRegistry()
+	usersScannerConfig := cortextsdb.UsersScannerConfig{Strategy: cortextsdb.UserScanStrategyList}
+	bucketStore, err := NewBucketRuleStore(bucketClient, usersScannerConfig, nil, log.NewNopLogger(), reg)
+	assert.NoError(t, err)
 
 	stores := map[string]struct {
 		store  rulestore.RuleStore
@@ -426,8 +434,11 @@ func TestListAllRuleGroupsWithNoNamespaceOrGroup(t *testing.T) {
 			"rules/user3/bnM=/Z3JvdXAx", // namespace "ns", group "group1"
 		},
 	}
+	usersScannerConfig := cortextsdb.UsersScannerConfig{Strategy: cortextsdb.UserScanStrategyList}
+	reg := prometheus.NewPedanticRegistry()
 
-	s := NewBucketRuleStore(obj, nil, log.NewNopLogger())
+	s, err := NewBucketRuleStore(obj, usersScannerConfig, nil, log.NewNopLogger(), reg)
+	require.NoError(t, err)
 	out, err := s.ListAllRuleGroups(context.Background())
 	require.NoError(t, err)
 

pkg/ruler/rulestore/config.go

@@ -8,14 +8,16 @@ import (
 	"github.com/cortexproject/cortex/pkg/ruler/rulestore/configdb"
 	"github.com/cortexproject/cortex/pkg/ruler/rulestore/local"
 	"github.com/cortexproject/cortex/pkg/storage/bucket"
+	"github.com/cortexproject/cortex/pkg/storage/tsdb"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 )
 
 // Config configures a rule store.
 type Config struct {
 	bucket.Config `yaml:",inline"`
-	ConfigDB      client.Config `yaml:"configdb"`
-	Local         local.Config  `yaml:"local"`
+	ConfigDB      client.Config           `yaml:"configdb"`
+	Local         local.Config            `yaml:"local"`
+	UsersScanner  tsdb.UsersScannerConfig `yaml:"users_scanner"`
 }
 
 // RegisterFlags registers the backend storage config.
@@ -26,6 +28,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.ConfigDB.RegisterFlagsWithPrefix(prefix, f)
 	cfg.Local.RegisterFlagsWithPrefix(prefix, f)
 	cfg.RegisterFlagsWithPrefix(prefix, f)
+	cfg.UsersScanner.RegisterFlagsWithPrefix(prefix, f)
 }
 
 // IsDefaults returns true if the storage options have not been set.

pkg/ruler/storage.go

@@ -36,5 +36,5 @@ func NewRuleStore(ctx context.Context, cfg rulestore.Config, cfgProvider bucket.
 		return nil, err
 	}
 
-	return bucketclient.NewBucketRuleStore(bucketClient, cfgProvider, logger), nil
+	return bucketclient.NewBucketRuleStore(bucketClient, cfg.UsersScanner, cfgProvider, logger, reg)
 }