-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #2 from khvysofq/master
Add support for etcd client version 3
- Loading branch information
Showing
6 changed files
with
283 additions
and
41 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,109 @@ | ||
| package etcd | ||
|
|
||
| import ( | ||
| "context" | ||
| "crypto/tls" | ||
| "crypto/x509" | ||
| "io/ioutil" | ||
| "time" | ||
|
|
||
| etcdv3 "github.com/coreos/etcd/clientv3" | ||
| ) | ||
|
|
||
| type clientv3 struct { | ||
| client *etcdv3.Client | ||
| ctx context.Context | ||
| timeout time.Duration | ||
| } | ||
|
|
||
| // NewClient returns Client with a connection to the named machines. It will | ||
| // return an error if a connection to the cluster cannot be made. The parameter | ||
| // machines needs to be a full URL with schemas. e.g. "http://localhost:2379" | ||
| // will work, but "localhost:2379" will not. | ||
| func NewClientV3(ctx context.Context, machines []string, options ClientOptions) (Client, error) { | ||
| if options.DialTimeout == 0 { | ||
| options.DialTimeout = defaultTTL | ||
| } | ||
| if options.DialKeepAlive == 0 { | ||
| options.DialKeepAlive = defaultTTL | ||
| } | ||
| if options.HeaderTimeoutPerRequest == 0 { | ||
| options.HeaderTimeoutPerRequest = defaultTTL | ||
| } | ||
|
|
||
| var tlsCfg *tls.Config | ||
| if options.Cert != "" && options.Key != "" { | ||
| tlsCert, err := tls.LoadX509KeyPair(options.Cert, options.Key) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| tlsCfg = &tls.Config{ | ||
| Certificates: []tls.Certificate{tlsCert}, | ||
| } | ||
| if caCertCt, err := ioutil.ReadFile(options.CACert); err == nil { | ||
| caCertPool := x509.NewCertPool() | ||
| caCertPool.AppendCertsFromPEM(caCertCt) | ||
| tlsCfg.RootCAs = caCertPool | ||
| } | ||
| } | ||
|
|
||
| ce, err := etcdv3.New(etcdv3.Config{ | ||
| Endpoints: machines, | ||
| DialTimeout: options.DialTimeout, | ||
| DialKeepAliveTime: options.DialKeepAlive, | ||
| DialKeepAliveTimeout: options.HeaderTimeoutPerRequest, | ||
| TLS: tlsCfg, | ||
| }) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &clientv3{ | ||
| client: ce, | ||
| ctx: ctx, | ||
| timeout: options.HeaderTimeoutPerRequest, | ||
| }, nil | ||
| } | ||
|
|
||
| // GetEntries implements the etcd Client interface. | ||
| func (c *clientv3) GetEntries(key string) ([]string, error) { | ||
|
|
||
| if c.client == nil { | ||
| return nil, ErrNilClient | ||
| } | ||
|
|
||
| // set the timeout for this requisition | ||
| timeoutCtx, cancel := context.WithTimeout(c.ctx, c.timeout) | ||
| resp, err := c.client.Get(timeoutCtx, key, etcdv3.WithPrefix()) | ||
| cancel() | ||
|
|
||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| // Special case. Note that it's possible that len(resp.Node.Nodes) == 0 and | ||
| // resp.Node.Value is also empty, in which case the key is empty and we | ||
| // should not return any entries. | ||
| if len(resp.Kvs) == 0 || resp.Count != int64(len(resp.Kvs)) { | ||
| return nil, nil | ||
| } | ||
|
|
||
| entries := make([]string, resp.Count) | ||
| for i, ev := range resp.Kvs { | ||
| entries[i] = string(ev.Value[:]) | ||
| } | ||
| return entries, nil | ||
| } | ||
|
|
||
| // WatchPrefix implements the etcd Client interface. | ||
| func (c *clientv3) WatchPrefix(prefix string, ch chan struct{}) { | ||
|
|
||
| if c.client == nil { | ||
| return | ||
| } | ||
| watch := c.client.Watch(c.ctx, prefix, etcdv3.WithPrefix()) | ||
| ch <- struct{}{} // make sure caller invokes GetEntries | ||
| for _ := range watch { | ||
| ch <- struct{}{} | ||
| } | ||
| } |
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,86 @@ | ||
| package etcd | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
| "time" | ||
| ) | ||
|
|
||
| func TestNewClient_withDefaultsV3(t *testing.T) { | ||
| client, err := NewClientV3( | ||
| context.Background(), | ||
| []string{"http://irrelevant:12345"}, | ||
| ClientOptions{}, | ||
| ) | ||
| if err == nil { | ||
| t.Fatalf("unexpected error creating client: %v", err) | ||
| } | ||
| if client != nil { | ||
| t.Fatal("expected new Client, got nil") | ||
| } | ||
| } | ||
|
|
||
| // NewClient should fail when providing invalid or missing endpoints. | ||
| func TestOptionsV3(t *testing.T) { | ||
| a, err := NewClientV3( | ||
| context.Background(), | ||
| []string{}, | ||
| ClientOptions{ | ||
| Cert: "", | ||
| Key: "", | ||
| CACert: "", | ||
| DialTimeout: 2 * time.Second, | ||
| DialKeepAlive: 2 * time.Second, | ||
| HeaderTimeoutPerRequest: 2 * time.Second, | ||
| }, | ||
| ) | ||
| if err == nil { | ||
| t.Errorf("expected error: %v", err) | ||
| } | ||
| if a != nil { | ||
| t.Fatalf("expected client to be nil on failure") | ||
| } | ||
|
|
||
| _, err = NewClientV3( | ||
| context.Background(), | ||
| []string{"http://irrelevant:12345"}, | ||
| ClientOptions{ | ||
| Cert: "blank.crt", | ||
| Key: "blank.key", | ||
| CACert: "blank.CACert", | ||
| DialTimeout: 2 * time.Second, | ||
| DialKeepAlive: 2 * time.Second, | ||
| HeaderTimeoutPerRequest: 2 * time.Second, | ||
| }, | ||
| ) | ||
| if err == nil { | ||
| t.Errorf("expected error: %v", err) | ||
| } | ||
| } | ||
|
|
||
| // --------------------------------------------------------------------------------------------------------------------- | ||
|
|
||
| func newFakeClientV3(ctx context.Context) Client { | ||
| return &clientv3{ | ||
| client: nil, | ||
| ctx: ctx, | ||
| timeout: 3 * time.Second, | ||
| } | ||
| } | ||
|
|
||
| func TestWatchPrefixV3(t *testing.T) { | ||
|
|
||
| cv3 := newFakeClientV3(context.Background()) | ||
|
|
||
| ch := make(chan struct{}) | ||
| cv3.WatchPrefix("prefix", ch) | ||
| } | ||
|
|
||
| func TestGetEntriesV3(t *testing.T) { | ||
| cv3 := newFakeClientV3(context.Background()) | ||
|
|
||
| res, err := cv3.GetEntries("prefix") | ||
| if res != nil || err == nil { | ||
| t.Errorf("expected client error") | ||
| } | ||
| } |
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
Oops, something went wrong.