Fix for alertmanagers not configuring users before becoming ACTIVE.

pkg/alertmanager/alertmanager_ring.go

@@ -32,6 +32,12 @@ var RingOp = ring.NewOp([]ring.InstanceState{ring.ACTIVE}, func(s ring.InstanceS
 	return s != ring.ACTIVE
 })
 
+// UserOwnedRingOp is the operation used for checking if a user is owned by an alertmanager.
+var UserOwnedRingOp = ring.NewOp([]ring.InstanceState{ring.ACTIVE, ring.JOINING}, func(s ring.InstanceState) bool {
+	// A user is owned by an alertmanager in both ACTIVE and JOINING states.
+	return (s != ring.ACTIVE && s != ring.JOINING)
+})
+
 // RingConfig masks the ring lifecycler config which contains
 // many options not really required by the alertmanager ring. This config
 // is used to strip down the config to the minimum, and avoid confusion

pkg/alertmanager/multitenant.go

@@ -710,7 +710,7 @@ func (am *MultitenantAlertmanager) isUserOwned(userID string) bool {
 		return true
 	}
 
-	alertmanagers, err := am.ring.Get(shardByUser(userID), RingOp, nil, nil, nil)
+	alertmanagers, err := am.ring.Get(shardByUser(userID), UserOwnedRingOp, nil, nil, nil)
 	if err != nil {
 		am.ringCheckErrors.Inc()
 		level.Error(am.logger).Log("msg", "failed to load alertmanager configuration", "user", userID, "err", err)

pkg/alertmanager/multitenant_test.go

@@ -15,6 +15,7 @@ import (
 	"path/filepath"
 	"regexp"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -1146,7 +1147,7 @@ func TestAlertmanager_StateReplicationWithSharding(t *testing.T) {
 				defer services.StopAndAwaitTerminated(ctx, am) //nolint:errcheck
 
 				if tt.withSharding {
-					clientPool.servers[amConfig.ShardingRing.InstanceAddr+":0"] = am
+					clientPool.setServer(amConfig.ShardingRing.InstanceAddr+":0", am)
 					am.alertmanagerClientsPool = clientPool
 				}
 
@@ -1334,7 +1335,7 @@ func TestAlertmanager_StateReplicationWithSharding_InitialSyncFromPeers(t *testi
 				am, err := createMultitenantAlertmanager(amConfig, nil, nil, mockStore, ringStore, log.NewNopLogger(), reg)
 				require.NoError(t, err)
 
-				clientPool.servers[amConfig.ShardingRing.InstanceAddr+":0"] = am
+				clientPool.setServer(amConfig.ShardingRing.InstanceAddr+":0", am)
 				am.alertmanagerClientsPool = clientPool
 
 				require.NoError(t, services.StartAndAwaitRunning(ctx, am))
@@ -1517,7 +1518,8 @@ func (am *passthroughAlertmanagerClient) RemoteAddress() string {
 // passthroughAlertmanagerClientPool allows testing the logic of gRPC calls between alertmanager instances
 // by invoking client calls directly to a peer instance in the unit test, without the server running.
 type passthroughAlertmanagerClientPool struct {
-	servers map[string]alertmanagerpb.AlertmanagerServer
+	serversMtx sync.Mutex
+	servers    map[string]alertmanagerpb.AlertmanagerServer
 }
 
 func newPassthroughAlertmanagerClientPool() *passthroughAlertmanagerClientPool {
@@ -1526,7 +1528,15 @@ func newPassthroughAlertmanagerClientPool() *passthroughAlertmanagerClientPool {
 	}
 }
 
+func (f *passthroughAlertmanagerClientPool) setServer(addr string, server alertmanagerpb.AlertmanagerServer) {
+	f.serversMtx.Lock()
+	defer f.serversMtx.Unlock()
+	f.servers[addr] = server
+}
+
 func (f *passthroughAlertmanagerClientPool) GetClientFor(addr string) (Client, error) {
+	f.serversMtx.Lock()
+	defer f.serversMtx.Unlock()
 	s, ok := f.servers[addr]
 	if !ok {
 		return nil, fmt.Errorf("client not found for address: %v", addr)