e2e: debugging TestCtlV3DelTimeout

clientv3/balancer.go

@@ -16,6 +16,7 @@ package clientv3
 
 import (
 	"context"
+	"fmt"
 	"net/url"
 	"strings"
 	"sync"
@@ -115,6 +116,7 @@ func newSimpleBalancer(eps []string) *simpleBalancer {
 		updateAddrsC: make(chan notifyMsg),
 		host2ep:      getHost2ep(eps),
 	}
+	fmt.Println("newSimpleBalancer close(sb.downc)")
 	close(sb.downc)
 	go sb.updateNotifyLoop()
 	return sb
@@ -190,14 +192,19 @@ func (b *simpleBalancer) next() {
 	b.mu.RLock()
 	downc := b.downc
 	b.mu.RUnlock()
+	fmt.Printf("---------- next b.updateAddrsC <-notifyNext 1\n")
 	select {
 	case b.updateAddrsC <- notifyNext:
+		fmt.Printf("---------- next b.updateAddrsC <-notifyNext 2\n")
 	case <-b.stopc:
+		fmt.Printf("---------- next <-b.stopc a\n")
 	}
 	// wait until disconnect so new RPCs are not issued on old connection
 	select {
 	case <-downc:
+		fmt.Printf("---------- next <-downc\n")
 	case <-b.stopc:
+		fmt.Printf("---------- next <-b.stopc b\n")
 	}
 }
 
@@ -230,33 +237,42 @@ func (b *simpleBalancer) updateNotifyLoop() {
 		}
 		switch {
 		case downc == nil && upc == nil:
+			fmt.Printf("################ updateNotifyLoop downc == nil && upc == nil (%v)\n", addr)
 			// stale
 			select {
 			case <-b.stopc:
 				return
 			default:
 			}
 		case downc == nil:
+			fmt.Printf("################ updateNotifyLoop downc == nil (%v)\n", addr)
 			b.notifyAddrs(notifyReset)
 			select {
 			case <-upc:
+				fmt.Printf("################ updateNotifyLoop downc == nil <-upc (%v)\n", addr)
 			case msg := <-b.updateAddrsC:
+				fmt.Printf("################ updateNotifyLoop downc == nil <-b.updateAddrsC (%v) (msg %v)\n", addr, msg)
 				b.notifyAddrs(msg)
 			case <-b.stopc:
 				return
 			}
 		case upc == nil:
+			fmt.Printf("################ updateNotifyLoop upc == nil (%v)\n", addr)
 			select {
 			// close connections that are not the pinned address
 			case b.notifyCh <- []grpc.Address{{Addr: addr}}:
+				fmt.Printf("################ updateNotifyLoop upc == nil b.notifyCh <- %v\n", addr)
 			case <-downc:
+				fmt.Printf("################ updateNotifyLoop upc == nil b.notifyCh <- %v <-downc\n", addr)
 			case <-b.stopc:
 				return
 			}
 			select {
 			case <-downc:
+				fmt.Printf("################ updateNotifyLoop upc == nil b.notifyCh <- %v <-downc <-downc\n", addr)
 				b.notifyAddrs(notifyReset)
 			case msg := <-b.updateAddrsC:
+				fmt.Printf("################ updateNotifyLoop upc == nil b.notifyCh <- %v <-downc <-b.updateAddrsC %v\n", addr, msg)
 				b.notifyAddrs(msg)
 			case <-b.stopc:
 				return
@@ -269,7 +285,9 @@ func (b *simpleBalancer) notifyAddrs(msg notifyMsg) {
 	if msg == notifyNext {
 		select {
 		case b.notifyCh <- []grpc.Address{}:
+			fmt.Printf("=======notifyAddrs notifyNext []grpc.Address{}\n")
 		case <-b.stopc:
+			fmt.Printf("=======notifyAddrs notifyNext <-b.stopc\n")
 			return
 		}
 	}
@@ -278,7 +296,9 @@ func (b *simpleBalancer) notifyAddrs(msg notifyMsg) {
 	b.mu.RUnlock()
 	select {
 	case b.notifyCh <- addrs:
+		fmt.Printf("=======notifyAddrs b.notifyCh <- %v\n", addrs)
 	case <-b.stopc:
+		fmt.Printf("=======notifyAddrs <-b.stopc\n")
 	}
 }
 
@@ -291,6 +311,8 @@ func (b *simpleBalancer) up(addr grpc.Address) (func(error), bool) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
+	fmt.Printf("up %v\n", addr)
+
 	// gRPC might call Up after it called Close. We add this check
 	// to "fix" it up at application layer. Otherwise, will panic
 	// if b.upc is already closed.
@@ -306,16 +328,21 @@ func (b *simpleBalancer) up(addr grpc.Address) (func(error), bool) {
 		return func(err error) {}, false
 	}
 	// notify waiting Get()s and pin first connected address
+	fmt.Println("up close(b.upc)")
 	close(b.upc)
 	b.downc = make(chan struct{})
 	b.pinAddr = addr.Addr
+	fmt.Printf("up pinAddr %v\n", addr)
+
 	// notify client that a connection is up
 	b.readyOnce.Do(func() { close(b.readyc) })
 	return func(err error) {
 		b.mu.Lock()
 		b.upc = make(chan struct{})
+		fmt.Println("up close(b.downc)")
 		close(b.downc)
 		b.pinAddr = ""
+		fmt.Printf("up error %v (%v)\n", addr, err)
 		b.mu.Unlock()
 	}, true
 }
@@ -393,6 +420,7 @@ func (b *simpleBalancer) Close() error {
 	case <-b.upc:
 	default:
 		// terminate all waiting Get()s
+		fmt.Println("Close close(b.upc)")
 		close(b.upc)
 	}
 

clientv3/health_balancer.go

@@ -174,17 +174,21 @@ func (hb *healthBalancer) mayPin(addr grpc.Address) bool {
 	skip := len(hb.addrs) == 1 || len(hb.unhealthy) == 0
 	_, bad := hb.unhealthy[addr.Addr]
 	hb.mu.RUnlock()
+	logger.Infof("healthBalancer mayPin %v (skip %v, bad %d)", addr, skip, bad)
 	if skip || !bad {
 		return true
 	}
-	if ok, _ := hb.healthCheck(addr.Addr); ok {
+	ok, err := hb.healthCheck(addr.Addr)
+	if ok {
+		logger.Infof("healthBalancer mayPin healthCheck %v (%v)", addr, err)
 		hb.mu.Lock()
 		delete(hb.unhealthy, addr.Addr)
 		hb.mu.Unlock()
 		return true
 	}
 	hb.mu.Lock()
 	hb.unhealthy[addr.Addr] = time.Now()
+	logger.Infof("healthBalancer mayPin unhealthy %v (%v)", addr, err)
 	hb.mu.Unlock()
 	return false
 }

clientv3/retry.go

@@ -16,6 +16,7 @@ package clientv3
 
 import (
 	"context"
+	"fmt"
 
 	"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
 	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
@@ -55,18 +56,24 @@ func (c *Client) newRetryWrapper(isStop retryStopErrFunc) retryRpcFunc {
 			if err == nil {
 				return nil
 			}
+			fmt.Printf("newRetryWrapper %+v\n", err)
 			notify := c.balancer.ConnectNotify()
 			if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable {
+				fmt.Printf("\nnewRetryWrapper c.balancer.next() 1\n")
 				c.balancer.next()
+				fmt.Printf("\nnewRetryWrapper c.balancer.next() 2\n")
 			}
 			if isStop(err) {
 				return err
 			}
 			select {
 			case <-notify:
+				fmt.Printf("newRetryWrapper <-notify\n")
 			case <-rpcCtx.Done():
+				fmt.Printf("newRetryWrapper <-rpcCtx.Done()\n")
 				return rpcCtx.Err()
 			case <-c.ctx.Done():
+				fmt.Printf("newRetryWrapper <-c.ctx.Done()\n")
 				return c.ctx.Err()
 			}
 		}