-
Notifications
You must be signed in to change notification settings - Fork 9.8k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
*: implement Endpoint Watch and new Resolver
- Loading branch information
limeng01
committed
Feb 8, 2021
1 parent
dae29bb
commit 8feb55f
Showing
7 changed files
with
198 additions
and
96 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
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
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 |
|---|---|---|
| @@ -1,24 +1,107 @@ | ||
| package resolver | ||
|
|
||
| import ( | ||
| "context" | ||
| "sync" | ||
|
|
||
| clientv3 "go.etcd.io/etcd/client/v3" | ||
| "google.golang.org/grpc/resolver" | ||
| "go.etcd.io/etcd/client/v3/naming/endpoints" | ||
|
|
||
| "google.golang.org/grpc/codes" | ||
| gresolver "google.golang.org/grpc/resolver" | ||
| "google.golang.org/grpc/status" | ||
| ) | ||
|
|
||
| type builder struct { | ||
| // ... | ||
| c *clientv3.Client | ||
| } | ||
|
|
||
| func (b builder) Build(target resolver.Target, cc resolver.ClientConn, opts resolver.BuildOptions) (resolver.Resolver, error) { | ||
| // To be implemented... | ||
| // Using endpoints.NewWatcher() to subscribe for endpoints changes. | ||
| return nil, nil | ||
| func (b builder) Build(target gresolver.Target, cc gresolver.ClientConn, opts gresolver.BuildOptions) (gresolver.Resolver, error) { | ||
| r := &resolver{ | ||
| c: b.c, | ||
| target: target.Endpoint, | ||
| cc: cc, | ||
| } | ||
| r.ctx, r.cancel = context.WithCancel(context.Background()) | ||
|
|
||
| em, err := endpoints.NewManager(r.c, r.target) | ||
| if err != nil { | ||
| return nil, status.Errorf(codes.InvalidArgument, "resolver: failed to new endpoint manager: %s", err) | ||
| } | ||
| r.wch, err = em.NewWatchChannel(r.ctx) | ||
| if err != nil { | ||
| return nil, status.Errorf(codes.Internal, "resolver: failed to new watch channer: %s", err) | ||
| } | ||
|
|
||
| r.wg.Add(1) | ||
| go r.watch() | ||
| return r, nil | ||
| } | ||
|
|
||
| func (b builder) Scheme() string { | ||
| return "etcd" | ||
| } | ||
|
|
||
| func NewBuilder(client *clientv3.Client) (resolver.Builder, error) { | ||
| return builder{}, nil | ||
| // NewBuilder creates a resolver builder. | ||
| func NewBuilder(client *clientv3.Client) (gresolver.Builder, error) { | ||
| return builder{c: client}, nil | ||
| } | ||
|
|
||
| type resolver struct { | ||
| c *clientv3.Client | ||
| target string | ||
| cc gresolver.ClientConn | ||
| wch endpoints.WatchChannel | ||
| ctx context.Context | ||
| cancel context.CancelFunc | ||
| wg sync.WaitGroup | ||
| } | ||
|
|
||
| func (r *resolver) watch() { | ||
| defer r.wg.Done() | ||
|
|
||
| allUps := make(map[string]*endpoints.Update) | ||
| for { | ||
| select { | ||
| case <-r.ctx.Done(): | ||
| return | ||
| case ups, ok := <-r.wch: | ||
| if !ok { | ||
| return | ||
| } | ||
|
|
||
| for _, up := range ups { | ||
| switch up.Op { | ||
| case endpoints.Add: | ||
| allUps[up.Key] = up | ||
| case endpoints.Delete: | ||
| delete(allUps, up.Key) | ||
| } | ||
| } | ||
|
|
||
| addrs := convertToGRPCAddress(allUps) | ||
| r.cc.UpdateState(gresolver.State{Addresses: addrs}) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func convertToGRPCAddress(ups map[string]*endpoints.Update) []gresolver.Address { | ||
| var addrs []gresolver.Address | ||
| for _, up := range ups { | ||
| addr := gresolver.Address{ | ||
| Addr: up.Endpoint.Addr, | ||
| Metadata: up.Endpoint.Metadata, | ||
| } | ||
| addrs = append(addrs, addr) | ||
| } | ||
| return addrs | ||
| } | ||
|
|
||
| // ResolveNow is a no-op here. | ||
| // It's just a hint, resolver can ignore this if it's not necessary. | ||
| func (r *resolver) ResolveNow(gresolver.ResolveNowOptions) {} | ||
|
|
||
| func (r *resolver) Close() { | ||
| r.cancel() | ||
| r.wg.Wait() | ||
| } |
Oops, something went wrong.