clientv3: fix balancer/retry

.words

@@ -1,14 +1,22 @@
+ConnectionError
 ErrCodeEnhanceYourCalm
+ErrConnClosing
+ErrEmptyKey
+ErrNoSpace
+ErrTimeout
+FailFast
 GoAway
 RPC
 RPCs
+StreamError
 backoff
 blackholed
 cancelable
 cancelation
 cluster_proxy
 defragment
 defragmenting
+downErr
 etcd
 gRPC
 goroutine
@@ -25,9 +33,10 @@ mutex
 prefetching
 protobuf
 serializable
+statusError
 teardown
+toRPCErr
 too_many_pings
 uncontended
 unprefixed
 unlisting
-

clientv3/balancer.go

@@ -45,7 +45,10 @@ type balancer interface {
 	// pinned returns the current pinned endpoint.
 	pinned() string
 	// endpointError handles error from server-side.
-	endpointError(addr string, err error)
+	endpointError(host string, err error)
+	// returns errorInfo of host, if any.
+	// ok is false, if host is healthy.
+	isFailed(host string) (ev errorInfo, ok bool)
 
 	// up is Up but includes whether the balancer will use the connection.
 	up(addr grpc.Address) (func(error), bool)
@@ -152,7 +155,13 @@ func (b *simpleBalancer) pinned() string {
 	return b.pinAddr
 }
 
-func (b *simpleBalancer) endpointError(addr string, err error) { return }
+func (b *simpleBalancer) endpointError(host string, err error) {
+	panic("'endpointError' not implemented")
+}
+
+func (b *simpleBalancer) isFailed(host string) (errorInfo, bool) {
+	panic("'error' not implemented")
+}
 
 func getHost2ep(eps []string) map[string]string {
 	hm := make(map[string]string, len(eps))
@@ -316,7 +325,7 @@ func (b *simpleBalancer) up(addr grpc.Address) (func(error), bool) {
 	}
 	if b.pinAddr != "" {
 		if logger.V(4) {
-			logger.Infof("clientv3/balancer: %s is up but not pinned (already pinned %s)", addr.Addr, b.pinAddr)
+			logger.Infof("clientv3/balancer: %q is up but not pinned (already pinned %q)", addr.Addr, b.pinAddr)
 		}
 		return func(err error) {}, false
 	}
@@ -325,7 +334,7 @@ func (b *simpleBalancer) up(addr grpc.Address) (func(error), bool) {
 	b.downc = make(chan struct{})
 	b.pinAddr = addr.Addr
 	if logger.V(4) {
-		logger.Infof("clientv3/balancer: pin %s", addr.Addr)
+		logger.Infof("clientv3/balancer: pin %q", addr.Addr)
 	}
 	// notify client that a connection is up
 	b.readyOnce.Do(func() { close(b.readyc) })
@@ -336,7 +345,7 @@ func (b *simpleBalancer) up(addr grpc.Address) (func(error), bool) {
 		b.pinAddr = ""
 		b.mu.Unlock()
 		if logger.V(4) {
-			logger.Infof("clientv3/balancer: unpin %s (%v)", addr.Addr, err)
+			logger.Infof("clientv3/balancer: unpin %q (%q)", addr.Addr, err.Error())
 		}
 	}, true
 }

clientv3/health_balancer.go

@@ -30,6 +30,11 @@ const unknownService = "unknown service grpc.health.v1.Health"
 
 type healthCheckFunc func(ep string) (bool, error)
 
+type errorInfo struct {
+	failed time.Time
+	err    error
+}
+
 // healthBalancer wraps a balancer so that it uses health checking
 // to choose its endpoints.
 type healthBalancer struct {
@@ -48,8 +53,8 @@ type healthBalancer struct {
 	// eps stores all client endpoints
 	eps []string
 
-	// unhealthy tracks the last unhealthy time of endpoints.
-	unhealthy map[string]time.Time
+	// unhealthyHosts tracks the last unhealthy time of endpoints.
+	unhealthyHosts map[string]errorInfo
 
 	stopc    chan struct{}
 	stopOnce sync.Once
@@ -61,13 +66,13 @@ type healthBalancer struct {
 
 func newHealthBalancer(b balancer, timeout time.Duration, hc healthCheckFunc) *healthBalancer {
 	hb := &healthBalancer{
-		balancer:    b,
-		healthCheck: hc,
-		eps:         b.endpoints(),
-		addrs:       eps2addrs(b.endpoints()),
-		host2ep:     getHost2ep(b.endpoints()),
-		unhealthy:   make(map[string]time.Time),
-		stopc:       make(chan struct{}),
+		balancer:       b,
+		healthCheck:    hc,
+		eps:            b.endpoints(),
+		addrs:          eps2addrs(b.endpoints()),
+		host2ep:        getHost2ep(b.endpoints()),
+		unhealthyHosts: make(map[string]errorInfo),
+		stopc:          make(chan struct{}),
 	}
 	if timeout < minHealthRetryDuration {
 		timeout = minHealthRetryDuration
@@ -94,11 +99,11 @@ func (hb *healthBalancer) Up(addr grpc.Address) func(error) {
 		// finding healthy endpoint on retry could take several timeouts and redials.
 		// To avoid wasting retries, gray-list unhealthy endpoints.
 		hb.mu.Lock()
-		hb.unhealthy[addr.Addr] = time.Now()
+		hb.unhealthyHosts[addr.Addr] = errorInfo{failed: time.Now(), err: err}
 		hb.mu.Unlock()
 		f(err)
 		if logger.V(4) {
-			logger.Infof("clientv3/health-balancer: %s becomes unhealthy (%v)", addr.Addr, err)
+			logger.Infof("clientv3/health-balancer: %q becomes unhealthy (%q)", addr.Addr, err.Error())
 		}
 	}
 }
@@ -120,7 +125,7 @@ func (hb *healthBalancer) updateAddrs(eps ...string) {
 	addrs, host2ep := eps2addrs(eps), getHost2ep(eps)
 	hb.mu.Lock()
 	hb.addrs, hb.eps, hb.host2ep = addrs, eps, host2ep
-	hb.unhealthy = make(map[string]time.Time)
+	hb.unhealthyHosts = make(map[string]errorInfo)
 	hb.mu.Unlock()
 	hb.balancer.updateAddrs(eps...)
 }
@@ -142,11 +147,18 @@ func (hb *healthBalancer) updateUnhealthy(timeout time.Duration) {
 		select {
 		case <-time.After(timeout):
 			hb.mu.Lock()
-			for k, v := range hb.unhealthy {
-				if time.Since(v) > timeout {
-					delete(hb.unhealthy, k)
+			for k, v := range hb.unhealthyHosts {
+				if _, ok := hb.host2ep[k]; !ok {
+					delete(hb.unhealthyHosts, k)
+					if logger.V(4) {
+						logger.Infof("clientv3/health-balancer: removes stale endpoint %q from unhealthy", k)
+					}
+					continue
+				}
+				if time.Since(v.failed) > timeout {
+					delete(hb.unhealthyHosts, k)
 					if logger.V(4) {
-						logger.Infof("clientv3/health-balancer: removes %s from unhealthy after %v", k, timeout)
+						logger.Infof("clientv3/health-balancer: removes %q from unhealthy after %v", k, timeout)
 					}
 				}
 			}
@@ -166,31 +178,42 @@ func (hb *healthBalancer) liveAddrs() []grpc.Address {
 	hb.mu.RLock()
 	defer hb.mu.RUnlock()
 	hbAddrs := hb.addrs
-	if len(hb.addrs) == 1 || len(hb.unhealthy) == 0 || len(hb.unhealthy) == len(hb.addrs) {
+	if len(hb.addrs) == 1 || len(hb.unhealthyHosts) == 0 || len(hb.unhealthyHosts) == len(hb.addrs) {
 		return hbAddrs
 	}
-	addrs := make([]grpc.Address, 0, len(hb.addrs)-len(hb.unhealthy))
+	addrs := make([]grpc.Address, 0, len(hb.addrs)-len(hb.unhealthyHosts))
 	for _, addr := range hb.addrs {
-		if _, unhealthy := hb.unhealthy[addr.Addr]; !unhealthy {
+		if _, unhealthy := hb.unhealthyHosts[addr.Addr]; !unhealthy {
 			addrs = append(addrs, addr)
 		}
 	}
 	return addrs
 }
 
-func (hb *healthBalancer) endpointError(addr string, err error) {
+func (hb *healthBalancer) endpointError(host string, err error) {
 	hb.mu.Lock()
-	hb.unhealthy[addr] = time.Now()
+	hb.unhealthyHosts[host] = errorInfo{failed: time.Now(), err: err}
 	hb.mu.Unlock()
 	if logger.V(4) {
-		logger.Infof("clientv3/health-balancer: marking %s as unhealthy (%v)", addr, err)
+		logger.Infof("clientv3/health-balancer: marking %q as unhealthy (%q)", host, err.Error())
 	}
 }
 
+func (hb *healthBalancer) isFailed(host string) (ev errorInfo, ok bool) {
+	hb.mu.RLock()
+	ev, ok = hb.unhealthyHosts[host]
+	hb.mu.RUnlock()
+	return ev, ok
+}
+
 func (hb *healthBalancer) mayPin(addr grpc.Address) bool {
 	hb.mu.RLock()
-	skip := len(hb.addrs) == 1 || len(hb.unhealthy) == 0 || len(hb.addrs) == len(hb.unhealthy)
-	failedTime, bad := hb.unhealthy[addr.Addr]
+	if _, ok := hb.host2ep[addr.Addr]; !ok {
+		hb.mu.RUnlock()
+		return false
+	}
+	skip := len(hb.addrs) == 1 || len(hb.unhealthyHosts) == 0 || len(hb.addrs) == len(hb.unhealthyHosts)
+	ef, bad := hb.unhealthyHosts[addr.Addr]
 	dur := hb.healthCheckTimeout
 	hb.mu.RUnlock()
 	if skip || !bad {
@@ -201,26 +224,27 @@ func (hb *healthBalancer) mayPin(addr grpc.Address) bool {
 	//   2. balancer 'Up' unpins with grpc: failed with network I/O error
 	//   3. grpc-healthcheck still SERVING, thus retry to pin
 	// instead, return before grpc-healthcheck if failed within healthcheck timeout
-	if elapsed := time.Since(failedTime); elapsed < dur {
+	if elapsed := time.Since(ef.failed); elapsed < dur {
 		if logger.V(4) {
-			logger.Infof("clientv3/health-balancer: %s is up but not pinned (failed %v ago, require minimum %v after failure)", addr.Addr, elapsed, dur)
+			logger.Infof("clientv3/health-balancer: %q is up but not pinned (failed %v ago, require minimum %v after failure)", addr.Addr, elapsed, dur)
 		}
 		return false
 	}
-	if ok, _ := hb.healthCheck(addr.Addr); ok {
+	ok, err := hb.healthCheck(addr.Addr)
+	if ok {
 		hb.mu.Lock()
-		delete(hb.unhealthy, addr.Addr)
+		delete(hb.unhealthyHosts, addr.Addr)
 		hb.mu.Unlock()
 		if logger.V(4) {
-			logger.Infof("clientv3/health-balancer: %s is healthy (health check success)", addr.Addr)
+			logger.Infof("clientv3/health-balancer: %q is healthy (health check success)", addr.Addr)
 		}
 		return true
 	}
 	hb.mu.Lock()
-	hb.unhealthy[addr.Addr] = time.Now()
+	hb.unhealthyHosts[addr.Addr] = errorInfo{failed: time.Now(), err: err}
 	hb.mu.Unlock()
 	if logger.V(4) {
-		logger.Infof("clientv3/health-balancer: %s becomes unhealthy (health check failed)", addr.Addr)
+		logger.Infof("clientv3/health-balancer: %q becomes unhealthy (health check failed)", addr.Addr)
 	}
 	return false
 }