Merge branch 'cortexproject:master' into list-rules-filter

Author: rajagopalanand

CHANGELOG.md

@@ -38,6 +38,7 @@
 * [BUGFIX] Ring: Add JOINING state to read operation. #5346
 * [BUGFIX] Compactor: Partial block with only visit marker should be deleted even there is no deletion marker. #5342
 * [BUGFIX] KV: Etcd calls will no longer block indefinitely and will now time out after the DialTimeout period. #5392
+* [BUGFIX] Ring: Allow RF greater than number of zones to select more than one instance per zone #5411
 
 ## 1.15.1 2023-04-26
 

docs/contributing/how-integration-tests-work.md

@@ -20,13 +20,13 @@ This will locally build the `quay.io/cortexproject/cortex:latest` image used by
 Once the Docker image is built, you can run integration tests:
 
 ```
-go test -v -tags=integration,requires_docker,integration_alertmanager,integration_memberlist,integration_querier,integration_ruler,integration_query_fuzz ./integration/...
+go test -v -tags=requires_docker ./integration/...
 ```
 
-If you want to run a single test you can use a filter. For example, to only run `TestRulerMetricsWhenIngesterFails`:
+If you want to run a single test you can use a filter. For example, to only run `TestChunksStorageAllIndexBackends`:
 
 ```
-go test -v -tags=integration,requires_docker,integration_ruler ./integration/ -run "^TestRulerMetricsWhenIngesterFails$" -count=1
+go test -v -tags=requires_docker ./integration -run "^TestChunksStorageAllIndexBackends$"
 ```
 
 ### Supported environment variables
