Add index updater to alertmanager

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

Author: SungJin1212

docs/blocks-storage/querier.md

@@ -1927,6 +1927,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

@@ -2013,6 +2013,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

@@ -992,6 +992,11 @@ users_scanner:
   # 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
@@ -2608,6 +2613,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
@@ -5796,6 +5806,11 @@ users_scanner:
   # 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

integration/alertmanager_test.go

@@ -69,6 +69,42 @@ 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",
+		"-alertmanager.sharding-enabled":                                  "true",
+	})
+
+	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)

pkg/alertmanager/alertstore/bucketclient/bucket_client.go

@@ -48,25 +48,40 @@ type BucketAlertStore struct {
 	cfgProvider  bucket.TenantConfigProvider
 	logger       log.Logger
 
-	usersScanner users.Scanner
+	usersScanner     users.Scanner
+	userIndexUpdater *users.UserIndexUpdater
 }
 
 func NewBucketAlertStore(bkt objstore.InstrumentedBucket, userScannerCfg users.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))
+	regWithComponent := extprom.WrapRegistererWith(prometheus.Labels{"component": "alertmanager"}, reg)
+	usersScanner, err := users.NewScanner(userScannerCfg, alertBucket, logger, regWithComponent)
 	if err != nil {
 		return nil, errors.Wrap(err, "unable to initialize alertmanager users scanner")
 	}
+
+	// We hardcode strategy to be list so can ignore error.
+	baseScanner, _ := users.NewScanner(users.UsersScannerConfig{
+		Strategy: users.UserScanStrategyList,
+	}, alertBucket, logger, regWithComponent)
+	userIndexUpdater := users.NewUserIndexUpdater(alertBucket, userScannerCfg.CleanUpInterval, baseScanner, regWithComponent)
+
 	return &BucketAlertStore{
-		usersScanner: usersScanner,
-		alertsBucket: alertBucket,
-		amBucket:     bucket.NewPrefixedBucketClient(bkt, alertmanagerPrefix),
-		cfgProvider:  cfgProvider,
-		logger:       logger,
+		usersScanner:     usersScanner,
+		userIndexUpdater: userIndexUpdater,
+		alertsBucket:     alertBucket,
+		amBucket:         bucket.NewPrefixedBucketClient(bkt, alertmanagerPrefix),
+		cfgProvider:      cfgProvider,
+		logger:           logger,
 	}, nil
 }
 
+// GetUserIndexUpdater implements alertstore.AlertStore.
+func (s *BucketAlertStore) GetUserIndexUpdater() *users.UserIndexUpdater {
+	return s.userIndexUpdater
+}
+
 // ListAllUsers implements alertstore.AlertStore.
 func (s *BucketAlertStore) ListAllUsers(ctx context.Context) ([]string, error) {
 	active, deleting, _, err := s.usersScanner.ScanUsers(ctx)

pkg/alertmanager/alertstore/configdb/store.go

@@ -7,6 +7,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/alertmanager/alertspb"
 	"github.com/cortexproject/cortex/pkg/configs/client"
 	"github.com/cortexproject/cortex/pkg/configs/userconfig"
+	"github.com/cortexproject/cortex/pkg/util/users"
 )
 
 const (
@@ -34,6 +35,11 @@ func NewStore(c client.Client) *Store {
 	}
 }
 
