forked from etcd-io/etcd
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Introduce grpc-1.30+ compatible client/v3/naming API.
This is not yet implementation, just API and tests to be filled with implementation in next CLs, tracked by: etcd-io#12652 We propose here 3 packages: - clientv3/naming/endpoints -> That is abstraction layer over etcd that allows to write, read & watch Endpoints information. It's independent from GRPC API. It hides the storage details. - clientv3/naming/endpoints/internal -> That contains the grpc's compatible Update class to preserve the internal JSON mashalling format. - clientv3/naming/resolver -> That implements the GRPC resolver API, such that etcd can be used for connection.Dial in grpc. Please see the grpc_naming.md document changes & grpcproxy/cluster.go new integration, to see how the new abstractions work. Signed-off-by: Chao Chen <chaochn@amazon.com>
- Loading branch information
Showing
12 changed files
with
590 additions
and
38 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| // Copyright 2016 The etcd Authors | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package naming | ||
|
|
||
| import ( | ||
| "context" | ||
| "reflect" | ||
| "testing" | ||
|
|
||
| etcd "go.etcd.io/etcd/clientv3" | ||
| "go.etcd.io/etcd/clientv3/naming/endpoints" | ||
|
|
||
| "go.etcd.io/etcd/integration" | ||
| "go.etcd.io/etcd/pkg/testutil" | ||
| ) | ||
|
|
||
| func TestEndpointManager(t *testing.T) { | ||
| t.Skip("Not implemented yet") | ||
|
|
||
| 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) | ||
| } | ||
| ctx, watchCancel := context.WithCancel(context.Background()) | ||
| defer watchCancel() | ||
| w, err := em.NewWatchChannel(ctx) | ||
| if err != nil { | ||
| t.Fatal("failed to establish watch", err) | ||
| } | ||
|
|
||
| e1 := endpoints.Endpoint{Addr: "127.0.0.1", Metadata: "metadata"} | ||
| err = em.AddEndpoint(context.TODO(), "foo/a1", e1) | ||
| if err != nil { | ||
| t.Fatal("failed to add foo", err) | ||
| } | ||
|
|
||
| us := <-w | ||
|
|
||
| if us == nil { | ||
| t.Fatal("failed to get update", err) | ||
| } | ||
|
|
||
| wu := endpoints.Update{ | ||
| Op: endpoints.Add, | ||
| Key: "foo/a1", | ||
| Endpoint: e1, | ||
| } | ||
|
|
||
| if !reflect.DeepEqual(us[0], wu) { | ||
| t.Fatalf("up = %#v, want %#v", us[0], wu) | ||
| } | ||
|
|
||
| err = em.DeleteEndpoint(context.TODO(), "foo/a1") | ||
| if err != nil { | ||
| t.Fatalf("failed to udpate %v", err) | ||
| } | ||
|
|
||
| us = <-w | ||
| if err != nil { | ||
| t.Fatalf("failed to get udpate %v", err) | ||
| } | ||
|
|
||
| wu = endpoints.Update{ | ||
| Op: endpoints.Delete, | ||
| Key: "foo/a1", | ||
| } | ||
|
|
||
| if !reflect.DeepEqual(us, wu) { | ||
| t.Fatalf("up = %#v, want %#v", us[1], wu) | ||
| } | ||
| } | ||
|
|
||
| // TestEndpointManagerAtomicity ensures the resolver will initialize | ||
| // correctly with multiple hosts and correctly receive multiple | ||
| // updates in a single revision. | ||
| func TestEndpointManagerAtomicity(t *testing.T) { | ||
| t.Skip("Not implemented yet") | ||
|
|
||
| defer testutil.AfterTest(t) | ||
|
|
||
| clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1}) | ||
| defer clus.Terminate(t) | ||
|
|
||
| c := clus.RandClient() | ||
| em, err := endpoints.NewManager(c, "foo") | ||
| if err != nil { | ||
| t.Fatal("failed to create EndpointManager", err) | ||
| } | ||
|
|
||
| err = em.Update(context.TODO(), []*endpoints.UpdateWithOpts{ | ||
| endpoints.NewAddUpdateOpts("foo/host", endpoints.Endpoint{Addr: "127.0.0.1:2000"}), | ||
| endpoints.NewAddUpdateOpts("foo/host2", endpoints.Endpoint{Addr: "127.0.0.1:2001"})}) | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
|
|
||
| ctx, watchCancel := context.WithCancel(context.Background()) | ||
| defer watchCancel() | ||
| w, err := em.NewWatchChannel(ctx) | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
|
|
||
| updates := <-w | ||
| if len(updates) != 2 { | ||
| t.Fatalf("expected two updates, got %+v", updates) | ||
| } | ||
|
|
||
| _, err = c.Txn(context.TODO()).Then(etcd.OpDelete("foo/host"), etcd.OpDelete("foo/host2")).Commit() | ||
| if err != nil { | ||
| t.Fatal(err) | ||
| } | ||
|
|
||
| updates = <-w | ||
| if len(updates) != 2 || (updates[0].Op != endpoints.Delete && updates[1].Op != endpoints.Delete) { | ||
| t.Fatalf("expected two delete updates, got %+v", updates) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| // Copyright 2016 The etcd Authors | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package naming | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "google.golang.org/grpc" | ||
|
|
||
| "go.etcd.io/etcd/clientv3/naming/endpoints" | ||
| "go.etcd.io/etcd/clientv3/naming/resolver" | ||
| "go.etcd.io/etcd/integration" | ||
| "go.etcd.io/etcd/pkg/testutil" | ||
| ) | ||
|
|
||
| // 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 | ||
|
|
||
| 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) | ||
| } | ||
|
|
||
| e1 := endpoints.Endpoint{Addr: "127.0.0.1:20000"} | ||
| e2 := endpoints.Endpoint{Addr: "127.0.0.1:20001"} | ||
|
|
||
| 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)) | ||
| if err != nil { | ||
| t.Fatal("failed to connect to foo (e1)", err) | ||
| } | ||
|
|
||
| // TODO: send requests to conn, ensure s1 received it. | ||
|
|
||
| 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. | ||
|
|
||
| conn.GetState() // this line is to avoid compiler warning that conn is unused. | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.