include all partial errors in log

Signed-off-by: Justin Jung <[email protected]>

Author: justinjung04

pkg/ring/replication_set.go

@@ -71,39 +71,32 @@ func (r ReplicationSet) Do(ctx context.Context, delay time.Duration, zoneResults
 		}(i, &r.Instances[i])
 	}
 
-	trackerFailed := false
-	cnt := 0
-
-track:
-	for !tracker.succeeded() {
+	for !tracker.succeeded() && !tracker.finished() {
 		select {
 		case res := <-ch:
 			tracker.done(res.instance, res.res, res.err)
 			if res.err != nil {
-				if tracker.failed() {
-					if !partialDataEnabled || tracker.failedCompletely() {
-						return nil, res.err
-					}
-					trackerFailed = true
+				if tracker.failed() && (!partialDataEnabled || tracker.failedCompletely()) {
+					return nil, res.err
 				}
 
 				// force one of the delayed requests to start
 				if delay > 0 && r.MaxUnavailableZones == 0 {
 					forceStart <- struct{}{}
 				}
 			}
-			cnt++
-			if cnt == len(r.Instances) {
-				break track
-			}
 
 		case <-ctx.Done():
 			return nil, ctx.Err()
 		}
 	}
 
-	if partialDataEnabled && trackerFailed {
-		return tracker.getResults(), fmt.Errorf("failed to get data from %s: %w", tracker.failedInstances(), partialdata.ErrPartialData)
+	if partialDataEnabled && tracker.failed() {
+		finalErr := partialdata.ErrPartialData
+		for _, partialErr := range tracker.getErrors() {
+			finalErr = fmt.Errorf("%w: %w", finalErr, partialErr)
+		}
+		return tracker.getResults(), finalErr
 	}
 
 	return tracker.getResults(), nil

pkg/ring/replication_set_test.go

@@ -203,7 +203,7 @@ func TestReplicationSet_Do(t *testing.T) {
 			queryPartialData:    true,
 			want:                []interface{}{1},
 			expectedError:       partialdata.ErrPartialData,
-			errStrContains:      []string{"failed to get data from", "10.0.0.1", "10.0.0.2"},
+			errStrContains:      []string{"10.0.0.1", "10.0.0.2", "zone failed"},
 		},
 		{
 			name:                "with partial data enabled, should fail on instances failing in all zones",

pkg/ring/replication_set_tracker.go

@@ -1,35 +1,40 @@
 package ring
 
+import "fmt"
+
 type replicationSetResultTracker interface {
 	// Signals an instance has done the execution, either successful (no error)
 	// or failed (with error). If successful, result will be recorded and can
 	// be accessed via getResults.
 	done(instance *InstanceDesc, result interface{}, err error)
 
+	// Returns true if all instances are done executing
+	finished() bool
+
 	// Returns true if the minimum number of successful results have been received.
 	succeeded() bool
 
 	// Returns true if the maximum number of failed executions have been reached.
 	failed() bool
 
-	// Returns list of instance addresses that failed
-	failedInstances() []string
-
 	// Returns true if executions failed in all instances or all zones.
 	failedCompletely() bool
 
 	// Returns recorded results.
 	getResults() []interface{}
+
+	// Returns errors
+	getErrors() []error
 }
 
 type defaultResultTracker struct {
 	minSucceeded int
 	numSucceeded int
 	numErrors    int
 	maxErrors    int
-	failedInst   []string
 	results      []interface{}
 	numInstances int
+	errors       []error
 }
 
 func newDefaultResultTracker(instances []InstanceDesc, maxErrors int) *defaultResultTracker {
@@ -38,7 +43,7 @@ func newDefaultResultTracker(instances []InstanceDesc, maxErrors int) *defaultRe
 		numSucceeded: 0,
 		numErrors:    0,
 		maxErrors:    maxErrors,
-		failedInst:   make([]string, 0, len(instances)),
+		errors:       make([]error, 0, len(instances)),
 		results:      make([]interface{}, 0, len(instances)),
 		numInstances: len(instances),
 	}
@@ -49,11 +54,15 @@ func (t *defaultResultTracker) done(instance *InstanceDesc, result interface{},
 		t.numSucceeded++
 		t.results = append(t.results, result)
 	} else {
-		t.failedInst = append(t.failedInst, instance.GetAddr())
+		t.errors = append(t.errors, fmt.Errorf("(%s) %w", instance.GetAddr(), err))
 		t.numErrors++
 	}
 }
 
+func (t *defaultResultTracker) finished() bool {
+	return t.numSucceeded+t.numErrors == t.numInstances
+}
+
 func (t *defaultResultTracker) succeeded() bool {
 	return t.numSucceeded >= t.minSucceeded
 }
@@ -66,14 +75,14 @@ func (t *defaultResultTracker) failedCompletely() bool {
 	return t.numInstances == t.numErrors
 }
 
-func (t *defaultResultTracker) failedInstances() []string {
-	return t.failedInst
-}
-
 func (t *defaultResultTracker) getResults() []interface{} {
 	return t.results
 }
 
+func (t *defaultResultTracker) getErrors() []error {
+	return t.errors
+}
+
 // zoneAwareResultTracker tracks the results per zone.
 // All instances in a zone must succeed in order for the zone to succeed.
 type zoneAwareResultTracker struct {
@@ -84,8 +93,9 @@ type zoneAwareResultTracker struct {
 	resultsPerZone      map[string][]interface{}
 	numInstances        int
 	zoneResultsQuorum   bool
-	failedInst          []string
 	zoneCount           int
+	doneCount           int
+	errors              []error
 }
 
 func newZoneAwareResultTracker(instances []InstanceDesc, maxUnavailableZones int, zoneResultsQuorum bool) *zoneAwareResultTracker {
@@ -95,7 +105,7 @@ func newZoneAwareResultTracker(instances []InstanceDesc, maxUnavailableZones int
 		maxUnavailableZones: maxUnavailableZones,
 		numInstances:        len(instances),
 		zoneResultsQuorum:   zoneResultsQuorum,
-		failedInst:          make([]string, 0, len(instances)),
+		errors:              make([]error, 0, len(instances)),
 	}
 
 	for _, instance := range instances {
@@ -111,7 +121,7 @@ func newZoneAwareResultTracker(instances []InstanceDesc, maxUnavailableZones int
 func (t *zoneAwareResultTracker) done(instance *InstanceDesc, result interface{}, err error) {
 	if err != nil {
 		t.failuresByZone[instance.Zone]++
-		t.failedInst = append(t.failedInst, instance.Addr)
+		t.errors = append(t.errors, fmt.Errorf("(%s) %w", instance.GetAddr(), err))
 	} else {
 		if _, ok := t.resultsPerZone[instance.Zone]; !ok {
 			// If it is the first result in the zone, then total number of instances
@@ -122,6 +132,11 @@ func (t *zoneAwareResultTracker) done(instance *InstanceDesc, result interface{}
 	}
 
 	t.waitingByZone[instance.Zone]--
+	t.doneCount++
+}
+
+func (t *zoneAwareResultTracker) finished() bool {
+	return t.doneCount == t.numInstances
 }
 
 func (t *zoneAwareResultTracker) succeeded() bool {
@@ -148,10 +163,6 @@ func (t *zoneAwareResultTracker) failedCompletely() bool {
 	return failedZones == t.zoneCount
 }
 
-func (t *zoneAwareResultTracker) failedInstances() []string {
-	return t.failedInst
-}
-
 func (t *zoneAwareResultTracker) getResults() []interface{} {
 	results := make([]interface{}, 0, t.numInstances)
 	if t.zoneResultsQuorum {
@@ -169,3 +180,7 @@ func (t *zoneAwareResultTracker) getResults() []interface{} {
 	}
 	return results
 }
+
+func (t *zoneAwareResultTracker) getErrors() []error {
+	return t.errors
+}

pkg/ring/replication_set_tracker_test.go

@@ -2,6 +2,7 @@ package ring
 
 import (
 	"errors"
+	"fmt"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -154,7 +155,7 @@ func TestDefaultResultTracker(t *testing.T) {
 				assert.Equal(t, []interface{}{[]int{1, 1, 1}, []int{2, 2, 2}, []int{3, 3, 3}}, tracker.getResults())
 			},
 		},
-		"failedCompletely should return true only if all instances have failed, regardless of max errors": {
+		"failedCompletely() should return true only if all instances have failed, regardless of max errors": {
 			instances: []InstanceDesc{instance1, instance2, instance3},
 			maxErrors: 1,
 			run: func(t *testing.T, tracker *defaultResultTracker) {
@@ -174,15 +175,28 @@ func TestDefaultResultTracker(t *testing.T) {
 				assert.True(t, tracker.failedCompletely())
 			},
 		},
-		"failedInstances should work": {
+		"finished() should return true only if all instances are done": {
 			instances: []InstanceDesc{instance1, instance2},
-			maxErrors: 2,
+			maxErrors: 1,
 			run: func(t *testing.T, tracker *defaultResultTracker) {
 				tracker.done(&instance1, nil, errors.New("test"))
-				assert.ElementsMatch(t, []string{"127.0.0.1"}, tracker.failedInstances())
+				assert.False(t, tracker.finished())
 
 				tracker.done(&instance2, nil, errors.New("test"))
-				assert.ElementsMatch(t, []string{"127.0.0.1", "127.0.0.2"}, tracker.failedInstances())
+				assert.True(t, tracker.finished())
+			},
+		},
+		"getErrors() should return list of all errors": {
+			instances: []InstanceDesc{instance1, instance2},
+			maxErrors: 1,
+			run: func(t *testing.T, tracker *defaultResultTracker) {
+				tracker.done(&instance1, nil, errors.New("test1"))
+				err1 := fmt.Errorf("(%s) %w", instance1.GetAddr(), errors.New("test1"))
+				assert.ElementsMatch(t, []error{err1}, tracker.getErrors())
+
+				tracker.done(&instance2, nil, errors.New("test2"))
+				err2 := fmt.Errorf("(%s) %w", instance2.GetAddr(), errors.New("test2"))
+				assert.ElementsMatch(t, []error{err1, err2}, tracker.getErrors())
 			},
 		},
 	}
@@ -430,7 +444,7 @@ func TestZoneAwareResultTracker(t *testing.T) {
 				assert.False(t, tracker.failed())
 			},
 		},
-		"failedCompletely should return true only if all zones have failed, regardless of max unavailable zones": {
+		"failedCompletely() should return true only if all zones have failed, regardless of max unavailable zones": {
 			instances:           []InstanceDesc{instance1, instance2, instance3, instance4, instance5, instance6},
 			maxUnavailableZones: 1,
 			run: func(t *testing.T, tracker *zoneAwareResultTracker) {
@@ -453,15 +467,28 @@ func TestZoneAwareResultTracker(t *testing.T) {
 				assert.True(t, tracker.failedCompletely())
 			},
 		},
-		"failedInstances should work": {
+		"finished() should return true only if all instances are done": {
 			instances:           []InstanceDesc{instance1, instance2},
 			maxUnavailableZones: 1,
 			run: func(t *testing.T, tracker *zoneAwareResultTracker) {
 				tracker.done(&instance1, nil, errors.New("test"))
-				assert.ElementsMatch(t, []string{"127.0.0.1"}, tracker.failedInstances())
+				assert.False(t, tracker.finished())
 
 				tracker.done(&instance2, nil, errors.New("test"))
-				assert.ElementsMatch(t, []string{"127.0.0.1", "127.0.0.2"}, tracker.failedInstances())
+				assert.True(t, tracker.finished())
+			},
+		},
+		"getErrors() should return list of all errors": {
+			instances:           []InstanceDesc{instance1, instance2},
+			maxUnavailableZones: 1,
+			run: func(t *testing.T, tracker *zoneAwareResultTracker) {
+				tracker.done(&instance1, nil, errors.New("test1"))
+				err1 := fmt.Errorf("(%s) %w", instance1.GetAddr(), errors.New("test1"))
+				assert.ElementsMatch(t, []error{err1}, tracker.getErrors())
+
+				tracker.done(&instance2, nil, errors.New("test2"))
+				err2 := fmt.Errorf("(%s) %w", instance2.GetAddr(), errors.New("test2"))
+				assert.ElementsMatch(t, []error{err1, err2}, tracker.getErrors())
 			},
 		},
 	}