Reintroduce old behaviour, of reading from LEAVING ingesters.

Author: tomwilkie

pkg/ring/http.go

@@ -102,7 +102,7 @@ func (r *Ring) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 		ing := r.ringDesc.Ingesters[id]
 		timestamp := time.Unix(ing.Timestamp, 0)
 		state := ing.State.String()
-		if !r.IsHealthy(ing) {
+		if !r.IsHealthy(ing, Write) {
 			state = unhealthy
 		}
 

pkg/ring/replication_strategy.go

@@ -10,7 +10,7 @@ import (
 // tolerate.
 // - Filters out dead ingesters so the one doesn't even try to write to them.
 // - Checks there is enough ingesters for an operation to succeed.
-func (r *Ring) replicationStrategy(ingesters []*IngesterDesc) (
+func (r *Ring) replicationStrategy(ingesters []*IngesterDesc, op Operation) (
 	liveIngesters []*IngesterDesc, maxFailure int, err error,
 ) {
 	// We need a response from a quorum of ingesters, which is n/2 + 1.  In the
@@ -28,7 +28,7 @@ func (r *Ring) replicationStrategy(ingesters []*IngesterDesc) (
 	// will cause the whole write to fail.
 	liveIngesters = make([]*IngesterDesc, 0, len(ingesters))
 	for _, ingester := range ingesters {
-		if r.IsHealthy(ingester) {
+		if r.IsHealthy(ingester, op) {
 			liveIngesters = append(liveIngesters, ingester)
 		} else {
 			maxFailure--
@@ -47,8 +47,10 @@ func (r *Ring) replicationStrategy(ingesters []*IngesterDesc) (
 }
 
 // IsHealthy checks whether an ingester appears to be alive and heartbeating
-func (r *Ring) IsHealthy(ingester *IngesterDesc) bool {
-	if ingester.State != ACTIVE {
+func (r *Ring) IsHealthy(ingester *IngesterDesc, op Operation) bool {
+	if op == Write && ingester.State != ACTIVE {
+		return false
+	} else if op == Read && ingester.State == JOINING {
 		return false
 	}
 	return time.Now().Sub(time.Unix(ingester.Timestamp, 0)) <= r.cfg.HeartbeatTimeout

pkg/ring/replication_strategy_test.go

@@ -12,6 +12,7 @@ import (
 func TestReplicationStrategy(t *testing.T) {
 	for i, tc := range []struct {
 		RF, LiveIngesters, DeadIngesters int
+		op                               Operation // Will default to READ
 		ExpectedMaxFailure               int
 		ExpectedError                    string
 	}{
@@ -90,7 +91,7 @@ func TestReplicationStrategy(t *testing.T) {
 		require.NoError(t, err)
 
 		t.Run(fmt.Sprintf("[%d]", i), func(t *testing.T) {
-			liveIngesters, maxFailure, err := r.replicationStrategy(ingesters)
+			liveIngesters, maxFailure, err := r.replicationStrategy(ingesters, tc.op)
 			if tc.ExpectedError == "" {
 				assert.NoError(t, err)
 				assert.Equal(t, tc.LiveIngesters, len(liveIngesters))

pkg/ring/ring.go

@@ -227,7 +227,7 @@ func (r *Ring) getInternal(key uint32, op Operation) (ReplicationSet, error) {
 		ingesters = append(ingesters, ingester)
 	}
 
-	liveIngesters, maxFailure, err := r.replicationStrategy(ingesters)
+	liveIngesters, maxFailure, err := r.replicationStrategy(ingesters, op)
 	if err != nil {
 		return ReplicationSet{}, err
 	}
@@ -251,7 +251,7 @@ func (r *Ring) GetAll() (ReplicationSet, error) {
 	maxErrors := r.cfg.ReplicationFactor / 2
 
 	for _, ingester := range r.ringDesc.Ingesters {
-		if !r.IsHealthy(ingester) {
+		if !r.IsHealthy(ingester, Read) {
 			maxErrors--
 			continue
 		}
@@ -323,7 +323,7 @@ func (r *Ring) Collect(ch chan<- prometheus.Metric) {
 		LEAVING.String(): 0,
 	}
 	for _, ingester := range r.ringDesc.Ingesters {
-		if !r.IsHealthy(ingester) {
+		if !r.IsHealthy(ingester, Write) {
 			byState[unhealthy]++
 		} else {
 			byState[ingester.State.String()]++