Merge pull request #17896 from ivanvc/release-3.4-fix-grpc-proxy-memb…

…er-list-not-updated-with-node-deletion

[3.4] fix grpc proxy member list not updated with node deletion

proxy/grpcproxy/cluster.go

@@ -105,9 +105,9 @@ func (cp *clusterProxy) monitor(wc endpoints.WatchChannel) {
 			for _, up := range updates {
 				switch up.Op {
 				case endpoints.Add:
-					cp.umap[up.Endpoint.Addr] = up.Endpoint
+					cp.umap[up.Key] = up.Endpoint
 				case endpoints.Delete:
-					delete(cp.umap, up.Endpoint.Addr)
+					delete(cp.umap, up.Key)
 				}
 			}
 			cp.umu.Unlock()
@@ -162,12 +162,12 @@ func (cp *clusterProxy) membersFromUpdates() ([]*pb.Member, error) {
 	cp.umu.RLock()
 	defer cp.umu.RUnlock()
 	mbs := make([]*pb.Member, 0, len(cp.umap))
-	for addr, upt := range cp.umap {
+	for _, upt := range cp.umap {
 		m, err := decodeMeta(fmt.Sprint(upt.Metadata))
 		if err != nil {
 			return nil, err
 		}
-		mbs = append(mbs, &pb.Member{Name: m.Name, ClientURLs: []string{addr}})
+		mbs = append(mbs, &pb.Member{Name: m.Name, ClientURLs: []string{upt.Addr}})
 	}
 	return mbs, nil
 }

proxy/grpcproxy/cluster_test.go

@@ -17,10 +17,13 @@ package grpcproxy
 import (
 	"context"
 	"net"
+	"os"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"go.etcd.io/etcd/clientv3"
+	"go.etcd.io/etcd/clientv3/naming/endpoints"
 	pb "go.etcd.io/etcd/etcdserver/etcdserverpb"
 	"go.etcd.io/etcd/integration"
 	"go.etcd.io/etcd/pkg/testutil"
@@ -34,7 +37,10 @@ func TestClusterProxyMemberList(t *testing.T) {
 	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
 	defer clus.Terminate(t)
 
-	cts := newClusterProxyServer([]string{clus.Members[0].GRPCAddr()}, t)
+	serverEps := []string{clus.Members[0].GRPCAddr()}
+	prefix := "test-prefix"
+	hostname, _ := os.Hostname()
+	cts := newClusterProxyServer(serverEps, prefix, t)
 	defer cts.close(t)
 
 	cfg := clientv3.Config{
@@ -48,7 +54,7 @@ func TestClusterProxyMemberList(t *testing.T) {
 	defer client.Close()
 
 	// wait some time for register-loop to write keys
-	time.Sleep(time.Second)
+	time.Sleep(200 * time.Millisecond)
 
 	var mresp *clientv3.MemberListResponse
 	mresp, err = client.Cluster.MemberList(context.Background())
@@ -62,9 +68,38 @@ func TestClusterProxyMemberList(t *testing.T) {
 	if len(mresp.Members[0].ClientURLs) != 1 {
 		t.Fatalf("len(mresp.Members[0].ClientURLs) expected 1, got %d (%+v)", len(mresp.Members[0].ClientURLs), mresp.Members[0].ClientURLs[0])
 	}
-	if mresp.Members[0].ClientURLs[0] != cts.caddr {
-		t.Fatalf("mresp.Members[0].ClientURLs[0] expected %q, got %q", cts.caddr, mresp.Members[0].ClientURLs[0])
+	assert.Contains(t, mresp.Members, &pb.Member{Name: hostname, ClientURLs: []string{cts.caddr}})
+
+	//test proxy member add
+	newMemberAddr := "127.0.0.2:6789"
+	Register(cts.c, prefix, newMemberAddr, 7)
+	// wait some time for proxy update members
+	time.Sleep(200 * time.Millisecond)
+
+	//check add member succ
+	mresp, err = client.Cluster.MemberList(context.Background())
+	if err != nil {
+		t.Fatalf("err %v, want nil", err)
+	}
+	if len(mresp.Members) != 2 {
+		t.Fatalf("len(mresp.Members) expected 2, got %d (%+v)", len(mresp.Members), mresp.Members)
+	}
+	assert.Contains(t, mresp.Members, &pb.Member{Name: hostname, ClientURLs: []string{newMemberAddr}})
+
+	//test proxy member delete
+	deregisterMember(cts.c, prefix, newMemberAddr, t)
+	// wait some time for proxy update members
+	time.Sleep(200 * time.Millisecond)
+
+	//check delete member succ
+	mresp, err = client.Cluster.MemberList(context.Background())
+	if err != nil {
+		t.Fatalf("err %v, want nil", err)
+	}
+	if len(mresp.Members) != 1 {
+		t.Fatalf("len(mresp.Members) expected 1, got %d (%+v)", len(mresp.Members), mresp.Members)
 	}
+	assert.Contains(t, mresp.Members, &pb.Member{Name: hostname, ClientURLs: []string{cts.caddr}})
 }
 
 type clusterproxyTestServer struct {
@@ -88,7 +123,7 @@ func (cts *clusterproxyTestServer) close(t *testing.T) {
 	}
 }
 
-func newClusterProxyServer(endpoints []string, t *testing.T) *clusterproxyTestServer {
+func newClusterProxyServer(endpoints []string, prefix string, t *testing.T) *clusterproxyTestServer {
 	cfg := clientv3.Config{
 		Endpoints:   endpoints,
 		DialTimeout: 5 * time.Second,
@@ -113,8 +148,8 @@ func newClusterProxyServer(endpoints []string, t *testing.T) *clusterproxyTestSe
 		cts.server.Serve(cts.l)
 	}()
 
-	Register(client, "test-prefix", cts.l.Addr().String(), 7)
-	cts.cp, cts.donec = NewClusterProxy(client, cts.l.Addr().String(), "test-prefix")
+	Register(client, prefix, cts.l.Addr().String(), 7)
+	cts.cp, cts.donec = NewClusterProxy(client, cts.l.Addr().String(), prefix)
 	cts.caddr = cts.l.Addr().String()
 	pb.RegisterClusterServer(cts.server, cts.cp)
 	close(servec)
@@ -124,3 +159,13 @@ func newClusterProxyServer(endpoints []string, t *testing.T) *clusterproxyTestSe
 
 	return cts
 }
+
+func deregisterMember(c *clientv3.Client, prefix, addr string, t *testing.T) {
+	em, err := endpoints.NewManager(c, prefix)
+	if err != nil {
+		t.Fatalf("new endpoint manager failed, err %v", err)
+	}
+	if err = em.DeleteEndpoint(c.Ctx(), prefix+"/"+addr); err != nil {
+		t.Fatalf("delete endpoint failed, err %v", err)
+	}
+}