refactor

Signed-off-by: Alan Protasio <[email protected]>

Author: alanprot

pkg/distributor/distributor_test.go

@@ -513,23 +513,23 @@ func TestPush_QuorumError(t *testing.T) {
 
 	d := dists[0]
 
-	// Using 489 just to make sure we are not hitting the &limits
+	// Using 429 just to make sure we are not hitting the &limits
 	// Simulating 2 4xx and 1 5xx -> Should return 4xx
-	ingesters[0].failResp.Store(httpgrpc.Errorf(489, "Throttling"))
+	ingesters[0].failResp.Store(httpgrpc.Errorf(429, "Throttling"))
 	ingesters[1].failResp.Store(httpgrpc.Errorf(500, "InternalServerError"))
-	ingesters[2].failResp.Store(httpgrpc.Errorf(489, "Throttling"))
+	ingesters[2].failResp.Store(httpgrpc.Errorf(429, "Throttling"))
 
 	for i := 0; i < 1000; i++ {
 		request := makeWriteRequest(0, 30, 20)
 		_, err := d.Push(ctx, request)
 		status, ok := status.FromError(err)
 		require.True(t, ok)
-		require.Equal(t, codes.Code(489), status.Code())
+		require.Equal(t, codes.Code(429), status.Code())
 	}
 
 	// Simulating 2 5xx and 1 4xx -> Should return 5xx
 	ingesters[0].failResp.Store(httpgrpc.Errorf(500, "InternalServerError"))
-	ingesters[1].failResp.Store(httpgrpc.Errorf(489, "Throttling"))
+	ingesters[1].failResp.Store(httpgrpc.Errorf(429, "Throttling"))
 	ingesters[2].failResp.Store(httpgrpc.Errorf(500, "InternalServerError"))
 
 	for i := 0; i < 10000; i++ {
@@ -542,15 +542,15 @@ func TestPush_QuorumError(t *testing.T) {
 
 	// Simulating 2 different errors and 1 success -> This case we may return any of the errors
 	ingesters[0].failResp.Store(httpgrpc.Errorf(500, "InternalServerError"))
-	ingesters[1].failResp.Store(httpgrpc.Errorf(489, "Throttling"))
+	ingesters[1].failResp.Store(httpgrpc.Errorf(429, "Throttling"))
 	ingesters[2].happy.Store(true)
 
 	for i := 0; i < 1000; i++ {
 		request := makeWriteRequest(0, 30, 20)
 		_, err := d.Push(ctx, request)
 		status, ok := status.FromError(err)
 		require.True(t, ok)
-		require.True(t, status.Code() == 489 || status.Code() == 500)
+		require.True(t, status.Code() == 429 || status.Code() == 500)
 	}
 
 	// Simulating 1 error -> Should return 2xx
@@ -581,7 +581,7 @@ func TestPush_QuorumError(t *testing.T) {
 	require.NoError(t, err)
 
 	// Give time to the ring get updated with the KV value
-	time.Sleep(5 * time.Second)
+	time.Sleep(time.Second)
 
 	for i := 0; i < 1000; i++ {
 		request := makeWriteRequest(0, 30, 20)

pkg/ring/batch.go

@@ -138,32 +138,33 @@ func (b *batchTracker) record(sampleTrackers []*itemTracker, err error) {
 	// The use of atomic increments here guarantees only a single sendSamples
 	// goroutine will write to either channel.
 	for i := range sampleTrackers {
-		verifyIfShouldReturn := func() {
-			if sampleTrackers[i].remaining.Dec() == 0 {
-				if b.rpcsFailed.Inc() == 1 {
-					b.err <- sampleTrackers[i].err.Load()
-				}
-			}
-		}
-
 		if err != nil {
+			// We count the error by error family (4xx and 5xx)
 			errCount := sampleTrackers[i].recordError(err)
-
-			if errCount <= int32(sampleTrackers[i].maxFailures) {
-				verifyIfShouldReturn()
-				continue
-			}
-
-			if b.rpcsFailed.Inc() == 1 {
-				b.err <- err
+			// We should return an error if we reach the maxFailure (quorum) on a give error family OR
+			// we dont have any remaining ingesters to try
+			// Ex: 2xx, 4xx, 5xx -> return 5xx
+			// Ex: 4xx, 4xx, 2xx -> return 4xx
+			if errCount > int32(sampleTrackers[i].maxFailures) || sampleTrackers[i].remaining.Dec() == 0 {
+				if b.rpcsFailed.Inc() == 1 {
+					b.err <- err
+				}
 			}
 		} else {
-			if sampleTrackers[i].succeeded.Inc() != int32(sampleTrackers[i].minSuccess) {
-				verifyIfShouldReturn()
+			// We should return success if we succeeded calling `minSuccess` ingesters
+			if sampleTrackers[i].succeeded.Inc() >= int32(sampleTrackers[i].minSuccess) {
+				if b.rpcsPending.Dec() == 0 {
+					b.done <- struct{}{}
+				}
 				continue
 			}
-			if b.rpcsPending.Dec() == 0 {
-				b.done <- struct{}{}
+
+			// If we suceeded to call this particular ingester but we dont have any remaining ingesters to try
+			// and we did not succeeded calling `minSuccess` ingesters we need to return the last error
+			if sampleTrackers[i].remaining.Dec() == 0 {
+				if b.rpcsFailed.Inc() == 1 {
+					b.err <- sampleTrackers[i].err.Load()
+				}
 			}
 		}
 	}