clientv3: balancer.go and retry.go

when a endpoint is unavailable, switch it in retry

fix: etcd-io#8326

clientv3/balancer.go

@@ -16,6 +16,7 @@ package clientv3
 
 import (
 	"net/url"
+	"time"
 	"strings"
 	"sync"
 
@@ -68,6 +69,13 @@ type simpleBalancer struct {
 	// intialization and shutdown.
 	pinAddr string
 
+
+	// pinAddrcond is a bell rings when pinAddr change
+	pinAddrCond sync.Cond
+
+	// last switch time, avoid switch too frequently
+	lastSwitchTime time.Time
+
 	closed bool
 }
 
@@ -251,6 +259,7 @@ func (b *simpleBalancer) Up(addr grpc.Address) func(error) {
 	close(b.upc)
 	b.downc = make(chan struct{})
 	b.pinAddr = addr.Addr
+	b.pinAddrCond.Broadcast()
 	// notify client that a connection is up
 	b.readyOnce.Do(func() { close(b.readyc) })
 	return func(err error) {
@@ -354,3 +363,85 @@ func getHost(ep string) string {
 	}
 	return url.Host
 }
+
+func (b *simpleBalancer) removeAddrs(eps []string) {
+
+	np := getHost2ep(eps)
+	addrs := make([]grpc.Address, 0, len(eps))
+
+	b.mu.Lock()
+
+	change := false
+
+	for k, v := range b.host2ep {
+		if np[k] == v {
+			delete(b.host2ep, k)
+			change = true
+		} else {
+			addrs = append(addrs, grpc.Address{Addr: k)})
+		}
+	}
+
+	if !change {
+		// same endpoints, so no need to update address
+		b.mu.Unlock()
+		return
+	}
+
+	b.addrs = addrs
+
+
+	// updating notifyCh can trigger new connections,
+	// only update addrs if all connections are down
+	// or addrs does not include pinAddr.
+	update := !hasAddr(addrs, b.pinAddr)
+	b.mu.Unlock()
+
+	if update {
+		select {
+		case b.updateAddrsC <- struct{}{}:
+		case <-b.stopc:
+		}
+	}
+}
+
+func (b *simpleBalancer) switchEndpoint() {
+	tNow := time.Now()
+	b.mu.RLock()
+	lastSwitchTime := b.lastSwitchTime
+	b.mu.RUnlock()
+
+	if lastSwitchTime.Add(30 * time.Second).After(tNow) {
+		// avoid too frequently switch
+		return
+	}
+
+	addrs := make([]grpc.Address, 0, len(eps))
+
+	b.mu.Lock()
+	prevPinAddr := b.pinAddr
+	prevPinAddrEp := b.host2ep[prevPinAddr]
+	delete(b.host2ep, prevPinAddr)
+	for _, addr := range b.addrs {
+		if prevPinAddr != addr {
+			addrs = append(addrs, addr)
+		}
+
+	}
+	b.mu.Unlock()
+
+	select {
+	case b.updateAddrsC <- struct{}{}:
+	case <-b.stopc:
+	}
+
+
+	b.mu.Lock()
+	for (prevPinAddr == b.pinAddr) {
+		b.pinAddrCond.Wait()
+	}
+
+	b.host2ep[prevPinAddr] = prevPinAddrEp
+	b.addrs = append(b.addrs, prevPinAddr)
+	b.mu.Unlock()
+}

clientv3/client.go

@@ -121,6 +121,12 @@ func (c *Client) SetEndpoints(eps ...string) {
 	c.balancer.updateAddrs(eps)
 }
 
+// SwitchEndpoint make balancer change it's fix endpoint.
+// It is called when the endpoint now is not available, like network partition happens
+func (c *Client) SwitchEndpoint() {
+	c.balancer.switchEndpoint()
+}
+
 // Sync synchronizes client's endpoints with the known endpoints from the etcd membership.
 func (c *Client) Sync(ctx context.Context) error {
 	mresp, err := c.MemberList(ctx)

clientv3/retry.go

@@ -36,6 +36,14 @@ func (c *Client) newRetryWrapper() retryRpcFunc {
 			eErr := rpctypes.Error(err)
 			// always stop retry on etcd errors
 			if _, ok := eErr.(rpctypes.EtcdError); ok {
+				if grpc.Code(err) != codes.Unavailable {
+					// try to request another endpoint at next rpc
+					// leave this not retry
+					// already do anti-too-frequently switch
+					// at most switch one time per 30 seconds
+					// can be securely called by concurrently
+					c.SwitchEndpoint()
+				}
 				return err
 			}
 
@@ -44,6 +52,7 @@ func (c *Client) newRetryWrapper() retryRpcFunc {
 				return err
 			}
 
+
 			select {
 			case <-c.balancer.ConnectNotify():
 			case <-rpcCtx.Done():