Add new ring method to get all instances and created a new method

in ruler to get Replicaset without requiring quorum

Signed-off-by: Emmanuel Lodovice <[email protected]>

Author: emanlodovice

docs/configuration/config-file-reference.md

@@ -4268,7 +4268,7 @@ ring:
 # with default state (state before any evaluation) and send this copy in list
 # API requests as backup in case the ruler who owns the rule fails to send its
 # rules. This allows the rules API to handle ruler outage by returning rules
-# with default state. Ring replication-factor needs to be set to 3 or more for
+# with default state. Ring replication-factor needs to be set to 2 or more for
 # this to be useful.
 # CLI flag: -experimental.ruler.api-enable-rules-backup
 [api_enable_rules_backup: <boolean> | default = false]

pkg/compactor/shuffle_sharding_grouper_test.go

@@ -766,14 +766,14 @@ func (r *RingMock) GetInstanceDescsForOperation(op ring.Operation) (map[string]r
 	return args.Get(0).(map[string]ring.InstanceDesc), args.Error(1)
 }
 
-func (r *RingMock) GetReplicationSetForOperation(op ring.Operation) (ring.ReplicationSet, error) {
+func (r *RingMock) GetAllInstanceDescs(op ring.Operation) ([]ring.InstanceDesc, []ring.InstanceDesc, error) {
 	args := r.Called(op)
-	return args.Get(0).(ring.ReplicationSet), args.Error(1)
+	return args.Get(0).([]ring.InstanceDesc), make([]ring.InstanceDesc, 0), args.Error(1)
 }
 
-func (r *RingMock) GetReplicationSetForOperationWithNoQuorum(op ring.Operation) (ring.ReplicationSet, map[string]struct{}, error) {
+func (r *RingMock) GetReplicationSetForOperation(op ring.Operation) (ring.ReplicationSet, error) {
 	args := r.Called(op)
-	return args.Get(0).(ring.ReplicationSet), make(map[string]struct{}), args.Error(1)
+	return args.Get(0).(ring.ReplicationSet), args.Error(1)
 }
 
 func (r *RingMock) ReplicationFactor() int {

pkg/ring/ring.go

@@ -50,6 +50,9 @@ type ReadRing interface {
 	// of unhealthy instances is greater than the tolerated max unavailable.
 	GetAllHealthy(op Operation) (ReplicationSet, error)
 
+	// GetAllInstanceDescs returns a slice of healthy and unhealthy InstanceDesc.
+	GetAllInstanceDescs(op Operation) ([]InstanceDesc, []InstanceDesc, error)
+
 	// GetInstanceDescsForOperation returns map of InstanceDesc with instance ID as the keys.
 	GetInstanceDescsForOperation(op Operation) (map[string]InstanceDesc, error)
 
@@ -59,13 +62,6 @@ type ReadRing interface {
 	// the input operation.
 	GetReplicationSetForOperation(op Operation) (ReplicationSet, error)
 
-	// GetReplicationSetForOperationWithNoQuorum returns all instances where the input operation should be executed.
-	// The resulting ReplicationSet contains all healthy instances in the ring, but the computation for MaxErrors
-	// does not require quorum so only 1 replica is needed to complete the operation. For MaxUnavailableZones, it is
-	// not automatically reduced when there are unhealthy instances in a zone because healthy instances in the zone
-	// are still returned, but the information about zones with unhealthy instances is returned.
-	GetReplicationSetForOperationWithNoQuorum(op Operation) (ReplicationSet, map[string]struct{}, error)
-
 	ReplicationFactor() int
 
 	// InstancesCount returns the number of instances in the ring.
@@ -471,6 +467,28 @@ func (r *Ring) GetAllHealthy(op Operation) (ReplicationSet, error) {
 	}, nil
 }
 
+// GetAllInstanceDescs implements ReadRing.
+func (r *Ring) GetAllInstanceDescs(op Operation) ([]InstanceDesc, []InstanceDesc, error) {
+	r.mtx.RLock()
+	defer r.mtx.RUnlock()
+
+	if r.ringDesc == nil || len(r.ringDesc.Ingesters) == 0 {
+		return nil, nil, ErrEmptyRing
+	}
+	healthyInstances := make([]InstanceDesc, 0, len(r.ringDesc.Ingesters))
+	unhealthyInstances := make([]InstanceDesc, 0, len(r.ringDesc.Ingesters))
+	storageLastUpdate := r.KVClient.LastUpdateTime(r.key)
+	for _, instance := range r.ringDesc.Ingesters {
+		if r.IsHealthy(&instance, op, storageLastUpdate) {
+			healthyInstances = append(healthyInstances, instance)
+		} else {
+			unhealthyInstances = append(unhealthyInstances, instance)
+		}
+	}
+
+	return healthyInstances, unhealthyInstances, nil
+}
+
 // GetInstanceDescsForOperation implements ReadRing.
 func (r *Ring) GetInstanceDescsForOperation(op Operation) (map[string]InstanceDesc, error) {
 	r.mtx.RLock()
@@ -491,12 +509,13 @@ func (r *Ring) GetInstanceDescsForOperation(op Operation) (map[string]InstanceDe
 	return instanceDescs, nil
 }
 
-func (r *Ring) getReplicationSetForOperation(op Operation, requireQuorum bool) (ReplicationSet, map[string]struct{}, error) {
+// GetReplicationSetForOperation implements ReadRing.
+func (r *Ring) GetReplicationSetForOperation(op Operation) (ReplicationSet, error) {
 	r.mtx.RLock()
 	defer r.mtx.RUnlock()
 
 	if r.ringDesc == nil || len(r.ringTokens) == 0 {
-		return ReplicationSet{}, make(map[string]struct{}), ErrEmptyRing
+		return ReplicationSet{}, ErrEmptyRing
 	}
 
 	// Build the initial replication set, excluding unhealthy instances.
@@ -518,24 +537,18 @@ func (r *Ring) getReplicationSetForOperation(op Operation, requireQuorum bool) (
 	maxUnavailableZones := 0
 
 	if r.cfg.ZoneAwarenessEnabled {
+		// Given data is replicated to RF different zones, we can tolerate a number of
+		// RF/2 failing zones. However, we need to protect from the case the ring currently
+		// contains instances in a number of zones < RF.
 		numReplicatedZones := utilmath.Min(len(r.ringZones), r.cfg.ReplicationFactor)
-		if requireQuorum {
-			// Given data is replicated to RF different zones, we can tolerate a number of
-			// RF/2 failing zones. However, we need to protect from the case the ring currently
-			// contains instances in a number of zones < RF.
-			minSuccessZones := (numReplicatedZones / 2) + 1
-			maxUnavailableZones = minSuccessZones - 1
-		} else {
-			// Given that quorum is not required, we only need at least one of the zone to be healthy to succeed. But we
-			// also need to handle case when RF < number of zones.
-			maxUnavailableZones = numReplicatedZones - 1
-		}
+		minSuccessZones := (numReplicatedZones / 2) + 1
+		maxUnavailableZones = minSuccessZones - 1
 
 		if len(zoneFailures) > maxUnavailableZones {
-			return ReplicationSet{}, zoneFailures, ErrTooManyUnhealthyInstances
+			return ReplicationSet{}, ErrTooManyUnhealthyInstances
 		}
 
-		if requireQuorum && len(zoneFailures) > 0 {
+		if len(zoneFailures) > 0 {
 			// We remove all instances (even healthy ones) from zones with at least
 			// 1 failing instance. Due to how replication works when zone-awareness is
 			// enabled (data is replicated to RF different zones), there's no benefit in
@@ -549,11 +562,11 @@ func (r *Ring) getReplicationSetForOperation(op Operation, requireQuorum bool) (
 			}
 
 			healthyInstances = filteredInstances
-
-			// Since we removed all instances from zones containing at least 1 failing
-			// instance, we have to decrease the max unavailable zones accordingly.
-			maxUnavailableZones -= len(zoneFailures)
 		}
+
+		// Since we removed all instances from zones containing at least 1 failing
+		// instance, we have to decrease the max unavailable zones accordingly.
+		maxUnavailableZones -= len(zoneFailures)
 	} else {
 		// Calculate the number of required instances;
 		// ensure we always require at least RF-1 when RF=3.
@@ -562,15 +575,10 @@ func (r *Ring) getReplicationSetForOperation(op Operation, requireQuorum bool) (
 			numRequired = r.cfg.ReplicationFactor
 		}
 		// We can tolerate this many failures
-		if requireQuorum {
-			numRequired -= r.cfg.ReplicationFactor / 2
-		} else {
-			// if quorum is not required then 1 replica is enough to handle the request
-			numRequired -= r.cfg.ReplicationFactor - 1
-		}
+		numRequired -= r.cfg.ReplicationFactor / 2
 
 		if len(healthyInstances) < numRequired {
-			return ReplicationSet{}, zoneFailures, ErrTooManyUnhealthyInstances
+			return ReplicationSet{}, ErrTooManyUnhealthyInstances
 		}
 
 		maxErrors = len(healthyInstances) - numRequired
@@ -580,18 +588,7 @@ func (r *Ring) getReplicationSetForOperation(op Operation, requireQuorum bool) (
 		Instances:           healthyInstances,
 		MaxErrors:           maxErrors,
 		MaxUnavailableZones: maxUnavailableZones,
-	}, zoneFailures, nil
-}
-
-// GetReplicationSetForOperation implements ReadRing.
-func (r *Ring) GetReplicationSetForOperation(op Operation) (ReplicationSet, error) {
-	replicationSet, _, err := r.getReplicationSetForOperation(op, true)
-	return replicationSet, err
-}
-
-// GetReplicationSetForOperationWithNoQuorum implements ReadRing.
-func (r *Ring) GetReplicationSetForOperationWithNoQuorum(op Operation) (ReplicationSet, map[string]struct{}, error) {
-	return r.getReplicationSetForOperation(op, false)
+	}, nil
 }
 
 // countTokens returns the number of tokens and tokens within the range for each instance.

pkg/ring/ring_test.go

@@ -959,11 +959,45 @@ func TestRing_GetInstanceDescsForOperation(t *testing.T) {
 	}, instanceDescs)
 }
 
-func validateGetReplicationSetForOperation(t *testing.T, requireQuorum bool) {
+func TestRing_GetAllInstanceDescs(t *testing.T) {
+	now := time.Now().Unix()
+	twoMinutesAgo := time.Now().Add(-2 * time.Minute).Unix()
+
+	ringDesc := &Desc{Ingesters: map[string]InstanceDesc{
+		"instance-1": {Addr: "127.0.0.1", Tokens: []uint32{1}, State: ACTIVE, Timestamp: now},
+		"instance-2": {Addr: "127.0.0.2", Tokens: []uint32{2}, State: LEAVING, Timestamp: now},          // not healthy state
+		"instance-3": {Addr: "127.0.0.3", Tokens: []uint32{3}, State: ACTIVE, Timestamp: twoMinutesAgo}, // heartbeat timed out
+	}}
+
+	ring := Ring{
+		cfg:                 Config{HeartbeatTimeout: time.Minute},
+		ringDesc:            ringDesc,
+		ringTokens:          ringDesc.GetTokens(),
+		ringTokensByZone:    ringDesc.getTokensByZone(),
+		ringInstanceByToken: ringDesc.getTokensInfo(),
+		ringZones:           getZones(ringDesc.getTokensByZone()),
+		strategy:            NewDefaultReplicationStrategy(),
+		KVClient:            &MockClient{},
+	}
+
+	testOp := NewOp([]InstanceState{ACTIVE}, nil)
+
+	healthyInstanceDescs, unhealthyInstanceDescs, err := ring.GetAllInstanceDescs(testOp)
+	require.NoError(t, err)
+	require.EqualValues(t, []InstanceDesc{
+		{Addr: "127.0.0.1", Tokens: []uint32{1}, State: ACTIVE, Timestamp: now},
+	}, healthyInstanceDescs)
+	require.EqualValues(t, []InstanceDesc{
+		{Addr: "127.0.0.2", Tokens: []uint32{2}, State: LEAVING, Timestamp: now},
+		{Addr: "127.0.0.3", Tokens: []uint32{3}, State: ACTIVE, Timestamp: twoMinutesAgo},
+	}, unhealthyInstanceDescs)
+}
+
+func TestRing_GetReplicationSetForOperation(t *testing.T) {
 	now := time.Now()
 	g := NewRandomTokenGenerator()
 
-	type testCase struct {
+	tests := map[string]struct {
 		ringInstances           map[string]InstanceDesc
 		ringHeartbeatTimeout    time.Duration
 		ringReplicationFactor   int
@@ -973,9 +1007,7 @@ func validateGetReplicationSetForOperation(t *testing.T, requireQuorum bool) {
 		expectedSetForWrite     []string
 		expectedErrForReporting error
 		expectedSetForReporting []string
-	}
-
-	tests := map[string]testCase{
+	}{
 		"should return error on empty ring": {
 			ringInstances:           nil,
 			ringHeartbeatTimeout:    time.Minute,
@@ -1040,10 +1072,7 @@ func validateGetReplicationSetForOperation(t *testing.T, requireQuorum bool) {
 			expectedSetForWrite:     []string{"127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4"},
 			expectedSetForReporting: []string{"127.0.0.1", "127.0.0.2", "127.0.0.3", "127.0.0.4"},
 		},
-	}
-
-	if requireQuorum {
-		tests["should fail on 2 unhealthy instances and RF=3"] = testCase{
+		"should fail on 2 unhealthy instances and RF=3": {
 			ringInstances: map[string]InstanceDesc{
 				"instance-1": {Addr: "127.0.0.1", State: ACTIVE, Timestamp: now.Unix(), Tokens: g.GenerateTokens(NewDesc(), "instance-1", "", 128, true)},
 				"instance-2": {Addr: "127.0.0.2", State: ACTIVE, Timestamp: now.Add(-10 * time.Second).Unix(), Tokens: g.GenerateTokens(NewDesc(), "instance-2", "", 128, true)},
@@ -1056,36 +1085,7 @@ func validateGetReplicationSetForOperation(t *testing.T, requireQuorum bool) {
 			expectedErrForRead:      ErrTooManyUnhealthyInstances,
 			expectedErrForWrite:     ErrTooManyUnhealthyInstances,
 			expectedErrForReporting: ErrTooManyUnhealthyInstances,
-		}
-	} else {
-		tests["should pass on 2 unhealthy instances and RF=3"] = testCase{
-			ringInstances: map[string]InstanceDesc{
-				"instance-1": {Addr: "127.0.0.1", State: ACTIVE, Timestamp: now.Unix(), Tokens: GenerateTokens(128, nil)},
-				"instance-2": {Addr: "127.0.0.2", State: ACTIVE, Timestamp: now.Add(-10 * time.Second).Unix(), Tokens: GenerateTokens(128, nil)},
-				"instance-3": {Addr: "127.0.0.3", State: ACTIVE, Timestamp: now.Add(-20 * time.Second).Unix(), Tokens: GenerateTokens(128, nil)},
-				"instance-4": {Addr: "127.0.0.4", State: ACTIVE, Timestamp: now.Add(-2 * time.Minute).Unix(), Tokens: GenerateTokens(128, nil)},
-				"instance-5": {Addr: "127.0.0.5", State: ACTIVE, Timestamp: now.Add(-2 * time.Minute).Unix(), Tokens: GenerateTokens(128, nil)},
-			},
-			ringHeartbeatTimeout:    time.Minute,
-			ringReplicationFactor:   3,
-			expectedSetForRead:      []string{"127.0.0.1", "127.0.0.2", "127.0.0.3"},
-			expectedSetForWrite:     []string{"127.0.0.1", "127.0.0.2", "127.0.0.3"},
-			expectedSetForReporting: []string{"127.0.0.1", "127.0.0.2", "127.0.0.3"},
-		}
-		tests["should fail on 3 unhealthy instances and RF=3"] = testCase{
-			ringInstances: map[string]InstanceDesc{
-				"instance-1": {Addr: "127.0.0.1", State: ACTIVE, Timestamp: now.Unix(), Tokens: GenerateTokens(128, nil)},
-				"instance-2": {Addr: "127.0.0.2", State: ACTIVE, Timestamp: now.Add(-10 * time.Second).Unix(), Tokens: GenerateTokens(128, nil)},
-				"instance-3": {Addr: "127.0.0.3", State: ACTIVE, Timestamp: now.Add(-2 * time.Minute).Unix(), Tokens: GenerateTokens(128, nil)},
-				"instance-4": {Addr: "127.0.0.4", State: ACTIVE, Timestamp: now.Add(-2 * time.Minute).Unix(), Tokens: GenerateTokens(128, nil)},
-				"instance-5": {Addr: "127.0.0.5", State: ACTIVE, Timestamp: now.Add(-2 * time.Minute).Unix(), Tokens: GenerateTokens(128, nil)},
-			},
-			ringHeartbeatTimeout:    time.Minute,
-			ringReplicationFactor:   3,
-			expectedErrForRead:      ErrTooManyUnhealthyInstances,
-			expectedErrForWrite:     ErrTooManyUnhealthyInstances,
-			expectedErrForReporting: ErrTooManyUnhealthyInstances,
-		}
+		},
 	}
 
 	for testName, testData := range tests {
@@ -1110,44 +1110,21 @@ func validateGetReplicationSetForOperation(t *testing.T, requireQuorum bool) {
 				KVClient:            &MockClient{},
 			}
 
-			var set ReplicationSet
-			var err error
-
-			if requireQuorum {
-				set, err = ring.GetReplicationSetForOperation(Read)
-			} else {
-				set, _, err = ring.GetReplicationSetForOperationWithNoQuorum(Read)
-			}
+			set, err := ring.GetReplicationSetForOperation(Read)
 			require.Equal(t, testData.expectedErrForRead, err)
 			assert.ElementsMatch(t, testData.expectedSetForRead, set.GetAddresses())
 
-			if requireQuorum {
-				set, err = ring.GetReplicationSetForOperation(Write)
-			} else {
-				set, _, err = ring.GetReplicationSetForOperationWithNoQuorum(Write)
-			}
+			set, err = ring.GetReplicationSetForOperation(Write)
 			require.Equal(t, testData.expectedErrForWrite, err)
 			assert.ElementsMatch(t, testData.expectedSetForWrite, set.GetAddresses())
 
-			if requireQuorum {
-				set, err = ring.GetReplicationSetForOperation(Reporting)
-			} else {
-				set, _, err = ring.GetReplicationSetForOperationWithNoQuorum(Reporting)
-			}
+			set, err = ring.GetReplicationSetForOperation(Reporting)
 			require.Equal(t, testData.expectedErrForReporting, err)
 			assert.ElementsMatch(t, testData.expectedSetForReporting, set.GetAddresses())
 		})
 	}
 }
 
-func TestRing_GetReplicationSetForOperation(t *testing.T) {
-	validateGetReplicationSetForOperation(t, true)
-}
-
-func TestRing_GetReplicationSetForOperationWithNoQuorum(t *testing.T) {
-	validateGetReplicationSetForOperation(t, false)
-}
-
 func TestRing_GetReplicationSetForOperation_WithZoneAwarenessEnabled(t *testing.T) {
 	g := NewRandomTokenGenerator()
 	tests := map[string]struct {

pkg/ring/util_test.go

@@ -36,14 +36,14 @@ func (r *RingMock) GetInstanceDescsForOperation(op Operation) (map[string]Instan
 	return args.Get(0).(map[string]InstanceDesc), args.Error(1)
 }
 
-func (r *RingMock) GetReplicationSetForOperation(op Operation) (ReplicationSet, error) {
+func (r *RingMock) GetAllInstanceDescs(op Operation) ([]InstanceDesc, []InstanceDesc, error) {
 	args := r.Called(op)
-	return args.Get(0).(ReplicationSet), args.Error(1)
+	return args.Get(0).([]InstanceDesc), make([]InstanceDesc, 0), args.Error(1)
 }
 
-func (r *RingMock) GetReplicationSetForOperationWithNoQuorum(op Operation) (ReplicationSet, map[string]struct{}, error) {
+func (r *RingMock) GetReplicationSetForOperation(op Operation) (ReplicationSet, error) {
 	args := r.Called(op)
-	return args.Get(0).(ReplicationSet), make(map[string]struct{}), args.Error(1)
+	return args.Get(0).(ReplicationSet), args.Error(1)
 }
 
 func (r *RingMock) ReplicationFactor() int {

pkg/ruler/rule_backup_manager.go

@@ -53,7 +53,7 @@ func (r *rulesBackupManager) getRuleGroups(userID string) rulespb.RuleGroupList
 	return result
 }
 
-// getRuleGroups updates the ruler_backup_rule_group metric by adding new groups that were backed up and removing
+// updateMetrics updates the ruler_backup_rule_group metric by adding new groups that were backed up and removing
 // those that are removed from the backup.
 func (r *rulesBackupManager) updateMetrics(newBackupGroups map[string]rulespb.RuleGroupList) {
 	for user, groups := range newBackupGroups {

pkg/ruler/ruler.go

@@ -198,7 +198,7 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.DurationVar(&cfg.FlushCheckPeriod, "ruler.flush-period", 1*time.Minute, "Period with which to attempt to flush rule groups.")
 	f.StringVar(&cfg.RulePath, "ruler.rule-path", "/rules", "file path to store temporary rule files for the prometheus rule managers")
 	f.BoolVar(&cfg.EnableAPI, "experimental.ruler.enable-api", false, "Enable the ruler api")
-	f.BoolVar(&cfg.APIEnableRulesBackup, "experimental.ruler.api-enable-rules-backup", false, "EXPERIMENTAL: Enable rulers to store a copy of rules owned by other rulers with default state (state before any evaluation) and send this copy in list API requests as backup in case the ruler who owns the rule fails to send its rules. This allows the rules API to handle ruler outage by returning rules with default state. Ring replication-factor needs to be set to 3 or more for this to be useful.")
+	f.BoolVar(&cfg.APIEnableRulesBackup, "experimental.ruler.api-enable-rules-backup", false, "EXPERIMENTAL: Enable rulers to store a copy of rules owned by other rulers with default state (state before any evaluation) and send this copy in list API requests as backup in case the ruler who owns the rule fails to send its rules. This allows the rules API to handle ruler outage by returning rules with default state. Ring replication-factor needs to be set to 2 or more for this to be useful.")
 	f.BoolVar(&cfg.APIDeduplicateRules, "experimental.ruler.api-deduplicate-rules", false, "EXPERIMENTAL: Remove duplicate rules in the prometheus rules and alerts API response. If there are duplicate rules the rule with the latest evaluation timestamp will be kept.")
 	f.DurationVar(&cfg.OutageTolerance, "ruler.for-outage-tolerance", time.Hour, `Max time to tolerate outage for restoring "for" state of alert.`)
 	f.DurationVar(&cfg.ForGracePeriod, "ruler.for-grace-period", 10*time.Minute, `Minimum duration between alert and restored "for" state. This is maintained only for alerts with configured "for" time greater than grace period.`)
@@ -1102,7 +1102,7 @@ func (r *Ruler) getShardedRules(ctx context.Context, userID string, rulesRequest
 		ring = r.ring.ShuffleShard(userID, shardSize)
 	}
 
-	rulers, failedZones, err := ring.GetReplicationSetForOperationWithNoQuorum(ListRuleRingOp)
+	rulers, failedZones, err := GetReplicationSetForListRule(ring, &r.cfg.Ring)
 	if err != nil {
 		return nil, err
 	}