Add user scanner to Ruler/Alertmanager

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

Author: SungJin1212

docs/blocks-storage/querier.md

@@ -1940,6 +1940,11 @@ blocks_storage:
     # CLI flag: -blocks-storage.users-scanner.user-index.max-stale-period
     [max_stale_period: <duration> | default = 1h]
 
+    # How frequently user index file is updated, it only take effect when user
+    # scan stratehy is user_index.
+    # CLI flag: -blocks-storage.users-scanner.user-index.cleanup-interval
+    [clean_up_interval: <duration> | default = 15m]
+
     # TTL of the cached users. 0 disables caching and relies on caching at
     # bucket client level.
     # CLI flag: -blocks-storage.users-scanner.cache-ttl

docs/blocks-storage/store-gateway.md

@@ -2017,6 +2017,11 @@ blocks_storage:
     # CLI flag: -blocks-storage.users-scanner.user-index.max-stale-period
     [max_stale_period: <duration> | default = 1h]
 
+    # How frequently user index file is updated, it only take effect when user
+    # scan stratehy is user_index.
+    # CLI flag: -blocks-storage.users-scanner.user-index.cleanup-interval
+    [clean_up_interval: <duration> | default = 15m]
+
     # TTL of the cached users. 0 disables caching and relies on caching at
     # bucket client level.
     # CLI flag: -blocks-storage.users-scanner.cache-ttl

docs/configuration/config-file-reference.md

@@ -986,6 +986,26 @@ 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]
+
+  # How frequently user index file is updated, it only take effect when user
+  # scan stratehy is user_index.
+  # CLI flag: -alertmanager-storage.users-scanner.user-index.cleanup-interval
+  [clean_up_interval: <duration> | default = 15m]
+
+  # 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`
@@ -2602,6 +2622,11 @@ users_scanner:
   # CLI flag: -blocks-storage.users-scanner.user-index.max-stale-period
   [max_stale_period: <duration> | default = 1h]
 
+  # How frequently user index file is updated, it only take effect when user
+  # scan stratehy is user_index.
+  # CLI flag: -blocks-storage.users-scanner.user-index.cleanup-interval
+  [clean_up_interval: <duration> | default = 15m]
+
   # TTL of the cached users. 0 disables caching and relies on caching at bucket
   # client level.
   # CLI flag: -blocks-storage.users-scanner.cache-ttl
@@ -5832,6 +5857,26 @@ 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]
+
+  # How frequently user index file is updated, it only take effect when user
+  # scan stratehy is user_index.
+  # CLI flag: -ruler-storage.users-scanner.user-index.cleanup-interval
+  [clean_up_interval: <duration> | default = 15m]
+
+  # 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`

integration/alertmanager_test.go

@@ -68,6 +68,41 @@ func TestAlertmanager(t *testing.T) {
 	assertServiceMetricsPrefixes(t, AlertManager, alertmanager)
 }
 
+func TestAlertmanagerWithUserIndexUpdater(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	require.NoError(t, writeFileToSharedDir(s, "alertmanager_configs/user-1.yaml", []byte(cortexAlertmanagerUserConfigYaml)))
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	minio := e2edb.NewMinio(9000, alertsBucketName)
+	require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+	baseFlags := mergeFlags(AlertmanagerFlags(), AlertmanagerS3Flags())
+	flags := mergeFlags(baseFlags, AlertmanagerShardingFlags(consul.NetworkHTTPEndpoint(), 1), map[string]string{
+		"-alertmanager-storage.users-scanner.strategy":                    "user_index",
+		"-alertmanager-storage.users-scanner.user-index.cleanup-interval": "15s",
+		"-alertmanager.configs.poll-interval":                             "5s",
+	})
+
+	am := e2ecortex.NewAlertmanager(
+		"alertmanager",
+		flags,
+		"",
+	)
+
+	require.NoError(t, s.StartAndWaitReady(am))
+	// To make sure user index file is updated/scanned
+	require.NoError(t, am.WaitSumMetricsWithOptions(e2e.Greater(float64(0)), []string{"cortex_user_index_last_successful_update_timestamp_seconds"}),
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "component", "alertmanager")),
+	)
+	require.NoError(t, am.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(float64(1)), []string{"cortex_user_index_scan_succeeded_total"}),
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "component", "alertmanager")),
+	)
+}
+
 func TestAlertmanagerStoreAPI(t *testing.T) {
 	s, err := e2e.NewScenario(networkName)
 	require.NoError(t, err)

integration/ruler_test.go

@@ -142,6 +142,65 @@ func TestRulerAPI(t *testing.T) {
 	}
 }
 
