Comments

Author: alanprot

pkg/distributor/distributor_test.go

@@ -494,6 +494,7 @@ func TestDistributor_PushIngestionRateLimiter(t *testing.T) {
 }
 
 func TestPush_QuorumError(t *testing.T) {
+
 	var limits validation.Limits
 	flagext.DefaultValues(&limits)
 
@@ -513,13 +514,16 @@ func TestPush_QuorumError(t *testing.T) {
 
 	d := dists[0]
 
+	// we should run several write request to make sure we dont have any race condition on the batchTracker#record code
+	numberOfWrites := 10000
+
 	// 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(429, "Throttling"))
 	ingesters[1].failResp.Store(httpgrpc.Errorf(500, "InternalServerError"))
 	ingesters[2].failResp.Store(httpgrpc.Errorf(429, "Throttling"))
 
-	for i := 0; i < 1000; i++ {
+	for i := 0; i < numberOfWrites; i++ {
 		request := makeWriteRequest(0, 30, 20)
 		_, err := d.Push(ctx, request)
 		status, ok := status.FromError(err)
@@ -532,7 +536,7 @@ func TestPush_QuorumError(t *testing.T) {
 	ingesters[1].failResp.Store(httpgrpc.Errorf(429, "Throttling"))
 	ingesters[2].failResp.Store(httpgrpc.Errorf(500, "InternalServerError"))
 
-	for i := 0; i < 10000; i++ {
+	for i := 0; i < numberOfWrites; i++ {
 		request := makeWriteRequest(0, 300, 200)
 		_, err := d.Push(ctx, request)
 		status, ok := status.FromError(err)
@@ -545,7 +549,7 @@ func TestPush_QuorumError(t *testing.T) {
 	ingesters[1].failResp.Store(httpgrpc.Errorf(429, "Throttling"))
 	ingesters[2].happy.Store(true)
 
-	for i := 0; i < 1; i++ {
+	for i := 0; i < numberOfWrites; i++ {
 		request := makeWriteRequest(0, 30, 20)
 		_, err := d.Push(ctx, request)
 		status, ok := status.FromError(err)
@@ -569,18 +573,6 @@ func TestPush_QuorumError(t *testing.T) {
 	ingesters[1].happy.Store(true)
 	ingesters[2].happy.Store(true)
 
-	// Wait group to check when the ring got updated
-	wg := &sync.WaitGroup{}
-	wg.Add(1)
-
-	go func() {
-		r.KVClient.WatchKey(context.Background(), ingester.RingKey, func(i interface{}) bool {
-			wg.Done()
-			// False will terminate the watch
-			return false
-		})
-	}()
-
 	err := r.KVClient.CAS(context.Background(), ingester.RingKey, func(in interface{}) (interface{}, bool, error) {
 		r := in.(*ring.Desc)
 		ingester2 := r.Ingesters["2"]
@@ -593,9 +585,15 @@ func TestPush_QuorumError(t *testing.T) {
 	require.NoError(t, err)
 
 	// Give time to the ring get updated with the KV value
-	wg.Wait()
+	for {
+		replicationSet, _ := r.GetAllHealthy(ring.Read)
+		if len(replicationSet.Instances) == 2 {
+			break
+		}
+		time.Sleep(100 * time.Millisecond)
+	}
 
-	for i := 0; i < 1000; i++ {
+	for i := 0; i < numberOfWrites; i++ {
 		request := makeWriteRequest(0, 30, 20)
 		_, err := d.Push(ctx, request)
 		require.Error(t, err)

pkg/ring/batch.go

@@ -128,20 +128,19 @@ func DoBatch(ctx context.Context, op Operation, r ReadRing, keys []uint32, callb
 }
 
 func (b *batchTracker) record(sampleTrackers []*itemTracker, err error) {
-	// If we succeed, decrement each sample's pending count by one.  If we reach
-	// the required number of successful puts on this sample, then decrement the
-	// number of pending samples by one.  If we successfully push all samples to
-	// min success instances, wake up the waiting rpc so it can return early.
-	// Similarly, track the number of errors, and if it exceeds maxFailures
-	// shortcut the waiting rpc.
+	// If we reach the required number of successful puts on this sample, then decrement the
+	// number of pending samples by one.
 	//
-	// The use of atomic increments here guarantees only a single sendSamples
-	// goroutine will write to either channel.
+	// The use of atomic increments here is needed as:
+	// * rpcsPending and rpcsPending guarantees only a single sendSamples goroutine will write to either channel
+	// * succeeded, failed4xx, failed5xx and remaining guarantees that the "return decision" is made atomically
+	// avoiding race condition
 	for i := range sampleTrackers {
 		if err != nil {
-			// We count the error by error family (4xx and 5xx)
+			// Track the number of errors by error family, and if it exceeds maxFailures
+			// shortcut the waiting rpc.
 			errCount := sampleTrackers[i].recordError(err)
-			// We should return an error if we reach the maxFailure (quorum) on a give error family OR
+			// We should return an error if we reach the maxFailure (quorum) on a given error family OR
 			// we dont have any remaining ingesters to try
 			// Ex: 2xx, 4xx, 5xx -> return 5xx
 			// Ex: 4xx, 4xx, _ -> return 4xx
@@ -152,7 +151,8 @@ func (b *batchTracker) record(sampleTrackers []*itemTracker, err error) {
 				}
 			}
 		} else {
-			// We should return success if we succeeded calling `minSuccess` ingesters.
+			// If we successfully push all samples to min success instances,
+			// wake up the waiting rpc so it can return early.
 			if sampleTrackers[i].succeeded.Inc() >= int32(sampleTrackers[i].minSuccess) {
 				if b.rpcsPending.Dec() == 0 {
 					b.done <- struct{}{}