Fix TestAlertmanager_StateReplicationWithSharding. (#4078)

* Fix TestAlertmanager_StateReplicationWithSharding.

The test had two issues:

1. The most common issue was that replicating the silence sometimes
   took a small amount of time, as it happens asynchronously. Replacing
   Equal with Eventually when checking the replication metrics is enough
   to make the tests reliable.

2. The second issue was more rare: When a particular shard does not have
   any users assigned to it, _and_ the test randomly picks _that_ shard
   to write the silence to, the test would panic (in rand.Intn). Due to
   the alertmanagersMtx being held, the test actually just hung up while
   trying to shutdown the alertmanagers.

Signed-off-by: Steve Simpson <[email protected]>

* Review comments.

Signed-off-by: Steve Simpson <[email protected]>

Author: stevesg

pkg/alertmanager/multitenant_test.go

@@ -1049,8 +1049,6 @@ func TestAlertmanager_ReplicasPosition(t *testing.T) {
 }
 
 func TestAlertmanager_StateReplicationWithSharding(t *testing.T) {
-	t.Skip("Flaky test under investigation")
-
 	tc := []struct {
 		name              string
 		replicationFactor int
@@ -1161,10 +1159,22 @@ func TestAlertmanager_StateReplicationWithSharding(t *testing.T) {
 
 			// With sharding enabled, we propagate messages over gRPC instead of using a gossip over TCP.
 			// 1. First, get a random multitenant instance
-			multitenantAM := instances[rand.Intn(len(instances))]
+			//    We must pick an instance which actually has a user configured.
+			var multitenantAM *MultitenantAlertmanager
+			for {
+				multitenantAM = instances[rand.Intn(len(instances))]
+
+				multitenantAM.alertmanagersMtx.Lock()
+				amount := len(multitenantAM.alertmanagers)
+				multitenantAM.alertmanagersMtx.Unlock()
+				if amount > 0 {
+					break
+				}
+			}
 
 			// 2. Then, get a random user that exists in that particular alertmanager instance.
 			multitenantAM.alertmanagersMtx.Lock()
+			require.Greater(t, len(multitenantAM.alertmanagers), 0)
 			k := rand.Intn(len(multitenantAM.alertmanagers))
 			var userID string
 			for u := range multitenantAM.alertmanagers {
@@ -1202,18 +1212,26 @@ func TestAlertmanager_StateReplicationWithSharding(t *testing.T) {
 				require.Regexp(t, regexp.MustCompile(`{"silenceID":".+"}`), string(body))
 			}
 
-			metrics := registries.BuildMetricFamiliesPerUser()
-
 			// If sharding is not enabled, we never propagate any messages amongst replicas in this way, and we can stop here.
 			if !tt.withSharding {
+				metrics := registries.BuildMetricFamiliesPerUser()
+
 				assert.Equal(t, float64(1), metrics.GetSumOfGauges("cortex_alertmanager_silences"))
 				assert.Equal(t, float64(0), metrics.GetSumOfCounters("cortex_alertmanager_state_replication_total"))
 				assert.Equal(t, float64(0), metrics.GetSumOfCounters("cortex_alertmanager_state_replication_failed_total"))
 				return
 			}
 
+			var metrics util.MetricFamiliesPerUser
+
 			// 5. Then, make sure it is propagated successfully.
-			assert.Equal(t, float64(tt.replicationFactor), metrics.GetSumOfGauges("cortex_alertmanager_silences"))
+			//    Replication is asynchronous, so we may have to wait a short period of time.
+			assert.Eventually(t, func() bool {
+				metrics = registries.BuildMetricFamiliesPerUser()
+				return (float64(tt.replicationFactor) == metrics.GetSumOfGauges("cortex_alertmanager_silences") &&
+					float64(tt.replicationFactor) == metrics.GetSumOfCounters("cortex_alertmanager_state_replication_total"))
+			}, 5*time.Second, 100*time.Millisecond)
+
 			assert.Equal(t, float64(tt.replicationFactor), metrics.GetSumOfCounters("cortex_alertmanager_state_replication_total"))
 			assert.Equal(t, float64(0), metrics.GetSumOfCounters("cortex_alertmanager_state_replication_failed_total"))
 
@@ -1224,7 +1242,12 @@ func TestAlertmanager_StateReplicationWithSharding(t *testing.T) {
 			//   For RF=3 1 -> 2 -> 4 = Total 6 merges
 			nFanOut := tt.replicationFactor - 1
 			nMerges := nFanOut + (nFanOut * nFanOut)
-			assert.Equal(t, float64(nMerges), metrics.GetSumOfCounters("cortex_alertmanager_partial_state_merges_total"))
+
+			assert.Eventually(t, func() bool {
+				metrics = registries.BuildMetricFamiliesPerUser()
+				return float64(nMerges) == metrics.GetSumOfCounters("cortex_alertmanager_partial_state_merges_total")
+			}, 5*time.Second, 100*time.Millisecond)
+
 			assert.Equal(t, float64(0), metrics.GetSumOfCounters("cortex_alertmanager_partial_state_merges_failed_total"))
 		})
 	}