+// GetUserIndexUpdater implements alertstore.AlertStore.
+func (c *Store) GetUserIndexUpdater() *users.UserIndexUpdater {
+	return nil
+}
+
 // ListAllUsers implements alertstore.AlertStore.
 func (c *Store) ListAllUsers(ctx context.Context) ([]string, error) {
 	configs, err := c.reloadConfigs(ctx)

pkg/alertmanager/alertstore/local/store.go

@@ -7,6 +7,7 @@ import (
 	"path/filepath"
 	"strings"
 
+	"github.com/cortexproject/cortex/pkg/util/users"
 	"github.com/pkg/errors"
 	"github.com/prometheus/alertmanager/config"
 
@@ -43,6 +44,11 @@ func NewStore(cfg StoreConfig) (*Store, error) {
 	return &Store{cfg}, nil
 }
 
+// GetUserIndexUpdater implements alertstore.AlertStore.
+func (f *Store) GetUserIndexUpdater() *users.UserIndexUpdater {
+	return nil
+}
+
 // ListAllUsers implements alertstore.AlertStore.
 func (f *Store) ListAllUsers(_ context.Context) ([]string, error) {
 	configs, err := f.reloadConfigs()

pkg/alertmanager/alertstore/store.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 
+	"github.com/cortexproject/cortex/pkg/util/users"
 	"github.com/go-kit/log"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/thanos-io/objstore"
@@ -53,6 +54,9 @@ type AlertStore interface {
 	// DeleteFullState deletes the alertmanager state for an user.
 	// If state for the user doesn't exist, no error is reported.
 	DeleteFullState(ctx context.Context, user string) error
+
+	// GetUserIndexUpdater is getter for UserIndexUpdater
+	GetUserIndexUpdater() *users.UserIndexUpdater
 }
 
 // NewAlertStore returns a alertmanager store backend client based on the provided cfg.

pkg/alertmanager/api_test.go

@@ -853,7 +853,7 @@ receivers:
 
 	// Create the Multitenant Alertmanager.
 	cfg := mockAlertmanagerConfig(t)
-	am, err := createMultitenantAlertmanager(cfg, nil, nil, alertStore, nil, nil, log.NewNopLogger(), reg)
+	am, err := createMultitenantAlertmanager(cfg, nil, nil, nil, alertStore, nil, nil, log.NewNopLogger(), reg)
 	require.NoError(t, err)
 	require.NoError(t, services.StartAndAwaitRunning(context.Background(), am))
 	defer services.StopAndAwaitTerminated(context.Background(), am) //nolint:errcheck

pkg/alertmanager/distributor.go

@@ -2,7 +2,6 @@ package alertmanager
 
 import (
 	"context"
-	"hash/fnv"
 	"io"
 	"math/rand"
 	"net/http"
@@ -161,7 +160,7 @@ func (d *Distributor) doQuorum(userID string, w http.ResponseWriter, r *http.Req
 	var responses []*httpgrpc.HTTPResponse
 	var responsesMtx sync.Mutex
 	grpcHeaders := httpToHttpgrpcHeaders(r.Header)
-	err = ring.DoBatch(r.Context(), RingOp, d.alertmanagerRing, nil, []uint32{shardByUser(userID)}, func(am ring.InstanceDesc, _ []int) error {
+	err = ring.DoBatch(r.Context(), RingOp, d.alertmanagerRing, nil, []uint32{users.ShardByUser(userID)}, func(am ring.InstanceDesc, _ []int) error {
 		// Use a background context to make sure all alertmanagers get the request even if we return early.
 		localCtx := opentracing.ContextWithSpan(user.InjectOrgID(context.Background(), userID), opentracing.SpanFromContext(r.Context()))
 		sp, localCtx := opentracing.StartSpanFromContext(localCtx, "Distributor.doQuorum")
@@ -207,7 +206,7 @@ func (d *Distributor) doQuorum(userID string, w http.ResponseWriter, r *http.Req
 }
 
 func (d *Distributor) doUnary(userID string, w http.ResponseWriter, r *http.Request, logger log.Logger) {
-	key := shardByUser(userID)
+	key := users.ShardByUser(userID)
 	replicationSet, err := d.alertmanagerRing.Get(key, RingOp, nil, nil, nil)
 	if err != nil {
 		level.Error(logger).Log("msg", "failed to get replication set from the ring", "err", err)
@@ -299,13 +298,6 @@ func (d *Distributor) doRequest(ctx context.Context, am ring.InstanceDesc, req *
 	return amClient.HandleRequest(ctx, req)
 }
 
-func shardByUser(userID string) uint32 {
-	ringHasher := fnv.New32a()
-	// Hasher never returns err.
-	_, _ = ringHasher.Write([]byte(userID))
-	return ringHasher.Sum32()
-}
-
 func httpToHttpgrpcHeaders(hs http.Header) []*httpgrpc.Header {
 	result := make([]*httpgrpc.Header, 0, len(hs))
 	for k, vs := range hs {