endpoints: implement Update method for EndpointManager.

- Add integration test for endpoints and resolver.

tests/integration/clientv3/naming/endpoints_test.go

@@ -133,3 +133,84 @@ func TestEndpointManagerAtomicity(t *testing.T) {
 		t.Fatalf("expected two delete updates, got %+v", updates)
 	}
 }
+
+func TestEndpointManagerCRUD(t *testing.T) {
+	defer testutil.AfterTest(t)
+
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	defer clus.Terminate(t)
+
+	em, err := endpoints.NewManager(clus.RandClient(), "foo")
+	if err != nil {
+		t.Fatal("failed to create EndpointManager", err)
+	}
+
+	// Add
+	k1 := "foo/a1"
+	e1 := endpoints.Endpoint{Addr: "127.0.0.1", Metadata: "metadata1"}
+	err = em.AddEndpoint(context.TODO(), k1, e1)
+	if err != nil {
+		t.Fatal("failed to add", k1, err)
+	}
+
+	k2 := "foo/a2"
+	e2 := endpoints.Endpoint{Addr: "127.0.0.2", Metadata: "metadata2"}
+	err = em.AddEndpoint(context.TODO(), k2, e2)
+	if err != nil {
+		t.Fatal("failed to add", k2, err)
+	}
+
+	eps, err := em.List(context.TODO())
+	if err != nil {
+		t.Fatal("failed to list foo")
+	}
+	if len(eps) != 2 {
+		t.Fatalf("unexpected the number of endpoints: %d", len(eps))
+	}
+	if !reflect.DeepEqual(eps[k1], e1) {
+		t.Fatalf("unexpected endpoints: %s", k1)
+	}
+	if !reflect.DeepEqual(eps[k2], e2) {
+		t.Fatalf("unexpected endpoints: %s", k2)
+	}
+
+	// Delete
+	err = em.DeleteEndpoint(context.TODO(), k1)
+	if err != nil {
+		t.Fatal("failed to delete", k2, err)
+	}
+
+	eps, err = em.List(context.TODO())
+	if err != nil {
+		t.Fatal("failed to list foo")
+	}
+	if len(eps) != 1 {
+		t.Fatalf("unexpected the number of endpoints: %d", len(eps))
+	}
+	if !reflect.DeepEqual(eps[k2], e2) {
+		t.Fatalf("unexpected endpoints: %s", k2)
+	}
+
+	// Update
+	k3 := "foo/a3"
+	e3 := endpoints.Endpoint{Addr: "127.0.0.3", Metadata: "metadata3"}
+	updates := []*endpoints.UpdateWithOpts{
+		{Update: endpoints.Update{Op: endpoints.Add, Key: k3, Endpoint: e3}},
+		{Update: endpoints.Update{Op: endpoints.Delete, Key: k2}},
+	}
+	err = em.Update(context.TODO(), updates)
+	if err != nil {
+		t.Fatal("failed to update", err)
+	}
+
+	eps, err = em.List(context.TODO())
+	if err != nil {
+		t.Fatal("failed to list foo")
+	}
+	if len(eps) != 1 {
+		t.Fatalf("unexpected the number of endpoints: %d", len(eps))
+	}
+	if !reflect.DeepEqual(eps[k3], e3) {
+		t.Fatalf("unexpected endpoints: %s", k3)
+	}
+}

tests/integration/clientv3/naming/resolver_test.go

@@ -15,56 +15,113 @@
 package naming_test
 
 import (
+	"bytes"
 	"context"
 	"testing"
+	"time"
 
+	etcdnaming "go.etcd.io/etcd/client/v3/naming"
 	"go.etcd.io/etcd/client/v3/naming/endpoints"
-	"go.etcd.io/etcd/client/v3/naming/resolver"
-	"google.golang.org/grpc"
-
+	grpctest "go.etcd.io/etcd/pkg/v3/grpc_testing"
 	"go.etcd.io/etcd/pkg/v3/testutil"
 	"go.etcd.io/etcd/tests/v3/integration"
+
+	"google.golang.org/grpc"
+	testpb "google.golang.org/grpc/test/grpc_testing"
 )
 
 // This test mimics scenario described in grpc_naming.md doc.
 
 func TestEtcdGrpcResolver(t *testing.T) {
-	t.Skip("Not implemented yet")
-
 	defer testutil.AfterTest(t)
 
-	// s1 :=  // TODO: Dummy GRPC service listening on 127.0.0.1:20000
-	// s2 :=  // TODO: Dummy GRPC service listening on 127.0.0.1:20001
+	s1PayloadBody := []byte{'1'}
+	s1 := newDummyStubServer(s1PayloadBody)
+	if err := s1.Start(nil); err != nil {
+		t.Fatal("failed to start dummy grpc server (s1)", err)
+	}
+	defer s1.Stop()
 
-	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	s2PayloadBody := []byte{'2'}
+	s2 := newDummyStubServer(s2PayloadBody)
+	if err := s2.Start(nil); err != nil {
+		t.Fatal("failed to start dummy grpc server (s2)", err)
+	}
+	defer s2.Stop()
+
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 3})
 	defer clus.Terminate(t)
 
