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
 

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)
 

pkg/storegateway/gateway_test.go

@@ -47,6 +47,7 @@ import (
 )
 
 func TestConfig_Validate(t *testing.T) {
+	t.Parallel()
 	tests := map[string]struct {
 		setup    func(cfg *Config, limits *validation.Limits)
 		expected error
@@ -80,7 +81,9 @@ func TestConfig_Validate(t *testing.T) {
 	}
 
 	for testName, testData := range tests {
+		testData := testData
 		t.Run(testName, func(t *testing.T) {
+			t.Parallel()
 			cfg := &Config{}
 			limits := &validation.Limits{}
 			flagext.DefaultValues(cfg, limits)
@@ -92,6 +95,7 @@ func TestConfig_Validate(t *testing.T) {
 }
 
 func TestStoreGateway_InitialSyncWithDefaultShardingEnabled(t *testing.T) {
+	t.Parallel()
 	tests := map[string]struct {
 		initialExists bool
 		initialState  ring.InstanceState
@@ -123,7 +127,9 @@ func TestStoreGateway_InitialSyncWithDefaultShardingEnabled(t *testing.T) {
 	}
 
 	for testName, testData := range tests {
+		testData := testData
 		t.Run(testName, func(t *testing.T) {
+			t.Parallel()
 			ctx := context.Background()
 			gatewayCfg := mockGatewayConfig()
 			gatewayCfg.ShardingEnabled = true
@@ -174,6 +180,7 @@ func TestStoreGateway_InitialSyncWithDefaultShardingEnabled(t *testing.T) {
 }
 
 func TestStoreGateway_InitialSyncWithShardingDisabled(t *testing.T) {
+	t.Parallel()
 	ctx := context.Background()
 	gatewayCfg := mockGatewayConfig()
 	gatewayCfg.ShardingEnabled = false
@@ -195,6 +202,7 @@ func TestStoreGateway_InitialSyncWithShardingDisabled(t *testing.T) {
 }
 
 func TestStoreGateway_InitialSyncFailure(t *testing.T) {
+	t.Parallel()
 	ctx := context.Background()
 	gatewayCfg := mockGatewayConfig()
 	gatewayCfg.ShardingEnabled = true
@@ -223,6 +231,7 @@ func TestStoreGateway_InitialSyncFailure(t *testing.T) {
 // their own blocks, regardless which store-gateway joined the ring first or last (even if starting
 // at the same time, they will join the ring at a slightly different time).
 func TestStoreGateway_InitialSyncWithWaitRingStability(t *testing.T) {
+	//parallel testing causes data race
 	bucketClient, storageDir := cortex_testutil.PrepareFilesystemBucket(t)
 
 	// This tests uses real TSDB blocks. 24h time range, 2h block range period,
@@ -302,6 +311,7 @@ func TestStoreGateway_InitialSyncWithWaitRingStability(t *testing.T) {
 	for testName, testData := range tests {
 		for _, bucketIndexEnabled := range []bool{true, false} {
 			t.Run(fmt.Sprintf("%s (bucket index enabled = %v)", testName, bucketIndexEnabled), func(t *testing.T) {
+				//parallel testing causes data race
 				// Randomise the seed but log it in case we need to reproduce the test on failure.
 				seed := time.Now().UnixNano()
 				rand.Seed(seed)
@@ -383,6 +393,7 @@ func TestStoreGateway_InitialSyncWithWaitRingStability(t *testing.T) {
 }
 
 func TestStoreGateway_BlocksSyncWithDefaultSharding_RingTopologyChangedAfterScaleUp(t *testing.T) {
+	t.Parallel()
 	const (
 		numUsers             = 2
 		numBlocks            = numUsers * 12
@@ -542,6 +553,7 @@ func TestStoreGateway_BlocksSyncWithDefaultSharding_RingTopologyChangedAfterScal
 }
 
 func TestStoreGateway_ShouldSupportLoadRingTokensFromFile(t *testing.T) {
+	t.Parallel()
 	tests := map[string]struct {
 		storedTokens      ring.Tokens
 		expectedNumTokens int
@@ -561,7 +573,9 @@ func TestStoreGateway_ShouldSupportLoadRingTokensFromFile(t *testing.T) {
 	}
 
 	for testName, testData := range tests {
+		testData := testData
 		t.Run(testName, func(t *testing.T) {
+			t.Parallel()
 			tokensFile, err := os.CreateTemp(os.TempDir(), "tokens-*")
 			require.NoError(t, err)
 			defer os.Remove(tokensFile.Name()) //nolint:errcheck
@@ -596,6 +610,7 @@ func TestStoreGateway_ShouldSupportLoadRingTokensFromFile(t *testing.T) {
 }
 
 func TestStoreGateway_SyncOnRingTopologyChanged(t *testing.T) {
+	t.Parallel()
 	registeredAt := time.Now()
 
 	tests := map[string]struct {
@@ -704,7 +719,9 @@ func TestStoreGateway_SyncOnRingTopologyChanged(t *testing.T) {
 	}
 
 	for testName, testData := range tests {
+		testData := testData
 		t.Run(testName, func(t *testing.T) {
+			t.Parallel()
 			ctx := context.Background()
 			gatewayCfg := mockGatewayConfig()
 			gatewayCfg.ShardingEnabled = true
@@ -764,6 +781,7 @@ func TestStoreGateway_SyncOnRingTopologyChanged(t *testing.T) {
 }
 
 func TestStoreGateway_RingLifecyclerShouldAutoForgetUnhealthyInstances(t *testing.T) {
+	t.Parallel()
 	const unhealthyInstanceID = "unhealthy-id"
 	const heartbeatTimeout = time.Minute
 
@@ -810,6 +828,7 @@ func TestStoreGateway_RingLifecyclerShouldAutoForgetUnhealthyInstances(t *testin
 }
 
 func TestStoreGateway_SeriesQueryingShouldRemoveExternalLabels(t *testing.T) {
+	t.Parallel()
 	ctx := context.Background()
 	logger := log.NewNopLogger()
 	userID := "user-1"
@@ -855,7 +874,9 @@ func TestStoreGateway_SeriesQueryingShouldRemoveExternalLabels(t *testing.T) {
 	}
 
 	for _, bucketIndexEnabled := range []bool{true, false} {
+		bucketIndexEnabled := bucketIndexEnabled
 		t.Run(fmt.Sprintf("bucket index enabled = %v", bucketIndexEnabled), func(t *testing.T) {
+			t.Parallel()
 			// Create a store-gateway used to query back the series from the blocks.
 			gatewayCfg := mockGatewayConfig()
 			gatewayCfg.ShardingEnabled = false
@@ -901,6 +922,7 @@ func TestStoreGateway_SeriesQueryingShouldRemoveExternalLabels(t *testing.T) {
 }
 
 func TestStoreGateway_SeriesQueryingShouldEnforceMaxChunksPerQueryLimit(t *testing.T) {
+	t.Parallel()
 	const chunksQueried = 10
 
 	tests := map[string]struct {
@@ -948,6 +970,7 @@ func TestStoreGateway_SeriesQueryingShouldEnforceMaxChunksPerQueryLimit(t *testi
 
 	for testName, testData := range tests {
 		t.Run(testName, func(t *testing.T) {
+			//parallel testing causes data race
 			// Customise the limits.
 			limits := defaultLimitsConfig()
 			limits.MaxChunksPerQuery = testData.limit
@@ -1036,6 +1059,7 @@ func TestStoreGateway_SeriesQueryingShouldEnforceMaxSeriesPerQueryLimit(t *testi
 
 	for testName, testData := range tests {
 		t.Run(testName, func(t *testing.T) {
+			//parallel testing causes data race
 			// Customise the limits.
 			limits := defaultLimitsConfig()
 			limits.MaxFetchedSeriesPerQuery = testData.limit