retry after pause when UpdateState returns an error

Since grpc-go 1.38 the UpdateState() call of the ClientConn can return an error.

When UpdateState() returns an error the grpcconsulresolver now polls consul and
retries calling UpdateState() again after a backoff pause.
This is analogue to how the dns resolver that is part of the grpc-go package
handles UpdateState() errors.

Instead of using a timer to pause before polling consul again, the WaitTime for
the consul query is set to the amount of the backoff timeout.
This has the advantage that if the address is consul is updated, the wait is
interrupted and UpdateState() is called again with the new addresses.

The backoff implementation was restored from the commit
87eb606.

Author: Fabian Holler

consul/backoff.go

@@ -0,0 +1,38 @@
+package consul
+
+import (
+	"math/rand"
+	"time"
+)
+
+type backoff struct {
+	intervals []time.Duration
+	jitterPct int
+	jitterSrc *rand.Rand
+}
+
+var defBackoff = backoff{
+	intervals: []time.Duration{
+		500 * time.Millisecond,
+		2 * time.Second,
+		5 * time.Second,
+		10 * time.Second,
+		30 * time.Second,
+		1 * time.Minute,
+	},
+
+	jitterPct: 20,
+	jitterSrc: rand.New(rand.NewSource(time.Now().UnixNano())),
+}
+
+func (b *backoff) Backoff(retry int) time.Duration {
+	idx := retry
+
+	if idx < 0 || idx > len(b.intervals)-1 {
+		idx = len(b.intervals) - 1
+	}
+
+	d := b.intervals[idx]
+
+	return d + time.Duration(((int64(d) / 100) * int64((b.jitterSrc.Intn(b.jitterPct)))))
+}

consul/backoff_test.go

@@ -0,0 +1,60 @@
+package consul
+
+import (
+	"testing"
+	"time"
+)
+
+func maxBackoffJitter(idx int) time.Duration {
+	return time.Duration(int64(defBackoff.intervals[idx]) / 100 * int64(defBackoff.jitterPct))
+}
+
+func TestBackoff(t *testing.T) {
+	for i := range defBackoff.intervals {
+		d := defBackoff.Backoff(i)
+		if d < defBackoff.intervals[i] {
+			t.Errorf("duration for retry %d is %d expected >= %d", i, d, defBackoff.intervals[i])
+		}
+
+		maxJitter := maxBackoffJitter(i)
+		if d >= d+maxJitter {
+			t.Errorf("duration for retry %d is %d expected <= %d", i, d, d+maxJitter)
+		}
+	}
+
+}
+
+func TestBackoff_IntervalIdxBounds(t *testing.T) {
+	tests := []struct {
+		name  string
+		retry int
+		idx   int
+	}{
+		{
+			name:  "retry: -1",
+			retry: -1,
+			idx:   0,
+		},
+		{
+			name:  "retry > interval len",
+			retry: len(defBackoff.intervals),
+			idx:   len(defBackoff.intervals) - 1,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			retry := tt.retry
+
+			d := defBackoff.Backoff(retry)
+			if d < defBackoff.intervals[tt.idx] {
+				t.Errorf("duration for retry %d is %d expected >= %d", retry, d, defBackoff.intervals[tt.idx])
+			}
+
+			maxJitter := maxBackoffJitter(tt.idx)
+			if d >= d+maxJitter {
+				t.Errorf("duration for retry %d is %d expected <= %d", retry, d, d+maxJitter)
+			}
+
+		})
+	}
+}

consul/resolver.go

@@ -30,6 +30,7 @@ type consulResolver struct {
 	healthFilter healthFilter
 	ctx          context.Context
 	cancel       context.CancelFunc
+	backoff      *backoff
 	resolveNow   chan struct{}
 	wgStop       sync.WaitGroup
 }