-	em, err := endpoints.NewManager(clus.RandClient(), "foo")
+	em, err := endpoints.NewManager(clus.Client(0), "foo")
 	if err != nil {
 		t.Fatal("failed to create EndpointManager", err)
 	}
 
-	e1 := endpoints.Endpoint{Addr: "127.0.0.1:20000"}
-	e2 := endpoints.Endpoint{Addr: "127.0.0.1:20001"}
+	e1 := endpoints.Endpoint{Addr: s1.Addr()}
+	e2 := endpoints.Endpoint{Addr: s2.Addr()}
 
 	err = em.AddEndpoint(context.TODO(), "foo/e1", e1)
 	if err != nil {
 		t.Fatal("failed to add foo", err)
 	}
-	etcdResolver, err := resolver.NewBuilder(clus.RandClient())
 
-	conn, err := grpc.Dial("etc://foo", grpc.WithResolvers(etcdResolver))
+	r := &etcdnaming.GRPCResolver{Client: clus.Client(1)}
+	b := grpc.RoundRobin(r)
+
+	conn, err := grpc.Dial("foo", grpc.WithInsecure(), grpc.WithBalancer(b))
 	if err != nil {
-		t.Fatal("failed to connect to foo (e1)", err)
+		t.Fatal("failed to connect to foo", err)
 	}
+	defer conn.Close()
 
-	// TODO: send requests to conn, ensure s1 received it.
+	c := testpb.NewTestServiceClient(conn)
+	resp, err := c.UnaryCall(context.TODO(), &testpb.SimpleRequest{}, grpc.WaitForReady(true))
+	if err != nil {
+		t.Fatal("failed to invoke rpc to foo (e1)", err)
+	}
+	if resp.GetPayload() == nil || !bytes.Equal(resp.GetPayload().GetBody(), s1PayloadBody) {
+		t.Fatalf("unexpected response from foo (e1): %s", resp.GetPayload().GetBody())
+	}
 
 	em.DeleteEndpoint(context.TODO(), "foo/e1")
 	em.AddEndpoint(context.TODO(), "foo/e2", e2)
 
-	// TODO: Send requests to conn and make sure s2 receive it.
-	// Might require restarting s1 to break the existing (open) connection.
+	// We use a loop with deadline of 30s to avoid test getting flake
+	// as it's asynchronous for gRPC Client to update underlying connections.
+	maxRetries := 300
+	retryPeriod := 100 * time.Millisecond
+	retries := 0
+	for {
+		time.Sleep(retryPeriod)
+		retries++
 
-	conn.GetState() // this line is to avoid compiler warning that conn is unused.
+		resp, err = c.UnaryCall(context.TODO(), &testpb.SimpleRequest{})
+		if err != nil {
+			if retries < maxRetries {
+				continue
+			}
+			t.Fatal("failed to invoke rpc to foo (e2)", err)
+		}
+		if resp.GetPayload() == nil || !bytes.Equal(resp.GetPayload().GetBody(), s2PayloadBody) {
+			if retries < maxRetries {
+				continue
+			}
+			t.Fatalf("unexpected response from foo (e2): %s", resp.GetPayload().GetBody())
+		}
+
+		break
+	}
+}
+
+func newDummyStubServer(body []byte) *grpctest.StubServer {
+	return &grpctest.StubServer{
+		UnaryCallF: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+			return &testpb.SimpleResponse{
+				Payload: &testpb.Payload{
+					Type: testpb.PayloadType_COMPRESSABLE,
+					Body: body,
+				},
+			}, nil
+		},
+	}
 }