Merge pull request #15835 from yellowzf/grpcproxy_fix_memberlist_results_not_update_when_proxy_node_down

grpcproxy: fix memberlist results not update when proxy node down

server/proxy/grpcproxy/cluster.go

@@ -113,9 +113,9 @@ func (cp *clusterProxy) monitor(wa 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()
@@ -139,12 +139,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
 }

tests/integration/proxy/grpcproxy/cluster_test.go

@@ -17,15 +17,18 @@ package grpcproxy
 import (
 	"context"
 	"net"
+	"os"
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/assert"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zaptest"
 	"google.golang.org/grpc"
 
 	pb "go.etcd.io/etcd/api/v3/etcdserverpb"
 	clientv3 "go.etcd.io/etcd/client/v3"
+	"go.etcd.io/etcd/client/v3/naming/endpoints"
 	"go.etcd.io/etcd/server/v3/proxy/grpcproxy"
 	integration2 "go.etcd.io/etcd/tests/v3/framework/integration"
 )
@@ -36,7 +39,11 @@ func TestClusterProxyMemberList(t *testing.T) {
 	clus := integration2.NewCluster(t, &integration2.ClusterConfig{Size: 1})
 	defer clus.Terminate(t)
 
-	cts := newClusterProxyServer(zaptest.NewLogger(t), []string{clus.Members[0].GRPCURL()}, t)
+	lg := zaptest.NewLogger(t)
+	serverEps := []string{clus.Members[0].GRPCURL()}
+	prefix := "test-prefix"
+	hostname, _ := os.Hostname()
+	cts := newClusterProxyServer(lg, serverEps, prefix, t)
 	defer cts.close(t)
 
 	cfg := clientv3.Config{
@@ -50,7 +57,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())
@@ -64,9 +71,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 {
+	assert.Contains(t, mresp.Members, &pb.Member{Name: hostname, ClientURLs: []string{cts.caddr}})
-		t.Fatalf("mresp.Members[0].ClientURLs[0] expected %q, got %q", cts.caddr, mresp.Members[0].ClientURLs[0])
+
+	//test proxy member add
+	newMemberAddr := "127.0.0.2:6789"
+	grpcproxy.Register(lg, 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 {
@@ -90,7 +126,7 @@ func (cts *clusterproxyTestServer) close(t *testing.T) {
 	}
 }
 
-func newClusterProxyServer(lg *zap.Logger, endpoints []string, t *testing.T) *clusterproxyTestServer {
+func newClusterProxyServer(lg *zap.Logger, endpoints []string, prefix string, t *testing.T) *clusterproxyTestServer {
 	cfg := clientv3.Config{
 		Endpoints:   endpoints,
 		DialTimeout: 5 * time.Second,
@@ -115,8 +151,8 @@ func newClusterProxyServer(lg *zap.Logger, endpoints []string, t *testing.T) *cl
 		cts.server.Serve(cts.l)
 	}()
 
-	grpcproxy.Register(lg, client, "test-prefix", cts.l.Addr().String(), 7)
+	grpcproxy.Register(lg, client, prefix, cts.l.Addr().String(), 7)
-	cts.cp, cts.donec = grpcproxy.NewClusterProxy(lg, client, cts.l.Addr().String(), "test-prefix")
+	cts.cp, cts.donec = grpcproxy.NewClusterProxy(lg, client, cts.l.Addr().String(), prefix)
 	cts.caddr = cts.l.Addr().String()
 	pb.RegisterClusterServer(cts.server, cts.cp)
 	close(servec)
@@ -126,3 +162,13 @@ func newClusterProxyServer(lg *zap.Logger, endpoints []string, t *testing.T) *cl
 
 	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)
+	}
+}