@@ -73,6 +74,7 @@ func newConsulResolver(
 		service:      consulService,
 		tags:         tags,
 		healthFilter: healthFilter,
+		backoff:      &defBackoff,
 		ctx:          ctx,
 		cancel:       cancel,
 		resolveNow:   make(chan struct{}, 1),
@@ -171,12 +173,24 @@ func (c *consulResolver) watcher() {
 	defer c.wgStop.Done()
 
 	for {
+		var retryCount int
+
 		for {
 			var addrs []resolver.Address
 			var err error
 
 			lastWaitIndex := opts.WaitIndex
 
+			if retryCount > 0 {
+				// if UpdateState returned an error, we retry resolving
+				// + setting UpdateState after a backoff pause.
+				// The pause is applied by doing a consul query
+				// and setting a timeout. This has the
+				// advantage that the pause is also interrupted
+				// if the entry in consul was updated.
+				opts.WaitTime = c.backoff.Backoff(retryCount)
+			}
+
 			queryStartTime := time.Now()
 			addrs, opts.WaitIndex, err = c.query(opts)
 			if err != nil {
@@ -206,7 +220,9 @@ func (c *consulResolver) watcher() {
 			// addresses (addrs is nil), we have to report an empty
 			// set of resolved addresses. It informs the grpc-balancer that resolution is not
 			// in progress anymore and grpc calls can failFast.
-			if addressesEqual(addrs, lastReportedAddrs) {
+			// If the previous call to UpdateState failed, retryCount is != 0, then this
+			// check is ignored and UpdateState is called again.
+			if addressesEqual(addrs, lastReportedAddrs) && retryCount == 0 {
 				// If the consul server responds with
 				// the same data then in the last
 				// query in less then 50ms, we sleep a
@@ -223,8 +239,15 @@ func (c *consulResolver) watcher() {
 				continue
 			}
 
-			c.cc.UpdateState(resolver.State{Addresses: addrs})
 			lastReportedAddrs = addrs
+			err = c.cc.UpdateState(resolver.State{Addresses: addrs})
+			if err != nil {
+				grpclog.Infof("grpcconsulresolver: updateState failed: %s, retrying after pause", err)
+				retryCount++
+				continue
+			}
+
+			retryCount = 0
 		}
 
 		select {

consul/resolver_test.go

@@ -565,3 +565,60 @@ func TestQueryResultsAreSorted(t *testing.T) {
 
 	r.Close()
 }
+
+func TestConsulIsPolledOnUpdateStateErr(t *testing.T) {
+	cc := mocks.NewClientConn()
+	health := mocks.NewConsulHealthClient()
+	cleanup := replaceCreateHealthClientFn(
+		func(cfg *consul.Config) (consulHealthEndpoint, error) {
+			return health, nil
+		},
+	)
+	t.Cleanup(cleanup)
+
+	health.SetRespEntries([]*consul.ServiceEntry{
+		{
+			Service: &consul.AgentService{
+				Address: "227.0.0.1",
+				Port:    1,
+			},
+		},
+	})
+
+	defBackoff.intervals = []time.Duration{
+		1 * time.Millisecond,
+		2 * time.Millisecond,
+		3 * time.Millisecond,
+		4 * time.Millisecond,
+		5 * time.Millisecond,
+	}
+
+	r, err := NewBuilder().Build(resolver.Target{Endpoint: "test"}, cc, resolver.BuildOptions{})
+	if err != nil {
+		t.Fatal("Build() failed:", err.Error())
+	}
+
+	cc.SetUpdateStateReturnVal(errors.New("mocked updatestate err"))
+
+	r.ResolveNow(resolver.ResolveNowOptions{})
+
+	// consul should be polled and updateState queried again after exponential sleep
+	for cc.UpdateStateCallCnt() < 3 {
+		time.Sleep(500 * time.Microsecond)
+	}
+
+	// change UpdateState to return a nil error, afterwards UpdateState should only be called 1 time again
+	cc.SetUpdateStateReturnVal(nil)
+	updateStateCalls := cc.UpdateStateCallCnt()
+
+	maxBackoff := defBackoff.intervals[len(defBackoff.intervals)-1]
+	maxSleep := maxBackoff + maxBackoff*(time.Duration((defBackoff.jitterPct / 100))) + time.Second
+	time.Sleep(maxSleep)
+
+	updateStateCallsAfterSuccess := cc.UpdateStateCallCnt()
+	if updateStateCalls+1 > updateStateCallsAfterSuccess {
+		t.Errorf("resolver called UpdateState %d times after resolving succeeded, expected <= %d calls",
+			updateStateCallsAfterSuccess, updateStateCalls+1,
+		)
+	}
+}

internal/mocks/clientconn.go

@@ -9,10 +9,11 @@ import (
 )
 
 type ClientConn struct {
-	mutex             sync.Mutex
-	addrs             []resolver.Address
-	newAddressCallCnt int
-	lastReportedError error
+	mutex                sync.Mutex
+	addrs                []resolver.Address
+	newAddressCallCnt    int
+	lastReportedError    error
+	updateStateReturnVal error
 }
 
 func NewClientConn() *ClientConn {
@@ -37,12 +38,14 @@ func (t *ClientConn) LastReportedError() error {
 	return t.lastReportedError
 }
 
-func (t *ClientConn) UpdateState(state resolver.State) {
+func (t *ClientConn) UpdateState(state resolver.State) error {
 	t.mutex.Lock()
 	defer t.mutex.Unlock()
 
 	t.addrs = state.Addresses
 	t.newAddressCallCnt++
+
+	return t.updateStateReturnVal
 }
 
 func (t *ClientConn) NewAddress(addrs []resolver.Address) {
@@ -67,5 +70,11 @@ func (t *ClientConn) Addrs() (addrs []resolver.Address) {
 	return t.addrs
 }
 
-func (*ClientConn) NewServiceConfig(string) {
+func (*ClientConn) NewServiceConfig(string) {}
+
+func (t *ClientConn) SetUpdateStateReturnVal(v error) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+
+	t.updateStateReturnVal = v
 }

internal/mocks/consulhealth.go

@@ -2,6 +2,7 @@ package mocks
 
 import (
 	"sync"
+	"time"
 
 	consul "github.com/hashicorp/consul/api"
 )
@@ -51,5 +52,7 @@ func (c *ConsulHealthClient) ServiceMultipleTags(service string, tags []string,
 		return nil, nil, q.Context().Err()
 	}
 
+	time.Sleep(q.WaitTime)
+
 	return c.entries, &c.queryMeta, c.err
 }