Merge pull request #8659 from gyuho/pinned

clientv3: add pinned() method to 'balancer'

clientv3/balancer.go

@@ -42,6 +42,8 @@ type balancer interface {
 
 	endpoint(host string) string
 	endpoints() []string
+	// pinned returns the current pinned endpoint.
+	pinned() string
 
 	// up is Up but includes whether the balancer will use the connection.
 	up(addr grpc.Address) (func(error), bool)
@@ -142,6 +144,12 @@ func (b *simpleBalancer) endpoints() []string {
 	return b.eps
 }
 
+func (b *simpleBalancer) pinned() string {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+	return b.pinAddr
+}
+
 func getHost2ep(eps []string) map[string]string {
 	hm := make(map[string]string, len(eps))
 	for i := range eps {

clientv3/retry.go

@@ -58,12 +58,13 @@ func (c *Client) newRetryWrapper(isStop retryStopErrFunc) retryRpcFunc {
 			case <-c.ctx.Done():
 				return c.ctx.Err()
 			}
+			pinned := c.balancer.pinned()
 			err := f(rpcCtx)
 			if err == nil {
 				return nil
 			}
 			if logger.V(4) {
-				logger.Infof("clientv3/retry: retry for error %v", err)
+				logger.Infof("clientv3/retry: error %v on pinned endpoint %s", err, pinned)
 			}
 			notify := c.balancer.ConnectNotify()
 			if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable {
@@ -86,12 +87,13 @@ func (c *Client) newRetryWrapper(isStop retryStopErrFunc) retryRpcFunc {
 func (c *Client) newAuthRetryWrapper() retryRpcFunc {
 	return func(rpcCtx context.Context, f rpcFunc) error {
 		for {
+			pinned := c.balancer.pinned()
 			err := f(rpcCtx)
 			if err == nil {
 				return nil
 			}
 			if logger.V(4) {
-				logger.Infof("clientv3/auth-retry: retry for error %v", err)
+				logger.Infof("clientv3/auth-retry: error %v on pinned endpoint %s", err, pinned)
 			}
 			// always stop retry on etcd errors other than invalid auth token
 			if rpctypes.Error(err) == rpctypes.ErrInvalidAuthToken {