@@ -46,4 +46,4 @@ Integration tests have `requires_docker` tag (`// +build requires_docker` line f
 
 ## Isolation
 
-Each integration test runs in isolation. For each integration test, we do create a Docker network, start Cortex and its dependencies containers, push/query series to/from Cortex and run assertions on it. Once the test has done, both the Docker network and containers are terminated and deleted.
+Each integration test runs in isolation. For each integration test, we do create a Docker network, start Cortex and its dependencies containers, push/query series to/from Cortex and run assertions on it. Once the test has done, both the Docker network and containers are terminated and deleted.

pkg/ring/ring.go

@@ -345,17 +345,17 @@ func (r *Ring) Get(key uint32, op Operation, bufDescs []InstanceDesc, bufHosts,
 	}
 
 	var (
-		n          = r.cfg.ReplicationFactor
-		instances  = bufDescs[:0]
-		start      = searchToken(r.ringTokens, key)
-		iterations = 0
+		replicationFactor = r.cfg.ReplicationFactor
+		instances         = bufDescs[:0]
+		start             = searchToken(r.ringTokens, key)
+		iterations        = 0
 
 		// We use a slice instead of a map because it's faster to search within a
 		// slice than lookup a map for a very low number of items.
 		distinctHosts = bufHosts[:0]
 		distinctZones = bufZones[:0]
 	)
-	for i := start; len(distinctHosts) < n && iterations < len(r.ringTokens); i++ {
+	for i := start; len(distinctHosts) < replicationFactor && iterations < len(r.ringTokens); i++ {
 		iterations++
 		// Wrap i around in the ring.
 		i %= len(r.ringTokens)
@@ -385,11 +385,16 @@ func (r *Ring) Get(key uint32, op Operation, bufDescs []InstanceDesc, bufHosts,
 		// Check whether the replica set should be extended given we're including
 		// this instance.
 		if op.ShouldExtendReplicaSetOnState(instance.State) {
-			n++
+			replicationFactor++
 		} else if r.cfg.ZoneAwarenessEnabled && info.Zone != "" {
 			// We should only add the zone if we are not going to extend,
 			// as we want to extend the instance in the same AZ.
 			distinctZones = append(distinctZones, info.Zone)
+
+			if len(distinctZones) == len(r.ringZones) {
+				// reset the zones to repeatedly get hosts from distinct zones
+				distinctZones = distinctZones[:0]
+			}
 		}
 
 		instances = append(instances, instance)

pkg/ring/ring_test.go

@@ -378,19 +378,26 @@ func TestRing_Get_ZoneAwareness(t *testing.T) {
 			zoneAwarenessEnabled: true,
 			expectedInstances:    3,
 		},
-		"should fail if there are instances in 1 zone only on RF = 3": {
-			numInstances:         16,
-			numZones:             1,
+		"should fail if there are not enough instances": {
+			numInstances:         1,
+			numZones:             3,
 			replicationFactor:    3,
 			zoneAwarenessEnabled: true,
 			expectedErr:          "at least 2 live replicas required across different availability zones, could only find 1",
 		},
-		"should succeed if there are instances in 2 zones on RF = 3": {
+		"should succeed if there are instances in 3 zones on RF = 4": {
 			numInstances:         16,
-			numZones:             2,
-			replicationFactor:    3,
+			numZones:             3,
+			replicationFactor:    4,
+			zoneAwarenessEnabled: true,
+			expectedInstances:    4,
+		},
+		"should succeed if there are instances in 3 zones on RF = 9": {
+			numInstances:         16,
+			numZones:             3,
+			replicationFactor:    9,
 			zoneAwarenessEnabled: true,
-			expectedInstances:    2,
+			expectedInstances:    9,
 		},
 		"should succeed if there are instances in 1 zone only on RF = 3 but zone-awareness is disabled": {
 			numInstances:         16,
@@ -426,7 +433,7 @@ func TestRing_Get_ZoneAwareness(t *testing.T) {
 				ringTokens:          r.GetTokens(),
 				ringTokensByZone:    r.getTokensByZone(),
 				ringInstanceByToken: r.getTokensInfo(),
-				ringZones:           getZones(r.getTokensByZone()),
+				ringZones:           []string{"zone-1", "zone-2", "zone-3"},
 				strategy:            NewDefaultReplicationStrategy(),
 				KVClient:            &MockClient{},
 			}
@@ -447,26 +454,33 @@ func TestRing_Get_ZoneAwareness(t *testing.T) {
 				set, err = ring.Get(testValues[i], Write, instances, bufHosts, bufZones)
 				if testData.expectedErr != "" {
 					require.EqualError(t, err, testData.expectedErr)
+					continue // Skip the rest of the assertions if we were expecting an error.
 				} else {
 					require.NoError(t, err)
 				}
 
-				// Skip the rest of the assertions if we were expecting an error.
-				if testData.expectedErr != "" {
-					continue
-				}
-
 				// Check that we have the expected number of instances for replication.
 				assert.Equal(t, testData.expectedInstances, len(set.Instances))
 
 				// Ensure all instances are in a different zone (only if zone-awareness is enabled).
 				if testData.zoneAwarenessEnabled {
-					zones := make(map[string]struct{})
+					zones := make(map[string]int)
+					maxNumOfHostsPerZone := math.Ceil(float64(testData.replicationFactor) / float64(testData.numZones))
+
 					for i := 0; i < len(set.Instances); i++ {
-						if _, ok := zones[set.Instances[i].Zone]; ok {
-							t.Fatal("found multiple instances in the same zone")
+						if _, ok := zones[set.Instances[i].Zone]; !ok {
+							zones[set.Instances[i].Zone] = 1
+						} else {
+							zones[set.Instances[i].Zone]++
+
+							if zones[set.Instances[i].Zone] > int(maxNumOfHostsPerZone) {
+								t.Fatal("instances not spread across zones evenly")
+							}
 						}
-						zones[set.Instances[i].Zone] = struct{}{}
+					}
+
+					if testData.replicationFactor >= testData.numZones && len(zones) != testData.numZones {
+						t.Fatalf("each zone must have at least one instance")
 					}
 				}
 			}

pkg/storegateway/bucket_index_metadata_fetcher_test.go

@@ -25,6 +25,7 @@ import (
 )
 
 func TestBucketIndexMetadataFetcher_Fetch(t *testing.T) {
+	t.Parallel()
 	const userID = "user-1"
 
 	bkt, _ := cortex_testutil.PrepareFilesystemBucket(t)
@@ -99,6 +100,7 @@ func TestBucketIndexMetadataFetcher_Fetch(t *testing.T) {
 }
 
 func TestBucketIndexMetadataFetcher_Fetch_NoBucketIndex(t *testing.T) {
+	t.Parallel()
 	const userID = "user-1"
 
 	bkt, _ := cortex_testutil.PrepareFilesystemBucket(t)
@@ -150,6 +152,7 @@ func TestBucketIndexMetadataFetcher_Fetch_NoBucketIndex(t *testing.T) {
 }
 
 func TestBucketIndexMetadataFetcher_Fetch_CorruptedBucketIndex(t *testing.T) {
+	t.Parallel()
 	const userID = "user-1"
 
 	bkt, _ := cortex_testutil.PrepareFilesystemBucket(t)
@@ -204,6 +207,7 @@ func TestBucketIndexMetadataFetcher_Fetch_CorruptedBucketIndex(t *testing.T) {
 }
 
 func TestBucketIndexMetadataFetcher_Fetch_ShouldResetGaugeMetrics(t *testing.T) {
+	t.Parallel()
 	const userID = "user-1"
 
 	bkt, _ := cortex_testutil.PrepareFilesystemBucket(t)

pkg/storegateway/bucket_store_metrics_test.go

@@ -12,6 +12,7 @@ import (
 )
 
 func TestBucketStoreMetrics(t *testing.T) {
+	t.Parallel()
 	mainReg := prometheus.NewPedanticRegistry()
 
 	tsdbMetrics := NewBucketStoreMetrics()

pkg/storegateway/bucket_stores_test.go

@@ -41,6 +41,7 @@ import (
 )
 
 func TestBucketStores_InitialSync(t *testing.T) {
+	t.Parallel()
 	userToMetric := map[string]string{
 		"user-1": "series_1",
 		"user-2": "series_2",
@@ -180,6 +181,7 @@ func TestBucketStores_InitialSyncShouldRetryOnFailure(t *testing.T) {
 }
 
 func TestBucketStores_SyncBlocks(t *testing.T) {
+	t.Parallel()
 	const (
 		userID     = "user-1"
 		metricName = "series_1"
@@ -249,6 +251,7 @@ func TestBucketStores_SyncBlocks(t *testing.T) {
 }
 
 func TestBucketStores_syncUsersBlocks(t *testing.T) {
+	t.Parallel()
 	allUsers := []string{"user-1", "user-2", "user-3"}
 
 	tests := map[string]struct {

pkg/storegateway/chunk_bytes_pool_test.go

@@ -15,6 +15,7 @@ import (
 )
 
 func TestChunkBytesPool_Get(t *testing.T) {
+	t.Parallel()
 	reg := prometheus.NewPedanticRegistry()
 	p, err := newChunkBytesPool(cortex_tsdb.ChunkPoolDefaultMinBucketSize, cortex_tsdb.ChunkPoolDefaultMaxBucketSize, 0, reg)
 	require.NoError(t, err)

pkg/storegateway/gateway_ring_test.go

@@ -10,7 +10,7 @@ import (
 )
 
 func TestIsHealthyForStoreGatewayOperations(t *testing.T) {
-	t.Parallel()
+	//parallel testing causes data race
 
 	tests := map[string]struct {
 		instance          *ring.InstanceDesc
@@ -60,6 +60,7 @@ func TestIsHealthyForStoreGatewayOperations(t *testing.T) {
 		testData := testData
 
 		t.Run(testName, func(t *testing.T) {
+			t.Parallel()
 			actual := testData.instance.IsHealthy(BlocksOwnerSync, testData.timeout, time.Now())
 			assert.Equal(t, testData.ownerSyncExpected, actual)