+func TestRulerWithUserIndexUpdater(t *testing.T) {
+	s, err := e2e.NewScenario(networkName)
+	require.NoError(t, err)
+	defer s.Close()
+
+	// Start dependencies.
+	consul := e2edb.NewConsul()
+	minio := e2edb.NewMinio(9000, rulestoreBucketName)
+	require.NoError(t, s.StartAndWaitReady(consul, minio))
+
+	// Configure the ruler.
+	rulerFlags := mergeFlags(
+		BlocksStorageFlags(),
+		RulerFlags(),
+		RulerShardingFlags(consul.NetworkHTTPEndpoint()),
+		map[string]string{
+			"-ruler.sharding-strategy":                                 "shuffle-sharding",
+			"-ruler-storage.users-scanner.strategy":                    "user_index",
+			"-ruler-storage.users-scanner.user-index.cleanup-interval": "15s",
+			"-ruler.tenant-shard-size":                                 "1",
+			// Since we're not going to run any rule, we don't need the
+			// store-gateway to be configured to a valid address.
+			"-querier.store-gateway-addresses": "localhost:12345",
+			// Enable the bucket index so we can skip the initial bucket scan.
+			"-blocks-storage.bucket-store.bucket-index.enabled": "true",
+			"-ruler.poll-interval":                              "2s",
+			"-log.level":                                        "info",
+		},
+	)
+
+	ruler := e2ecortex.NewRuler(
+		"ruler",
+		consul.NetworkHTTPEndpoint(),
+		rulerFlags,
+		"",
+	)
+
+	require.NoError(t, s.StartAndWaitReady(ruler))
+
+	// Create a client with the ruler address configured
+	c, err := e2ecortex.NewClient("", "", "", ruler.HTTPEndpoint(), "user-1")
+	require.NoError(t, err)
+
+	ruleGroup := createTestRuleGroup(t)
+	ns := "ns"
+
+	// Set the rule group into the ruler
+	require.NoError(t, c.SetRuleGroup(ruleGroup, ns))
+
+	// To make sure user index file is updated/scanned
+	require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.Greater(float64(0)), []string{"cortex_user_index_last_successful_update_timestamp_seconds"}),
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "component", "ruler")),
+	)
+
+	require.NoError(t, ruler.WaitSumMetricsWithOptions(e2e.GreaterOrEqual(float64(1)), []string{"cortex_user_index_scan_succeeded_total"}),
+		e2e.WithLabelMatchers(labels.MustNewMatcher(labels.MatchEqual, "component", "ruler")),
+	)
+}
+
 func TestRulerAPISingleBinary(t *testing.T) {
 	s, err := e2e.NewScenario(networkName)
 	require.NoError(t, err)

pkg/alertmanager/multitenant.go

@@ -33,12 +33,12 @@ import (
 	"github.com/cortexproject/cortex/pkg/ring"
 	"github.com/cortexproject/cortex/pkg/ring/client"
 	"github.com/cortexproject/cortex/pkg/ring/kv"
-	"github.com/cortexproject/cortex/pkg/tenant"
 	"github.com/cortexproject/cortex/pkg/util"
 	"github.com/cortexproject/cortex/pkg/util/concurrency"
 	"github.com/cortexproject/cortex/pkg/util/flagext"
 	util_log "github.com/cortexproject/cortex/pkg/util/log"
 	"github.com/cortexproject/cortex/pkg/util/services"
+	"github.com/cortexproject/cortex/pkg/util/users"
 )
 
 const (
@@ -91,6 +91,8 @@ type MultitenantAlertmanagerConfig struct {
 
 	EnabledTenants  flagext.StringSliceCSV `yaml:"enabled_tenants"`
 	DisabledTenants flagext.StringSliceCSV `yaml:"disabled_tenants"`
+
+	CleanUpInterval time.Duration `yaml:"-"`
 }
 
 type ClusterConfig struct {
@@ -284,14 +286,16 @@ type MultitenantAlertmanager struct {
 
 	limits Limits
 
-	allowedTenants *util.AllowedTenants
+	allowedTenants *users.AllowedTenants
 
 	registry          prometheus.Registerer
 	ringCheckErrors   prometheus.Counter
 	tenantsOwned      prometheus.Gauge
 	tenantsDiscovered prometheus.Gauge
 	syncTotal         *prometheus.CounterVec
 	syncFailures      *prometheus.CounterVec
+
+	userIndexUpdater *users.UserIndexUpdater
 }
 
 // NewMultitenantAlertmanager creates a new MultitenantAlertmanager.
@@ -374,10 +378,11 @@ func createMultitenantAlertmanager(cfg *MultitenantAlertmanagerConfig, fallbackC
 		multitenantMetrics:  newMultitenantAlertmanagerMetrics(registerer),
 		peer:                peer,
 		store:               store,
+		userIndexUpdater:    store.GetUserIndexUpdater(),
 		logger:              log.With(logger, "component", "MultiTenantAlertmanager"),
 		registry:            registerer,
 		limits:              limits,
-		allowedTenants:      util.NewAllowedTenants(cfg.EnabledTenants, cfg.DisabledTenants),
+		allowedTenants:      users.NewAllowedTenants(cfg.EnabledTenants, cfg.DisabledTenants),
 		ringCheckErrors: promauto.With(registerer).NewCounter(prometheus.CounterOpts{
 			Name: "cortex_alertmanager_ring_check_errors_total",
 			Help: "Number of errors that have occurred when checking the ring for ownership.",
@@ -667,6 +672,10 @@ func (am *MultitenantAlertmanager) run(ctx context.Context) error {
 		ringTickerChan = ringTicker.C
 	}
 
+	if am.cfg.ShardingEnabled && am.userIndexUpdater != nil {
+		go am.userIndexUpdateLoop(ctx)
+	}
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -693,6 +702,32 @@ func (am *MultitenantAlertmanager) run(ctx context.Context) error {
 	}
 }
 
+func (am *MultitenantAlertmanager) userIndexUpdateLoop(ctx context.Context) {
+	// Hardcode ID to check which alertmanager owns updating user index.
+	userID := users.UserIndexCompressedFilename
+	// Align with clean up interval.
+	ticker := time.NewTicker(util.DurationWithJitter(am.userIndexUpdater.GetCleanUpInterval(), 0.1))
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			level.Error(am.logger).Log("msg", "context timeout, exit user index update loop", "err", ctx.Err())
+			return
+		case <-ticker.C:
+			owned := am.isUserOwned(userID)
+			if !owned {
+				continue
+			}
+			if err := am.userIndexUpdater.UpdateUserIndex(ctx); err != nil {
+				level.Error(am.logger).Log("msg", "failed to update user index", "err", err)
+				// Wait for next interval. Worst case, the user index scanner will fallback to list strategy.
+				continue
+			}
+		}
+	}
+}
+
 func (am *MultitenantAlertmanager) loadAndSyncConfigs(ctx context.Context, syncReason string) error {
 	level.Info(am.logger).Log("msg", "synchronizing alertmanager configs for users")
 	am.syncTotal.WithLabelValues(syncReason).Inc()
@@ -795,7 +830,7 @@ func (am *MultitenantAlertmanager) isUserOwned(userID string) bool {
 		return true
 	}
 
-	alertmanagers, err := am.ring.Get(shardByUser(userID), SyncRingOp, nil, nil, nil)
+	alertmanagers, err := am.ring.Get(users.ShardByUser(userID), SyncRingOp, nil, nil, nil)
 	if err != nil {
 		am.ringCheckErrors.Inc()
 		level.Error(am.logger).Log("msg", "failed to load alertmanager configuration", "user", userID, "err", err)
@@ -1005,7 +1040,7 @@ func (am *MultitenantAlertmanager) GetPositionForUser(userID string) int {
 		return 0
 	}
 
-	set, err := am.ring.Get(shardByUser(userID), RingOp, nil, nil, nil)
+	set, err := am.ring.Get(users.ShardByUser(userID), RingOp, nil, nil, nil)
 	if err != nil {
 		level.Error(am.logger).Log("msg", "unable to read the ring while trying to determine the alertmanager position", "err", err)
 		// If we're  unable to determine the position, we don't want a tenant to miss out on the notification - instead,
@@ -1048,7 +1083,7 @@ func (am *MultitenantAlertmanager) HandleRequest(ctx context.Context, in *httpgr
 
 // serveRequest serves the Alertmanager's web UI and API.
 func (am *MultitenantAlertmanager) serveRequest(w http.ResponseWriter, req *http.Request) {
-	userID, err := tenant.TenantID(req.Context())
+	userID, err := users.TenantID(req.Context())
 	if err != nil {
 		http.Error(w, err.Error(), http.StatusUnauthorized)
 		return
@@ -1106,7 +1141,7 @@ func (am *MultitenantAlertmanager) ReplicateStateForUser(ctx context.Context, us
 	level.Debug(am.logger).Log("msg", "message received for replication", "user", userID, "key", part.Key)
 
 	selfAddress := am.ringLifecycler.GetInstanceAddr()
-	err := ring.DoBatch(ctx, RingOp, am.ring, nil, []uint32{shardByUser(userID)}, func(desc ring.InstanceDesc, _ []int) error {
+	err := ring.DoBatch(ctx, RingOp, am.ring, nil, []uint32{users.ShardByUser(userID)}, func(desc ring.InstanceDesc, _ []int) error {
 		if desc.GetAddr() == selfAddress {
 			return nil
 		}
@@ -1137,7 +1172,7 @@ func (am *MultitenantAlertmanager) ReplicateStateForUser(ctx context.Context, us
 // state from all replicas, but will consider it a success if state is obtained from at least one replica.
 func (am *MultitenantAlertmanager) ReadFullStateForUser(ctx context.Context, userID string) ([]*clusterpb.FullState, error) {
 	// Only get the set of replicas which contain the specified user.
-	key := shardByUser(userID)
+	key := users.ShardByUser(userID)
 	replicationSet, err := am.ring.Get(key, RingOp, nil, nil, nil)
 	if err != nil {
 		return nil, err
@@ -1197,7 +1232,7 @@ func (am *MultitenantAlertmanager) ReadFullStateForUser(ctx context.Context, use
 
 // UpdateState implements the Alertmanager service.
 func (am *MultitenantAlertmanager) UpdateState(ctx context.Context, part *clusterpb.Part) (*alertmanagerpb.UpdateStateResponse, error) {
-	userID, err := tenant.TenantID(ctx)
+	userID, err := users.TenantID(ctx)
 	if err != nil {
 		return nil, err
 	}
@@ -1307,7 +1342,7 @@ func (am *MultitenantAlertmanager) getPerUserDirectories() map[string]string {
 
 // UpdateState implements the Alertmanager service.
 func (am *MultitenantAlertmanager) ReadState(ctx context.Context, req *alertmanagerpb.ReadStateRequest) (*alertmanagerpb.ReadStateResponse, error) {
-	userID, err := tenant.TenantID(ctx)
+	userID, err := users.TenantID(ctx)
 	if err != nil {
 		return nil, err
 	}