Merge pull request #14170 from ahrtr/3.4_proxy_fix_20220628

Fix deadlock in 'go test -tags cluster_proxy -v ./integration/... ./client'

clientv3/naming/grpc_test.go

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package naming
+package naming_test
 
 import (
 	"context"
@@ -21,6 +21,7 @@ import (
 	"testing"
 
 	etcd "go.etcd.io/etcd/clientv3"
+	namingv3 "go.etcd.io/etcd/clientv3/naming"
 	"go.etcd.io/etcd/integration"
 	"go.etcd.io/etcd/pkg/testutil"
 
@@ -33,7 +34,7 @@ func TestGRPCResolver(t *testing.T) {
 	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
 	defer clus.Terminate(t)
 
-	r := GRPCResolver{
+	r := namingv3.GRPCResolver{
 		Client: clus.RandClient(),
 	}
 
@@ -107,7 +108,7 @@ func TestGRPCResolverMulti(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	r := GRPCResolver{c}
+	r := namingv3.GRPCResolver{c}
 
 	w, err := r.Resolve("foo")
 	if err != nil {

etcdmain/grpc_proxy.go

@@ -350,12 +350,12 @@ func newGRPCProxyServer(lg *zap.Logger, client *clientv3.Client) *grpc.Server {
 	}
 
 	kvp, _ := grpcproxy.NewKvProxy(client)
-	watchp, _ := grpcproxy.NewWatchProxy(client)
+	watchp, _ := grpcproxy.NewWatchProxy(client.Ctx(), client)
 	if grpcProxyResolverPrefix != "" {
 		grpcproxy.Register(client, grpcProxyResolverPrefix, grpcProxyAdvertiseClientURL, grpcProxyResolverTTL)
 	}
 	clusterp, _ := grpcproxy.NewClusterProxy(client, grpcProxyAdvertiseClientURL, grpcProxyResolverPrefix)
-	leasep, _ := grpcproxy.NewLeaseProxy(client)
+	leasep, _ := grpcproxy.NewLeaseProxy(client.Ctx(), client)
 	mainp := grpcproxy.NewMaintenanceProxy(client)
 	authp := grpcproxy.NewAuthProxy(client)
 	electionp := grpcproxy.NewElectionProxy(client)

integration/cluster_proxy.go

@@ -17,6 +17,7 @@
 package integration
 
 import (
+	"context"
 	"sync"
 
 	"go.etcd.io/etcd/clientv3"
@@ -35,16 +36,23 @@ var (
 const proxyNamespace = "proxy-namespace"
 
 type grpcClientProxy struct {
-	grpc    grpcAPI
+	ctx       context.Context
-	wdonec  <-chan struct{}
+	ctxCancel func()
-	kvdonec <-chan struct{}
+	grpc      grpcAPI
-	lpdonec <-chan struct{}
+	wdonec    <-chan struct{}
+	kvdonec   <-chan struct{}
+	lpdonec   <-chan struct{}
 }
 
 func toGRPC(c *clientv3.Client) grpcAPI {
 	pmu.Lock()
 	defer pmu.Unlock()
 
+	// dedicated context bound to 'grpc-proxy' lifetype
+	// (so in practice lifetime of the client connection to the proxy).
+	// TODO: Refactor to a separate clientv3.Client instance instead of the context alone.
+	ctx, ctxCancel := context.WithCancel(context.WithValue(context.TODO(), "_name", "grpcProxyContext"))
+
 	if v, ok := proxies[c]; ok {
 		return v.grpc
 	}
@@ -55,8 +63,8 @@ func toGRPC(c *clientv3.Client) grpcAPI {
 	c.Lease = namespace.NewLease(c.Lease, proxyNamespace)
 	// test coalescing/caching proxy
 	kvp, kvpch := grpcproxy.NewKvProxy(c)
-	wp, wpch := grpcproxy.NewWatchProxy(c)
+	wp, wpch := grpcproxy.NewWatchProxy(ctx, c)
-	lp, lpch := grpcproxy.NewLeaseProxy(c)
+	lp, lpch := grpcproxy.NewLeaseProxy(ctx, c)
 	mp := grpcproxy.NewMaintenanceProxy(c)
 	clp, _ := grpcproxy.NewClusterProxy(c, "", "") // without registering proxy URLs
 	authp := grpcproxy.NewAuthProxy(c)
@@ -73,20 +81,21 @@ func toGRPC(c *clientv3.Client) grpcAPI {
 		adapter.LockServerToLockClient(lockp),
 		adapter.ElectionServerToElectionClient(electp),
 	}
-	proxies[c] = grpcClientProxy{grpc: grpc, wdonec: wpch, kvdonec: kvpch, lpdonec: lpch}
+	proxies[c] = grpcClientProxy{ctx: ctx, ctxCancel: ctxCancel, grpc: grpc, wdonec: wpch, kvdonec: kvpch, lpdonec: lpch}
 	return grpc
 }
 
 type proxyCloser struct {
 	clientv3.Watcher
-	wdonec  <-chan struct{}
+	proxyCtxCancel func()
-	kvdonec <-chan struct{}
+	wdonec         <-chan struct{}
-	lclose  func()
+	kvdonec        <-chan struct{}
-	lpdonec <-chan struct{}
+	lclose         func()
+	lpdonec        <-chan struct{}
 }
 
 func (pc *proxyCloser) Close() error {
-	// client ctx is canceled before calling close, so kv and lp will close out
+	pc.proxyCtxCancel()
 	<-pc.kvdonec
 	err := pc.Watcher.Close()
 	<-pc.wdonec
@@ -106,11 +115,12 @@ func newClientV3(cfg clientv3.Config) (*clientv3.Client, error) {
 	lc := c.Lease
 	c.Lease = clientv3.NewLeaseFromLeaseClient(rpc.Lease, c, cfg.DialTimeout)
 	c.Watcher = &proxyCloser{
-		Watcher: clientv3.NewWatchFromWatchClient(rpc.Watch, c),
+		Watcher:        clientv3.NewWatchFromWatchClient(rpc.Watch, c),
-		wdonec:  proxies[c].wdonec,
+		wdonec:         proxies[c].wdonec,
-		kvdonec: proxies[c].kvdonec,
+		kvdonec:        proxies[c].kvdonec,
-		lclose:  func() { lc.Close() },
+		lclose:         func() { lc.Close() },
-		lpdonec: proxies[c].lpdonec,
+		lpdonec:        proxies[c].lpdonec,
+		proxyCtxCancel: proxies[c].ctxCancel,
 	}
 	pmu.Unlock()
 	return c, nil

proxy/grpcproxy/lease.go

@@ -48,13 +48,13 @@ type leaseProxy struct {
 	wg sync.WaitGroup
 }
 
-func NewLeaseProxy(c *clientv3.Client) (pb.LeaseServer, <-chan struct{}) {
+func NewLeaseProxy(ctx context.Context, c *clientv3.Client) (pb.LeaseServer, <-chan struct{}) {
-	cctx, cancel := context.WithCancel(c.Ctx())
+	cctx, cancel := context.WithCancel(ctx)
 	lp := &leaseProxy{
 		leaseClient: pb.NewLeaseClient(c.ActiveConnection()),
 		lessor:      c.Lease,
 		ctx:         cctx,
-		leader:      newLeader(c.Ctx(), c.Watcher),
+		leader:      newLeader(ctx, c.Watcher),
 	}
 	ch := make(chan struct{})
 	go func() {

proxy/grpcproxy/watch.go

@@ -46,12 +46,12 @@ type watchProxy struct {
 	kv clientv3.KV
 }
 
-func NewWatchProxy(c *clientv3.Client) (pb.WatchServer, <-chan struct{}) {
+func NewWatchProxy(ctx context.Context, c *clientv3.Client) (pb.WatchServer, <-chan struct{}) {
-	cctx, cancel := context.WithCancel(c.Ctx())
+	cctx, cancel := context.WithCancel(ctx)
 	wp := &watchProxy{
 		cw:     c.Watcher,
 		ctx:    cctx,
-		leader: newLeader(c.Ctx(), c.Watcher),
+		leader: newLeader(ctx, c.Watcher),
 
 		kv: c.KV, // for permission checking
 	}