This update provides:
1) workaround around the build failures. In
summary, it forces the download of some packages during the build
process.
2) update the set of packages that should go inside the vendor
directory
3) Update the dockerfile to use go 1.10
Change-Id: I2bfd090ce0f25b0c10aa214755ae2da7e5384d60
diff --git a/vendor/github.com/coreos/etcd/clientv3/README.md b/vendor/github.com/coreos/etcd/clientv3/README.md
new file mode 100644
index 0000000..376bfba
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/README.md
@@ -0,0 +1,85 @@
+# etcd/clientv3
+
+[](https://godoc.org/github.com/coreos/etcd/clientv3)
+
+`etcd/clientv3` is the official Go etcd client for v3.
+
+## Install
+
+```bash
+go get github.com/coreos/etcd/clientv3
+```
+
+## Get started
+
+Create client using `clientv3.New`:
+
+```go
+cli, err := clientv3.New(clientv3.Config{
+ Endpoints: []string{"localhost:2379", "localhost:22379", "localhost:32379"},
+ DialTimeout: 5 * time.Second,
+})
+if err != nil {
+ // handle error!
+}
+defer cli.Close()
+```
+
+etcd v3 uses [`gRPC`](http://www.grpc.io) for remote procedure calls. And `clientv3` uses
+[`grpc-go`](https://github.com/grpc/grpc-go) to connect to etcd. Make sure to close the client after using it.
+If the client is not closed, the connection will have leaky goroutines. To specify client request timeout,
+pass `context.WithTimeout` to APIs:
+
+```go
+ctx, cancel := context.WithTimeout(context.Background(), timeout)
+resp, err := cli.Put(ctx, "sample_key", "sample_value")
+cancel()
+if err != nil {
+ // handle error!
+}
+// use the response
+```
+
+etcd uses `cmd/vendor` directory to store external dependencies, which are
+to be compiled into etcd release binaries. `client` can be imported without
+vendoring. For full compatibility, it is recommended to vendor builds using
+etcd's vendored packages, using tools like godep, as in
+[vendor directories](https://golang.org/cmd/go/#hdr-Vendor_Directories).
+For more detail, please read [Go vendor design](https://golang.org/s/go15vendor).
+
+## Error Handling
+
+etcd client returns 2 types of errors:
+
+1. context error: canceled or deadline exceeded.
+2. gRPC error: see [api/v3rpc/rpctypes](https://godoc.org/github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes).
+
+Here is the example code to handle client errors:
+
+```go
+resp, err := cli.Put(ctx, "", "")
+if err != nil {
+ switch err {
+ case context.Canceled:
+ log.Fatalf("ctx is canceled by another routine: %v", err)
+ case context.DeadlineExceeded:
+ log.Fatalf("ctx is attached with a deadline is exceeded: %v", err)
+ case rpctypes.ErrEmptyKey:
+ log.Fatalf("client-side error: %v", err)
+ default:
+ log.Fatalf("bad cluster endpoints, which are not etcd servers: %v", err)
+ }
+}
+```
+
+## Metrics
+
+The etcd client optionally exposes RPC metrics through [go-grpc-prometheus](https://github.com/grpc-ecosystem/go-grpc-prometheus). See the [examples](https://github.com/coreos/etcd/blob/master/clientv3/example_metrics_test.go).
+
+## Namespacing
+
+The [namespace](https://godoc.org/github.com/coreos/etcd/clientv3/namespace) package provides `clientv3` interface wrappers to transparently isolate client requests to a user-defined prefix.
+
+## Examples
+
+More code examples can be found at [GoDoc](https://godoc.org/github.com/coreos/etcd/clientv3).
diff --git a/vendor/github.com/coreos/etcd/clientv3/auth.go b/vendor/github.com/coreos/etcd/clientv3/auth.go
new file mode 100644
index 0000000..7545bb6
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/auth.go
@@ -0,0 +1,233 @@
+// 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 clientv3
+
+import (
+ "context"
+ "fmt"
+ "strings"
+
+ "github.com/coreos/etcd/auth/authpb"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+ "google.golang.org/grpc"
+)
+
+type (
+ AuthEnableResponse pb.AuthEnableResponse
+ AuthDisableResponse pb.AuthDisableResponse
+ AuthenticateResponse pb.AuthenticateResponse
+ AuthUserAddResponse pb.AuthUserAddResponse
+ AuthUserDeleteResponse pb.AuthUserDeleteResponse
+ AuthUserChangePasswordResponse pb.AuthUserChangePasswordResponse
+ AuthUserGrantRoleResponse pb.AuthUserGrantRoleResponse
+ AuthUserGetResponse pb.AuthUserGetResponse
+ AuthUserRevokeRoleResponse pb.AuthUserRevokeRoleResponse
+ AuthRoleAddResponse pb.AuthRoleAddResponse
+ AuthRoleGrantPermissionResponse pb.AuthRoleGrantPermissionResponse
+ AuthRoleGetResponse pb.AuthRoleGetResponse
+ AuthRoleRevokePermissionResponse pb.AuthRoleRevokePermissionResponse
+ AuthRoleDeleteResponse pb.AuthRoleDeleteResponse
+ AuthUserListResponse pb.AuthUserListResponse
+ AuthRoleListResponse pb.AuthRoleListResponse
+
+ PermissionType authpb.Permission_Type
+ Permission authpb.Permission
+)
+
+const (
+ PermRead = authpb.READ
+ PermWrite = authpb.WRITE
+ PermReadWrite = authpb.READWRITE
+)
+
+type Auth interface {
+ // AuthEnable enables auth of an etcd cluster.
+ AuthEnable(ctx context.Context) (*AuthEnableResponse, error)
+
+ // AuthDisable disables auth of an etcd cluster.
+ AuthDisable(ctx context.Context) (*AuthDisableResponse, error)
+
+ // UserAdd adds a new user to an etcd cluster.
+ UserAdd(ctx context.Context, name string, password string) (*AuthUserAddResponse, error)
+
+ // UserDelete deletes a user from an etcd cluster.
+ UserDelete(ctx context.Context, name string) (*AuthUserDeleteResponse, error)
+
+ // UserChangePassword changes a password of a user.
+ UserChangePassword(ctx context.Context, name string, password string) (*AuthUserChangePasswordResponse, error)
+
+ // UserGrantRole grants a role to a user.
+ UserGrantRole(ctx context.Context, user string, role string) (*AuthUserGrantRoleResponse, error)
+
+ // UserGet gets a detailed information of a user.
+ UserGet(ctx context.Context, name string) (*AuthUserGetResponse, error)
+
+ // UserList gets a list of all users.
+ UserList(ctx context.Context) (*AuthUserListResponse, error)
+
+ // UserRevokeRole revokes a role of a user.
+ UserRevokeRole(ctx context.Context, name string, role string) (*AuthUserRevokeRoleResponse, error)
+
+ // RoleAdd adds a new role to an etcd cluster.
+ RoleAdd(ctx context.Context, name string) (*AuthRoleAddResponse, error)
+
+ // RoleGrantPermission grants a permission to a role.
+ RoleGrantPermission(ctx context.Context, name string, key, rangeEnd string, permType PermissionType) (*AuthRoleGrantPermissionResponse, error)
+
+ // RoleGet gets a detailed information of a role.
+ RoleGet(ctx context.Context, role string) (*AuthRoleGetResponse, error)
+
+ // RoleList gets a list of all roles.
+ RoleList(ctx context.Context) (*AuthRoleListResponse, error)
+
+ // RoleRevokePermission revokes a permission from a role.
+ RoleRevokePermission(ctx context.Context, role string, key, rangeEnd string) (*AuthRoleRevokePermissionResponse, error)
+
+ // RoleDelete deletes a role.
+ RoleDelete(ctx context.Context, role string) (*AuthRoleDeleteResponse, error)
+}
+
+type auth struct {
+ remote pb.AuthClient
+ callOpts []grpc.CallOption
+}
+
+func NewAuth(c *Client) Auth {
+ api := &auth{remote: RetryAuthClient(c)}
+ if c != nil {
+ api.callOpts = c.callOpts
+ }
+ return api
+}
+
+func (auth *auth) AuthEnable(ctx context.Context) (*AuthEnableResponse, error) {
+ resp, err := auth.remote.AuthEnable(ctx, &pb.AuthEnableRequest{}, auth.callOpts...)
+ return (*AuthEnableResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) AuthDisable(ctx context.Context) (*AuthDisableResponse, error) {
+ resp, err := auth.remote.AuthDisable(ctx, &pb.AuthDisableRequest{}, auth.callOpts...)
+ return (*AuthDisableResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) UserAdd(ctx context.Context, name string, password string) (*AuthUserAddResponse, error) {
+ resp, err := auth.remote.UserAdd(ctx, &pb.AuthUserAddRequest{Name: name, Password: password}, auth.callOpts...)
+ return (*AuthUserAddResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) UserDelete(ctx context.Context, name string) (*AuthUserDeleteResponse, error) {
+ resp, err := auth.remote.UserDelete(ctx, &pb.AuthUserDeleteRequest{Name: name}, auth.callOpts...)
+ return (*AuthUserDeleteResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) UserChangePassword(ctx context.Context, name string, password string) (*AuthUserChangePasswordResponse, error) {
+ resp, err := auth.remote.UserChangePassword(ctx, &pb.AuthUserChangePasswordRequest{Name: name, Password: password}, auth.callOpts...)
+ return (*AuthUserChangePasswordResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) UserGrantRole(ctx context.Context, user string, role string) (*AuthUserGrantRoleResponse, error) {
+ resp, err := auth.remote.UserGrantRole(ctx, &pb.AuthUserGrantRoleRequest{User: user, Role: role}, auth.callOpts...)
+ return (*AuthUserGrantRoleResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) UserGet(ctx context.Context, name string) (*AuthUserGetResponse, error) {
+ resp, err := auth.remote.UserGet(ctx, &pb.AuthUserGetRequest{Name: name}, auth.callOpts...)
+ return (*AuthUserGetResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) UserList(ctx context.Context) (*AuthUserListResponse, error) {
+ resp, err := auth.remote.UserList(ctx, &pb.AuthUserListRequest{}, auth.callOpts...)
+ return (*AuthUserListResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) UserRevokeRole(ctx context.Context, name string, role string) (*AuthUserRevokeRoleResponse, error) {
+ resp, err := auth.remote.UserRevokeRole(ctx, &pb.AuthUserRevokeRoleRequest{Name: name, Role: role}, auth.callOpts...)
+ return (*AuthUserRevokeRoleResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) RoleAdd(ctx context.Context, name string) (*AuthRoleAddResponse, error) {
+ resp, err := auth.remote.RoleAdd(ctx, &pb.AuthRoleAddRequest{Name: name}, auth.callOpts...)
+ return (*AuthRoleAddResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) RoleGrantPermission(ctx context.Context, name string, key, rangeEnd string, permType PermissionType) (*AuthRoleGrantPermissionResponse, error) {
+ perm := &authpb.Permission{
+ Key: []byte(key),
+ RangeEnd: []byte(rangeEnd),
+ PermType: authpb.Permission_Type(permType),
+ }
+ resp, err := auth.remote.RoleGrantPermission(ctx, &pb.AuthRoleGrantPermissionRequest{Name: name, Perm: perm}, auth.callOpts...)
+ return (*AuthRoleGrantPermissionResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) RoleGet(ctx context.Context, role string) (*AuthRoleGetResponse, error) {
+ resp, err := auth.remote.RoleGet(ctx, &pb.AuthRoleGetRequest{Role: role}, auth.callOpts...)
+ return (*AuthRoleGetResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) RoleList(ctx context.Context) (*AuthRoleListResponse, error) {
+ resp, err := auth.remote.RoleList(ctx, &pb.AuthRoleListRequest{}, auth.callOpts...)
+ return (*AuthRoleListResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) RoleRevokePermission(ctx context.Context, role string, key, rangeEnd string) (*AuthRoleRevokePermissionResponse, error) {
+ resp, err := auth.remote.RoleRevokePermission(ctx, &pb.AuthRoleRevokePermissionRequest{Role: role, Key: key, RangeEnd: rangeEnd}, auth.callOpts...)
+ return (*AuthRoleRevokePermissionResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *auth) RoleDelete(ctx context.Context, role string) (*AuthRoleDeleteResponse, error) {
+ resp, err := auth.remote.RoleDelete(ctx, &pb.AuthRoleDeleteRequest{Role: role}, auth.callOpts...)
+ return (*AuthRoleDeleteResponse)(resp), toErr(ctx, err)
+}
+
+func StrToPermissionType(s string) (PermissionType, error) {
+ val, ok := authpb.Permission_Type_value[strings.ToUpper(s)]
+ if ok {
+ return PermissionType(val), nil
+ }
+ return PermissionType(-1), fmt.Errorf("invalid permission type: %s", s)
+}
+
+type authenticator struct {
+ conn *grpc.ClientConn // conn in-use
+ remote pb.AuthClient
+ callOpts []grpc.CallOption
+}
+
+func (auth *authenticator) authenticate(ctx context.Context, name string, password string) (*AuthenticateResponse, error) {
+ resp, err := auth.remote.Authenticate(ctx, &pb.AuthenticateRequest{Name: name, Password: password}, auth.callOpts...)
+ return (*AuthenticateResponse)(resp), toErr(ctx, err)
+}
+
+func (auth *authenticator) close() {
+ auth.conn.Close()
+}
+
+func newAuthenticator(endpoint string, opts []grpc.DialOption, c *Client) (*authenticator, error) {
+ conn, err := grpc.Dial(endpoint, opts...)
+ if err != nil {
+ return nil, err
+ }
+
+ api := &authenticator{
+ conn: conn,
+ remote: pb.NewAuthClient(conn),
+ }
+ if c != nil {
+ api.callOpts = c.callOpts
+ }
+ return api, nil
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/client.go b/vendor/github.com/coreos/etcd/clientv3/client.go
new file mode 100644
index 0000000..7132807
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/client.go
@@ -0,0 +1,576 @@
+// 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 clientv3
+
+import (
+ "context"
+ "crypto/tls"
+ "errors"
+ "fmt"
+ "net"
+ "net/url"
+ "strconv"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/credentials"
+ "google.golang.org/grpc/keepalive"
+ "google.golang.org/grpc/metadata"
+ "google.golang.org/grpc/status"
+)
+
+var (
+ ErrNoAvailableEndpoints = errors.New("etcdclient: no available endpoints")
+ ErrOldCluster = errors.New("etcdclient: old cluster version")
+)
+
+// Client provides and manages an etcd v3 client session.
+type Client struct {
+ Cluster
+ KV
+ Lease
+ Watcher
+ Auth
+ Maintenance
+
+ conn *grpc.ClientConn
+ dialerrc chan error
+
+ cfg Config
+ creds *credentials.TransportCredentials
+ balancer *healthBalancer
+ mu *sync.Mutex
+
+ ctx context.Context
+ cancel context.CancelFunc
+
+ // Username is a user name for authentication.
+ Username string
+ // Password is a password for authentication.
+ Password string
+ // tokenCred is an instance of WithPerRPCCredentials()'s argument
+ tokenCred *authTokenCredential
+
+ callOpts []grpc.CallOption
+}
+
+// New creates a new etcdv3 client from a given configuration.
+func New(cfg Config) (*Client, error) {
+ if len(cfg.Endpoints) == 0 {
+ return nil, ErrNoAvailableEndpoints
+ }
+
+ return newClient(&cfg)
+}
+
+// NewCtxClient creates a client with a context but no underlying grpc
+// connection. This is useful for embedded cases that override the
+// service interface implementations and do not need connection management.
+func NewCtxClient(ctx context.Context) *Client {
+ cctx, cancel := context.WithCancel(ctx)
+ return &Client{ctx: cctx, cancel: cancel}
+}
+
+// NewFromURL creates a new etcdv3 client from a URL.
+func NewFromURL(url string) (*Client, error) {
+ return New(Config{Endpoints: []string{url}})
+}
+
+// Close shuts down the client's etcd connections.
+func (c *Client) Close() error {
+ c.cancel()
+ c.Watcher.Close()
+ c.Lease.Close()
+ if c.conn != nil {
+ return toErr(c.ctx, c.conn.Close())
+ }
+ return c.ctx.Err()
+}
+
+// Ctx is a context for "out of band" messages (e.g., for sending
+// "clean up" message when another context is canceled). It is
+// canceled on client Close().
+func (c *Client) Ctx() context.Context { return c.ctx }
+
+// Endpoints lists the registered endpoints for the client.
+func (c *Client) Endpoints() (eps []string) {
+ // copy the slice; protect original endpoints from being changed
+ eps = make([]string, len(c.cfg.Endpoints))
+ copy(eps, c.cfg.Endpoints)
+ return
+}
+
+// SetEndpoints updates client's endpoints.
+func (c *Client) SetEndpoints(eps ...string) {
+ c.mu.Lock()
+ c.cfg.Endpoints = eps
+ c.mu.Unlock()
+ c.balancer.updateAddrs(eps...)
+
+ // updating notifyCh can trigger new connections,
+ // need update addrs if all connections are down
+ // or addrs does not include pinAddr.
+ c.balancer.mu.RLock()
+ update := !hasAddr(c.balancer.addrs, c.balancer.pinAddr)
+ c.balancer.mu.RUnlock()
+ if update {
+ select {
+ case c.balancer.updateAddrsC <- notifyNext:
+ case <-c.balancer.stopc:
+ }
+ }
+}
+
+// Sync synchronizes client's endpoints with the known endpoints from the etcd membership.
+func (c *Client) Sync(ctx context.Context) error {
+ mresp, err := c.MemberList(ctx)
+ if err != nil {
+ return err
+ }
+ var eps []string
+ for _, m := range mresp.Members {
+ eps = append(eps, m.ClientURLs...)
+ }
+ c.SetEndpoints(eps...)
+ return nil
+}
+
+func (c *Client) autoSync() {
+ if c.cfg.AutoSyncInterval == time.Duration(0) {
+ return
+ }
+
+ for {
+ select {
+ case <-c.ctx.Done():
+ return
+ case <-time.After(c.cfg.AutoSyncInterval):
+ ctx, cancel := context.WithTimeout(c.ctx, 5*time.Second)
+ err := c.Sync(ctx)
+ cancel()
+ if err != nil && err != c.ctx.Err() {
+ logger.Println("Auto sync endpoints failed:", err)
+ }
+ }
+ }
+}
+
+type authTokenCredential struct {
+ token string
+ tokenMu *sync.RWMutex
+}
+
+func (cred authTokenCredential) RequireTransportSecurity() bool {
+ return false
+}
+
+func (cred authTokenCredential) GetRequestMetadata(ctx context.Context, s ...string) (map[string]string, error) {
+ cred.tokenMu.RLock()
+ defer cred.tokenMu.RUnlock()
+ return map[string]string{
+ "token": cred.token,
+ }, nil
+}
+
+func parseEndpoint(endpoint string) (proto string, host string, scheme string) {
+ proto = "tcp"
+ host = endpoint
+ url, uerr := url.Parse(endpoint)
+ if uerr != nil || !strings.Contains(endpoint, "://") {
+ return proto, host, scheme
+ }
+ scheme = url.Scheme
+
+ // strip scheme:// prefix since grpc dials by host
+ host = url.Host
+ switch url.Scheme {
+ case "http", "https":
+ case "unix", "unixs":
+ proto = "unix"
+ host = url.Host + url.Path
+ default:
+ proto, host = "", ""
+ }
+ return proto, host, scheme
+}
+
+func (c *Client) processCreds(scheme string) (creds *credentials.TransportCredentials) {
+ creds = c.creds
+ switch scheme {
+ case "unix":
+ case "http":
+ creds = nil
+ case "https", "unixs":
+ if creds != nil {
+ break
+ }
+ tlsconfig := &tls.Config{}
+ emptyCreds := credentials.NewTLS(tlsconfig)
+ creds = &emptyCreds
+ default:
+ creds = nil
+ }
+ return creds
+}
+
+// dialSetupOpts gives the dial opts prior to any authentication
+func (c *Client) dialSetupOpts(endpoint string, dopts ...grpc.DialOption) (opts []grpc.DialOption) {
+ if c.cfg.DialTimeout > 0 {
+ opts = []grpc.DialOption{grpc.WithTimeout(c.cfg.DialTimeout)}
+ }
+ if c.cfg.DialKeepAliveTime > 0 {
+ params := keepalive.ClientParameters{
+ Time: c.cfg.DialKeepAliveTime,
+ Timeout: c.cfg.DialKeepAliveTimeout,
+ }
+ opts = append(opts, grpc.WithKeepaliveParams(params))
+ }
+ opts = append(opts, dopts...)
+
+ f := func(host string, t time.Duration) (net.Conn, error) {
+ proto, host, _ := parseEndpoint(c.balancer.endpoint(host))
+ if host == "" && endpoint != "" {
+ // dialing an endpoint not in the balancer; use
+ // endpoint passed into dial
+ proto, host, _ = parseEndpoint(endpoint)
+ }
+ if proto == "" {
+ return nil, fmt.Errorf("unknown scheme for %q", host)
+ }
+ select {
+ case <-c.ctx.Done():
+ return nil, c.ctx.Err()
+ default:
+ }
+ dialer := &net.Dialer{Timeout: t}
+ conn, err := dialer.DialContext(c.ctx, proto, host)
+ if err != nil {
+ select {
+ case c.dialerrc <- err:
+ default:
+ }
+ }
+ return conn, err
+ }
+ opts = append(opts, grpc.WithDialer(f))
+
+ creds := c.creds
+ if _, _, scheme := parseEndpoint(endpoint); len(scheme) != 0 {
+ creds = c.processCreds(scheme)
+ }
+ if creds != nil {
+ opts = append(opts, grpc.WithTransportCredentials(*creds))
+ } else {
+ opts = append(opts, grpc.WithInsecure())
+ }
+
+ return opts
+}
+
+// Dial connects to a single endpoint using the client's config.
+func (c *Client) Dial(endpoint string) (*grpc.ClientConn, error) {
+ return c.dial(endpoint)
+}
+
+func (c *Client) getToken(ctx context.Context) error {
+ var err error // return last error in a case of fail
+ var auth *authenticator
+
+ for i := 0; i < len(c.cfg.Endpoints); i++ {
+ endpoint := c.cfg.Endpoints[i]
+ host := getHost(endpoint)
+ // use dial options without dopts to avoid reusing the client balancer
+ auth, err = newAuthenticator(host, c.dialSetupOpts(endpoint), c)
+ if err != nil {
+ continue
+ }
+ defer auth.close()
+
+ var resp *AuthenticateResponse
+ resp, err = auth.authenticate(ctx, c.Username, c.Password)
+ if err != nil {
+ continue
+ }
+
+ c.tokenCred.tokenMu.Lock()
+ c.tokenCred.token = resp.Token
+ c.tokenCred.tokenMu.Unlock()
+
+ return nil
+ }
+
+ return err
+}
+
+func (c *Client) dial(endpoint string, dopts ...grpc.DialOption) (*grpc.ClientConn, error) {
+ opts := c.dialSetupOpts(endpoint, dopts...)
+ host := getHost(endpoint)
+ if c.Username != "" && c.Password != "" {
+ c.tokenCred = &authTokenCredential{
+ tokenMu: &sync.RWMutex{},
+ }
+
+ ctx := c.ctx
+ if c.cfg.DialTimeout > 0 {
+ cctx, cancel := context.WithTimeout(ctx, c.cfg.DialTimeout)
+ defer cancel()
+ ctx = cctx
+ }
+
+ err := c.getToken(ctx)
+ if err != nil {
+ if toErr(ctx, err) != rpctypes.ErrAuthNotEnabled {
+ if err == ctx.Err() && ctx.Err() != c.ctx.Err() {
+ err = context.DeadlineExceeded
+ }
+ return nil, err
+ }
+ } else {
+ opts = append(opts, grpc.WithPerRPCCredentials(c.tokenCred))
+ }
+ }
+
+ opts = append(opts, c.cfg.DialOptions...)
+
+ conn, err := grpc.DialContext(c.ctx, host, opts...)
+ if err != nil {
+ return nil, err
+ }
+ return conn, nil
+}
+
+// WithRequireLeader requires client requests to only succeed
+// when the cluster has a leader.
+func WithRequireLeader(ctx context.Context) context.Context {
+ md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
+ return metadata.NewOutgoingContext(ctx, md)
+}
+
+func newClient(cfg *Config) (*Client, error) {
+ if cfg == nil {
+ cfg = &Config{}
+ }
+ var creds *credentials.TransportCredentials
+ if cfg.TLS != nil {
+ c := credentials.NewTLS(cfg.TLS)
+ creds = &c
+ }
+
+ // use a temporary skeleton client to bootstrap first connection
+ baseCtx := context.TODO()
+ if cfg.Context != nil {
+ baseCtx = cfg.Context
+ }
+
+ ctx, cancel := context.WithCancel(baseCtx)
+ client := &Client{
+ conn: nil,
+ dialerrc: make(chan error, 1),
+ cfg: *cfg,
+ creds: creds,
+ ctx: ctx,
+ cancel: cancel,
+ mu: new(sync.Mutex),
+ callOpts: defaultCallOpts,
+ }
+ if cfg.Username != "" && cfg.Password != "" {
+ client.Username = cfg.Username
+ client.Password = cfg.Password
+ }
+ if cfg.MaxCallSendMsgSize > 0 || cfg.MaxCallRecvMsgSize > 0 {
+ if cfg.MaxCallRecvMsgSize > 0 && cfg.MaxCallSendMsgSize > cfg.MaxCallRecvMsgSize {
+ return nil, fmt.Errorf("gRPC message recv limit (%d bytes) must be greater than send limit (%d bytes)", cfg.MaxCallRecvMsgSize, cfg.MaxCallSendMsgSize)
+ }
+ callOpts := []grpc.CallOption{
+ defaultFailFast,
+ defaultMaxCallSendMsgSize,
+ defaultMaxCallRecvMsgSize,
+ }
+ if cfg.MaxCallSendMsgSize > 0 {
+ callOpts[1] = grpc.MaxCallSendMsgSize(cfg.MaxCallSendMsgSize)
+ }
+ if cfg.MaxCallRecvMsgSize > 0 {
+ callOpts[2] = grpc.MaxCallRecvMsgSize(cfg.MaxCallRecvMsgSize)
+ }
+ client.callOpts = callOpts
+ }
+
+ client.balancer = newHealthBalancer(cfg.Endpoints, cfg.DialTimeout, func(ep string) (bool, error) {
+ return grpcHealthCheck(client, ep)
+ })
+
+ // use Endpoints[0] so that for https:// without any tls config given, then
+ // grpc will assume the certificate server name is the endpoint host.
+ conn, err := client.dial(cfg.Endpoints[0], grpc.WithBalancer(client.balancer))
+ if err != nil {
+ client.cancel()
+ client.balancer.Close()
+ return nil, err
+ }
+ client.conn = conn
+
+ // wait for a connection
+ if cfg.DialTimeout > 0 {
+ hasConn := false
+ waitc := time.After(cfg.DialTimeout)
+ select {
+ case <-client.balancer.ready():
+ hasConn = true
+ case <-ctx.Done():
+ case <-waitc:
+ }
+ if !hasConn {
+ err := context.DeadlineExceeded
+ select {
+ case err = <-client.dialerrc:
+ default:
+ }
+ client.cancel()
+ client.balancer.Close()
+ conn.Close()
+ return nil, err
+ }
+ }
+
+ client.Cluster = NewCluster(client)
+ client.KV = NewKV(client)
+ client.Lease = NewLease(client)
+ client.Watcher = NewWatcher(client)
+ client.Auth = NewAuth(client)
+ client.Maintenance = NewMaintenance(client)
+
+ if cfg.RejectOldCluster {
+ if err := client.checkVersion(); err != nil {
+ client.Close()
+ return nil, err
+ }
+ }
+
+ go client.autoSync()
+ return client, nil
+}
+
+func (c *Client) checkVersion() (err error) {
+ var wg sync.WaitGroup
+ errc := make(chan error, len(c.cfg.Endpoints))
+ ctx, cancel := context.WithCancel(c.ctx)
+ if c.cfg.DialTimeout > 0 {
+ ctx, cancel = context.WithTimeout(ctx, c.cfg.DialTimeout)
+ }
+ wg.Add(len(c.cfg.Endpoints))
+ for _, ep := range c.cfg.Endpoints {
+ // if cluster is current, any endpoint gives a recent version
+ go func(e string) {
+ defer wg.Done()
+ resp, rerr := c.Status(ctx, e)
+ if rerr != nil {
+ errc <- rerr
+ return
+ }
+ vs := strings.Split(resp.Version, ".")
+ maj, min := 0, 0
+ if len(vs) >= 2 {
+ maj, _ = strconv.Atoi(vs[0])
+ min, rerr = strconv.Atoi(vs[1])
+ }
+ if maj < 3 || (maj == 3 && min < 2) {
+ rerr = ErrOldCluster
+ }
+ errc <- rerr
+ }(ep)
+ }
+ // wait for success
+ for i := 0; i < len(c.cfg.Endpoints); i++ {
+ if err = <-errc; err == nil {
+ break
+ }
+ }
+ cancel()
+ wg.Wait()
+ return err
+}
+
+// ActiveConnection returns the current in-use connection
+func (c *Client) ActiveConnection() *grpc.ClientConn { return c.conn }
+
+// isHaltErr returns true if the given error and context indicate no forward
+// progress can be made, even after reconnecting.
+func isHaltErr(ctx context.Context, err error) bool {
+ if ctx != nil && ctx.Err() != nil {
+ return true
+ }
+ if err == nil {
+ return false
+ }
+ ev, _ := status.FromError(err)
+ // Unavailable codes mean the system will be right back.
+ // (e.g., can't connect, lost leader)
+ // Treat Internal codes as if something failed, leaving the
+ // system in an inconsistent state, but retrying could make progress.
+ // (e.g., failed in middle of send, corrupted frame)
+ // TODO: are permanent Internal errors possible from grpc?
+ return ev.Code() != codes.Unavailable && ev.Code() != codes.Internal
+}
+
+// isUnavailableErr returns true if the given error is an unavailable error
+func isUnavailableErr(ctx context.Context, err error) bool {
+ if ctx != nil && ctx.Err() != nil {
+ return false
+ }
+ if err == nil {
+ return false
+ }
+ ev, _ := status.FromError(err)
+ // Unavailable codes mean the system will be right back.
+ // (e.g., can't connect, lost leader)
+ return ev.Code() == codes.Unavailable
+}
+
+func toErr(ctx context.Context, err error) error {
+ if err == nil {
+ return nil
+ }
+ err = rpctypes.Error(err)
+ if _, ok := err.(rpctypes.EtcdError); ok {
+ return err
+ }
+ ev, _ := status.FromError(err)
+ code := ev.Code()
+ switch code {
+ case codes.DeadlineExceeded:
+ fallthrough
+ case codes.Canceled:
+ if ctx.Err() != nil {
+ err = ctx.Err()
+ }
+ case codes.Unavailable:
+ case codes.FailedPrecondition:
+ err = grpc.ErrClientConnClosing
+ }
+ return err
+}
+
+func canceledByCaller(stopCtx context.Context, err error) bool {
+ if stopCtx.Err() == nil || err == nil {
+ return false
+ }
+
+ return err == context.Canceled || err == context.DeadlineExceeded
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/cluster.go b/vendor/github.com/coreos/etcd/clientv3/cluster.go
new file mode 100644
index 0000000..785672b
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/cluster.go
@@ -0,0 +1,114 @@
+// 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 clientv3
+
+import (
+ "context"
+
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/pkg/types"
+
+ "google.golang.org/grpc"
+)
+
+type (
+ Member pb.Member
+ MemberListResponse pb.MemberListResponse
+ MemberAddResponse pb.MemberAddResponse
+ MemberRemoveResponse pb.MemberRemoveResponse
+ MemberUpdateResponse pb.MemberUpdateResponse
+)
+
+type Cluster interface {
+ // MemberList lists the current cluster membership.
+ MemberList(ctx context.Context) (*MemberListResponse, error)
+
+ // MemberAdd adds a new member into the cluster.
+ MemberAdd(ctx context.Context, peerAddrs []string) (*MemberAddResponse, error)
+
+ // MemberRemove removes an existing member from the cluster.
+ MemberRemove(ctx context.Context, id uint64) (*MemberRemoveResponse, error)
+
+ // MemberUpdate updates the peer addresses of the member.
+ MemberUpdate(ctx context.Context, id uint64, peerAddrs []string) (*MemberUpdateResponse, error)
+}
+
+type cluster struct {
+ remote pb.ClusterClient
+ callOpts []grpc.CallOption
+}
+
+func NewCluster(c *Client) Cluster {
+ api := &cluster{remote: RetryClusterClient(c)}
+ if c != nil {
+ api.callOpts = c.callOpts
+ }
+ return api
+}
+
+func NewClusterFromClusterClient(remote pb.ClusterClient, c *Client) Cluster {
+ api := &cluster{remote: remote}
+ if c != nil {
+ api.callOpts = c.callOpts
+ }
+ return api
+}
+
+func (c *cluster) MemberAdd(ctx context.Context, peerAddrs []string) (*MemberAddResponse, error) {
+ // fail-fast before panic in rafthttp
+ if _, err := types.NewURLs(peerAddrs); err != nil {
+ return nil, err
+ }
+
+ r := &pb.MemberAddRequest{PeerURLs: peerAddrs}
+ resp, err := c.remote.MemberAdd(ctx, r, c.callOpts...)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ return (*MemberAddResponse)(resp), nil
+}
+
+func (c *cluster) MemberRemove(ctx context.Context, id uint64) (*MemberRemoveResponse, error) {
+ r := &pb.MemberRemoveRequest{ID: id}
+ resp, err := c.remote.MemberRemove(ctx, r, c.callOpts...)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ return (*MemberRemoveResponse)(resp), nil
+}
+
+func (c *cluster) MemberUpdate(ctx context.Context, id uint64, peerAddrs []string) (*MemberUpdateResponse, error) {
+ // fail-fast before panic in rafthttp
+ if _, err := types.NewURLs(peerAddrs); err != nil {
+ return nil, err
+ }
+
+ // it is safe to retry on update.
+ r := &pb.MemberUpdateRequest{ID: id, PeerURLs: peerAddrs}
+ resp, err := c.remote.MemberUpdate(ctx, r, c.callOpts...)
+ if err == nil {
+ return (*MemberUpdateResponse)(resp), nil
+ }
+ return nil, toErr(ctx, err)
+}
+
+func (c *cluster) MemberList(ctx context.Context) (*MemberListResponse, error) {
+ // it is safe to retry on list.
+ resp, err := c.remote.MemberList(ctx, &pb.MemberListRequest{}, c.callOpts...)
+ if err == nil {
+ return (*MemberListResponse)(resp), nil
+ }
+ return nil, toErr(ctx, err)
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/compact_op.go b/vendor/github.com/coreos/etcd/clientv3/compact_op.go
new file mode 100644
index 0000000..41e80c1
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/compact_op.go
@@ -0,0 +1,51 @@
+// 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 clientv3
+
+import (
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+// CompactOp represents a compact operation.
+type CompactOp struct {
+ revision int64
+ physical bool
+}
+
+// CompactOption configures compact operation.
+type CompactOption func(*CompactOp)
+
+func (op *CompactOp) applyCompactOpts(opts []CompactOption) {
+ for _, opt := range opts {
+ opt(op)
+ }
+}
+
+// OpCompact wraps slice CompactOption to create a CompactOp.
+func OpCompact(rev int64, opts ...CompactOption) CompactOp {
+ ret := CompactOp{revision: rev}
+ ret.applyCompactOpts(opts)
+ return ret
+}
+
+func (op CompactOp) toRequest() *pb.CompactionRequest {
+ return &pb.CompactionRequest{Revision: op.revision, Physical: op.physical}
+}
+
+// WithCompactPhysical makes Compact wait until all compacted entries are
+// removed from the etcd server's storage.
+func WithCompactPhysical() CompactOption {
+ return func(op *CompactOp) { op.physical = true }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/compare.go b/vendor/github.com/coreos/etcd/clientv3/compare.go
new file mode 100644
index 0000000..b5f0a25
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/compare.go
@@ -0,0 +1,140 @@
+// 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 clientv3
+
+import (
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+type CompareTarget int
+type CompareResult int
+
+const (
+ CompareVersion CompareTarget = iota
+ CompareCreated
+ CompareModified
+ CompareValue
+)
+
+type Cmp pb.Compare
+
+func Compare(cmp Cmp, result string, v interface{}) Cmp {
+ var r pb.Compare_CompareResult
+
+ switch result {
+ case "=":
+ r = pb.Compare_EQUAL
+ case "!=":
+ r = pb.Compare_NOT_EQUAL
+ case ">":
+ r = pb.Compare_GREATER
+ case "<":
+ r = pb.Compare_LESS
+ default:
+ panic("Unknown result op")
+ }
+
+ cmp.Result = r
+ switch cmp.Target {
+ case pb.Compare_VALUE:
+ val, ok := v.(string)
+ if !ok {
+ panic("bad compare value")
+ }
+ cmp.TargetUnion = &pb.Compare_Value{Value: []byte(val)}
+ case pb.Compare_VERSION:
+ cmp.TargetUnion = &pb.Compare_Version{Version: mustInt64(v)}
+ case pb.Compare_CREATE:
+ cmp.TargetUnion = &pb.Compare_CreateRevision{CreateRevision: mustInt64(v)}
+ case pb.Compare_MOD:
+ cmp.TargetUnion = &pb.Compare_ModRevision{ModRevision: mustInt64(v)}
+ case pb.Compare_LEASE:
+ cmp.TargetUnion = &pb.Compare_Lease{Lease: mustInt64orLeaseID(v)}
+ default:
+ panic("Unknown compare type")
+ }
+ return cmp
+}
+
+func Value(key string) Cmp {
+ return Cmp{Key: []byte(key), Target: pb.Compare_VALUE}
+}
+
+func Version(key string) Cmp {
+ return Cmp{Key: []byte(key), Target: pb.Compare_VERSION}
+}
+
+func CreateRevision(key string) Cmp {
+ return Cmp{Key: []byte(key), Target: pb.Compare_CREATE}
+}
+
+func ModRevision(key string) Cmp {
+ return Cmp{Key: []byte(key), Target: pb.Compare_MOD}
+}
+
+// LeaseValue compares a key's LeaseID to a value of your choosing. The empty
+// LeaseID is 0, otherwise known as `NoLease`.
+func LeaseValue(key string) Cmp {
+ return Cmp{Key: []byte(key), Target: pb.Compare_LEASE}
+}
+
+// KeyBytes returns the byte slice holding with the comparison key.
+func (cmp *Cmp) KeyBytes() []byte { return cmp.Key }
+
+// WithKeyBytes sets the byte slice for the comparison key.
+func (cmp *Cmp) WithKeyBytes(key []byte) { cmp.Key = key }
+
+// ValueBytes returns the byte slice holding the comparison value, if any.
+func (cmp *Cmp) ValueBytes() []byte {
+ if tu, ok := cmp.TargetUnion.(*pb.Compare_Value); ok {
+ return tu.Value
+ }
+ return nil
+}
+
+// WithValueBytes sets the byte slice for the comparison's value.
+func (cmp *Cmp) WithValueBytes(v []byte) { cmp.TargetUnion.(*pb.Compare_Value).Value = v }
+
+// WithRange sets the comparison to scan the range [key, end).
+func (cmp Cmp) WithRange(end string) Cmp {
+ cmp.RangeEnd = []byte(end)
+ return cmp
+}
+
+// WithPrefix sets the comparison to scan all keys prefixed by the key.
+func (cmp Cmp) WithPrefix() Cmp {
+ cmp.RangeEnd = getPrefix(cmp.Key)
+ return cmp
+}
+
+// mustInt64 panics if val isn't an int or int64. It returns an int64 otherwise.
+func mustInt64(val interface{}) int64 {
+ if v, ok := val.(int64); ok {
+ return v
+ }
+ if v, ok := val.(int); ok {
+ return int64(v)
+ }
+ panic("bad value")
+}
+
+// mustInt64orLeaseID panics if val isn't a LeaseID, int or int64. It returns an
+// int64 otherwise.
+func mustInt64orLeaseID(val interface{}) int64 {
+ if v, ok := val.(LeaseID); ok {
+ return int64(v)
+ }
+ return mustInt64(val)
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/concurrency/doc.go b/vendor/github.com/coreos/etcd/clientv3/concurrency/doc.go
new file mode 100644
index 0000000..dcdbf51
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/concurrency/doc.go
@@ -0,0 +1,17 @@
+// 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 concurrency implements concurrency operations on top of
+// etcd such as distributed locks, barriers, and elections.
+package concurrency
diff --git a/vendor/github.com/coreos/etcd/clientv3/concurrency/election.go b/vendor/github.com/coreos/etcd/clientv3/concurrency/election.go
new file mode 100644
index 0000000..e18a0ed
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/concurrency/election.go
@@ -0,0 +1,245 @@
+// 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 concurrency
+
+import (
+ "context"
+ "errors"
+ "fmt"
+
+ v3 "github.com/coreos/etcd/clientv3"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/mvcc/mvccpb"
+)
+
+var (
+ ErrElectionNotLeader = errors.New("election: not leader")
+ ErrElectionNoLeader = errors.New("election: no leader")
+)
+
+type Election struct {
+ session *Session
+
+ keyPrefix string
+
+ leaderKey string
+ leaderRev int64
+ leaderSession *Session
+ hdr *pb.ResponseHeader
+}
+
+// NewElection returns a new election on a given key prefix.
+func NewElection(s *Session, pfx string) *Election {
+ return &Election{session: s, keyPrefix: pfx + "/"}
+}
+
+// ResumeElection initializes an election with a known leader.
+func ResumeElection(s *Session, pfx string, leaderKey string, leaderRev int64) *Election {
+ return &Election{
+ session: s,
+ leaderKey: leaderKey,
+ leaderRev: leaderRev,
+ leaderSession: s,
+ }
+}
+
+// Campaign puts a value as eligible for the election. It blocks until
+// it is elected, an error occurs, or the context is cancelled.
+func (e *Election) Campaign(ctx context.Context, val string) error {
+ s := e.session
+ client := e.session.Client()
+
+ k := fmt.Sprintf("%s%x", e.keyPrefix, s.Lease())
+ txn := client.Txn(ctx).If(v3.Compare(v3.CreateRevision(k), "=", 0))
+ txn = txn.Then(v3.OpPut(k, val, v3.WithLease(s.Lease())))
+ txn = txn.Else(v3.OpGet(k))
+ resp, err := txn.Commit()
+ if err != nil {
+ return err
+ }
+ e.leaderKey, e.leaderRev, e.leaderSession = k, resp.Header.Revision, s
+ if !resp.Succeeded {
+ kv := resp.Responses[0].GetResponseRange().Kvs[0]
+ e.leaderRev = kv.CreateRevision
+ if string(kv.Value) != val {
+ if err = e.Proclaim(ctx, val); err != nil {
+ e.Resign(ctx)
+ return err
+ }
+ }
+ }
+
+ _, err = waitDeletes(ctx, client, e.keyPrefix, e.leaderRev-1)
+ if err != nil {
+ // clean up in case of context cancel
+ select {
+ case <-ctx.Done():
+ e.Resign(client.Ctx())
+ default:
+ e.leaderSession = nil
+ }
+ return err
+ }
+ e.hdr = resp.Header
+
+ return nil
+}
+
+// Proclaim lets the leader announce a new value without another election.
+func (e *Election) Proclaim(ctx context.Context, val string) error {
+ if e.leaderSession == nil {
+ return ErrElectionNotLeader
+ }
+ client := e.session.Client()
+ cmp := v3.Compare(v3.CreateRevision(e.leaderKey), "=", e.leaderRev)
+ txn := client.Txn(ctx).If(cmp)
+ txn = txn.Then(v3.OpPut(e.leaderKey, val, v3.WithLease(e.leaderSession.Lease())))
+ tresp, terr := txn.Commit()
+ if terr != nil {
+ return terr
+ }
+ if !tresp.Succeeded {
+ e.leaderKey = ""
+ return ErrElectionNotLeader
+ }
+
+ e.hdr = tresp.Header
+ return nil
+}
+
+// Resign lets a leader start a new election.
+func (e *Election) Resign(ctx context.Context) (err error) {
+ if e.leaderSession == nil {
+ return nil
+ }
+ client := e.session.Client()
+ cmp := v3.Compare(v3.CreateRevision(e.leaderKey), "=", e.leaderRev)
+ resp, err := client.Txn(ctx).If(cmp).Then(v3.OpDelete(e.leaderKey)).Commit()
+ if err == nil {
+ e.hdr = resp.Header
+ }
+ e.leaderKey = ""
+ e.leaderSession = nil
+ return err
+}
+
+// Leader returns the leader value for the current election.
+func (e *Election) Leader(ctx context.Context) (*v3.GetResponse, error) {
+ client := e.session.Client()
+ resp, err := client.Get(ctx, e.keyPrefix, v3.WithFirstCreate()...)
+ if err != nil {
+ return nil, err
+ } else if len(resp.Kvs) == 0 {
+ // no leader currently elected
+ return nil, ErrElectionNoLeader
+ }
+ return resp, nil
+}
+
+// Observe returns a channel that reliably observes ordered leader proposals
+// as GetResponse values on every current elected leader key. It will not
+// necessarily fetch all historical leader updates, but will always post the
+// most recent leader value.
+//
+// The channel closes when the context is canceled or the underlying watcher
+// is otherwise disrupted.
+func (e *Election) Observe(ctx context.Context) <-chan v3.GetResponse {
+ retc := make(chan v3.GetResponse)
+ go e.observe(ctx, retc)
+ return retc
+}
+
+func (e *Election) observe(ctx context.Context, ch chan<- v3.GetResponse) {
+ client := e.session.Client()
+
+ defer close(ch)
+ for {
+ resp, err := client.Get(ctx, e.keyPrefix, v3.WithFirstCreate()...)
+ if err != nil {
+ return
+ }
+
+ var kv *mvccpb.KeyValue
+ var hdr *pb.ResponseHeader
+
+ if len(resp.Kvs) == 0 {
+ cctx, cancel := context.WithCancel(ctx)
+ // wait for first key put on prefix
+ opts := []v3.OpOption{v3.WithRev(resp.Header.Revision), v3.WithPrefix()}
+ wch := client.Watch(cctx, e.keyPrefix, opts...)
+ for kv == nil {
+ wr, ok := <-wch
+ if !ok || wr.Err() != nil {
+ cancel()
+ return
+ }
+ // only accept puts; a delete will make observe() spin
+ for _, ev := range wr.Events {
+ if ev.Type == mvccpb.PUT {
+ hdr, kv = &wr.Header, ev.Kv
+ // may have multiple revs; hdr.rev = the last rev
+ // set to kv's rev in case batch has multiple Puts
+ hdr.Revision = kv.ModRevision
+ break
+ }
+ }
+ }
+ cancel()
+ } else {
+ hdr, kv = resp.Header, resp.Kvs[0]
+ }
+
+ select {
+ case ch <- v3.GetResponse{Header: hdr, Kvs: []*mvccpb.KeyValue{kv}}:
+ case <-ctx.Done():
+ return
+ }
+
+ cctx, cancel := context.WithCancel(ctx)
+ wch := client.Watch(cctx, string(kv.Key), v3.WithRev(hdr.Revision+1))
+ keyDeleted := false
+ for !keyDeleted {
+ wr, ok := <-wch
+ if !ok {
+ cancel()
+ return
+ }
+ for _, ev := range wr.Events {
+ if ev.Type == mvccpb.DELETE {
+ keyDeleted = true
+ break
+ }
+ resp.Header = &wr.Header
+ resp.Kvs = []*mvccpb.KeyValue{ev.Kv}
+ select {
+ case ch <- *resp:
+ case <-cctx.Done():
+ cancel()
+ return
+ }
+ }
+ }
+ cancel()
+ }
+}
+
+// Key returns the leader key if elected, empty string otherwise.
+func (e *Election) Key() string { return e.leaderKey }
+
+// Rev returns the leader key's creation revision, if elected.
+func (e *Election) Rev() int64 { return e.leaderRev }
+
+// Header is the response header from the last successful election proposal.
+func (e *Election) Header() *pb.ResponseHeader { return e.hdr }
diff --git a/vendor/github.com/coreos/etcd/clientv3/concurrency/key.go b/vendor/github.com/coreos/etcd/clientv3/concurrency/key.go
new file mode 100644
index 0000000..4b6e399
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/concurrency/key.go
@@ -0,0 +1,65 @@
+// 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 concurrency
+
+import (
+ "context"
+ "fmt"
+
+ v3 "github.com/coreos/etcd/clientv3"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/mvcc/mvccpb"
+)
+
+func waitDelete(ctx context.Context, client *v3.Client, key string, rev int64) error {
+ cctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ var wr v3.WatchResponse
+ wch := client.Watch(cctx, key, v3.WithRev(rev))
+ for wr = range wch {
+ for _, ev := range wr.Events {
+ if ev.Type == mvccpb.DELETE {
+ return nil
+ }
+ }
+ }
+ if err := wr.Err(); err != nil {
+ return err
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ return fmt.Errorf("lost watcher waiting for delete")
+}
+
+// waitDeletes efficiently waits until all keys matching the prefix and no greater
+// than the create revision.
+func waitDeletes(ctx context.Context, client *v3.Client, pfx string, maxCreateRev int64) (*pb.ResponseHeader, error) {
+ getOpts := append(v3.WithLastCreate(), v3.WithMaxCreateRev(maxCreateRev))
+ for {
+ resp, err := client.Get(ctx, pfx, getOpts...)
+ if err != nil {
+ return nil, err
+ }
+ if len(resp.Kvs) == 0 {
+ return resp.Header, nil
+ }
+ lastKey := string(resp.Kvs[0].Key)
+ if err = waitDelete(ctx, client, lastKey, resp.Header.Revision); err != nil {
+ return nil, err
+ }
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/concurrency/mutex.go b/vendor/github.com/coreos/etcd/clientv3/concurrency/mutex.go
new file mode 100644
index 0000000..77b3582
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/concurrency/mutex.go
@@ -0,0 +1,117 @@
+// 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 concurrency
+
+import (
+ "context"
+ "fmt"
+ "sync"
+
+ v3 "github.com/coreos/etcd/clientv3"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+// Mutex implements the sync Locker interface with etcd
+type Mutex struct {
+ s *Session
+
+ pfx string
+ myKey string
+ myRev int64
+ hdr *pb.ResponseHeader
+}
+
+func NewMutex(s *Session, pfx string) *Mutex {
+ return &Mutex{s, pfx + "/", "", -1, nil}
+}
+
+// Lock locks the mutex with a cancelable context. If the context is canceled
+// while trying to acquire the lock, the mutex tries to clean its stale lock entry.
+func (m *Mutex) Lock(ctx context.Context) error {
+ s := m.s
+ client := m.s.Client()
+
+ m.myKey = fmt.Sprintf("%s%x", m.pfx, s.Lease())
+ cmp := v3.Compare(v3.CreateRevision(m.myKey), "=", 0)
+ // put self in lock waiters via myKey; oldest waiter holds lock
+ put := v3.OpPut(m.myKey, "", v3.WithLease(s.Lease()))
+ // reuse key in case this session already holds the lock
+ get := v3.OpGet(m.myKey)
+ // fetch current holder to complete uncontended path with only one RPC
+ getOwner := v3.OpGet(m.pfx, v3.WithFirstCreate()...)
+ resp, err := client.Txn(ctx).If(cmp).Then(put, getOwner).Else(get, getOwner).Commit()
+ if err != nil {
+ return err
+ }
+ m.myRev = resp.Header.Revision
+ if !resp.Succeeded {
+ m.myRev = resp.Responses[0].GetResponseRange().Kvs[0].CreateRevision
+ }
+ // if no key on prefix / the minimum rev is key, already hold the lock
+ ownerKey := resp.Responses[1].GetResponseRange().Kvs
+ if len(ownerKey) == 0 || ownerKey[0].CreateRevision == m.myRev {
+ m.hdr = resp.Header
+ return nil
+ }
+
+ // wait for deletion revisions prior to myKey
+ hdr, werr := waitDeletes(ctx, client, m.pfx, m.myRev-1)
+ // release lock key if wait failed
+ if werr != nil {
+ m.Unlock(client.Ctx())
+ } else {
+ m.hdr = hdr
+ }
+ return werr
+}
+
+func (m *Mutex) Unlock(ctx context.Context) error {
+ client := m.s.Client()
+ if _, err := client.Delete(ctx, m.myKey); err != nil {
+ return err
+ }
+ m.myKey = "\x00"
+ m.myRev = -1
+ return nil
+}
+
+func (m *Mutex) IsOwner() v3.Cmp {
+ return v3.Compare(v3.CreateRevision(m.myKey), "=", m.myRev)
+}
+
+func (m *Mutex) Key() string { return m.myKey }
+
+// Header is the response header received from etcd on acquiring the lock.
+func (m *Mutex) Header() *pb.ResponseHeader { return m.hdr }
+
+type lockerMutex struct{ *Mutex }
+
+func (lm *lockerMutex) Lock() {
+ client := lm.s.Client()
+ if err := lm.Mutex.Lock(client.Ctx()); err != nil {
+ panic(err)
+ }
+}
+func (lm *lockerMutex) Unlock() {
+ client := lm.s.Client()
+ if err := lm.Mutex.Unlock(client.Ctx()); err != nil {
+ panic(err)
+ }
+}
+
+// NewLocker creates a sync.Locker backed by an etcd mutex.
+func NewLocker(s *Session, pfx string) sync.Locker {
+ return &lockerMutex{NewMutex(s, pfx)}
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/concurrency/session.go b/vendor/github.com/coreos/etcd/clientv3/concurrency/session.go
new file mode 100644
index 0000000..c399d64
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/concurrency/session.go
@@ -0,0 +1,141 @@
+// 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 concurrency
+
+import (
+ "context"
+ "time"
+
+ v3 "github.com/coreos/etcd/clientv3"
+)
+
+const defaultSessionTTL = 60
+
+// Session represents a lease kept alive for the lifetime of a client.
+// Fault-tolerant applications may use sessions to reason about liveness.
+type Session struct {
+ client *v3.Client
+ opts *sessionOptions
+ id v3.LeaseID
+
+ cancel context.CancelFunc
+ donec <-chan struct{}
+}
+
+// NewSession gets the leased session for a client.
+func NewSession(client *v3.Client, opts ...SessionOption) (*Session, error) {
+ ops := &sessionOptions{ttl: defaultSessionTTL, ctx: client.Ctx()}
+ for _, opt := range opts {
+ opt(ops)
+ }
+
+ id := ops.leaseID
+ if id == v3.NoLease {
+ resp, err := client.Grant(ops.ctx, int64(ops.ttl))
+ if err != nil {
+ return nil, err
+ }
+ id = v3.LeaseID(resp.ID)
+ }
+
+ ctx, cancel := context.WithCancel(ops.ctx)
+ keepAlive, err := client.KeepAlive(ctx, id)
+ if err != nil || keepAlive == nil {
+ cancel()
+ return nil, err
+ }
+
+ donec := make(chan struct{})
+ s := &Session{client: client, opts: ops, id: id, cancel: cancel, donec: donec}
+
+ // keep the lease alive until client error or cancelled context
+ go func() {
+ defer close(donec)
+ for range keepAlive {
+ // eat messages until keep alive channel closes
+ }
+ }()
+
+ return s, nil
+}
+
+// Client is the etcd client that is attached to the session.
+func (s *Session) Client() *v3.Client {
+ return s.client
+}
+
+// Lease is the lease ID for keys bound to the session.
+func (s *Session) Lease() v3.LeaseID { return s.id }
+
+// Done returns a channel that closes when the lease is orphaned, expires, or
+// is otherwise no longer being refreshed.
+func (s *Session) Done() <-chan struct{} { return s.donec }
+
+// Orphan ends the refresh for the session lease. This is useful
+// in case the state of the client connection is indeterminate (revoke
+// would fail) or when transferring lease ownership.
+func (s *Session) Orphan() {
+ s.cancel()
+ <-s.donec
+}
+
+// Close orphans the session and revokes the session lease.
+func (s *Session) Close() error {
+ s.Orphan()
+ // if revoke takes longer than the ttl, lease is expired anyway
+ ctx, cancel := context.WithTimeout(s.opts.ctx, time.Duration(s.opts.ttl)*time.Second)
+ _, err := s.client.Revoke(ctx, s.id)
+ cancel()
+ return err
+}
+
+type sessionOptions struct {
+ ttl int
+ leaseID v3.LeaseID
+ ctx context.Context
+}
+
+// SessionOption configures Session.
+type SessionOption func(*sessionOptions)
+
+// WithTTL configures the session's TTL in seconds.
+// If TTL is <= 0, the default 60 seconds TTL will be used.
+func WithTTL(ttl int) SessionOption {
+ return func(so *sessionOptions) {
+ if ttl > 0 {
+ so.ttl = ttl
+ }
+ }
+}
+
+// WithLease specifies the existing leaseID to be used for the session.
+// This is useful in process restart scenario, for example, to reclaim
+// leadership from an election prior to restart.
+func WithLease(leaseID v3.LeaseID) SessionOption {
+ return func(so *sessionOptions) {
+ so.leaseID = leaseID
+ }
+}
+
+// WithContext assigns a context to the session instead of defaulting to
+// using the client context. This is useful for canceling NewSession and
+// Close operations immediately without having to close the client. If the
+// context is canceled before Close() completes, the session's lease will be
+// abandoned and left to expire instead of being revoked.
+func WithContext(ctx context.Context) SessionOption {
+ return func(so *sessionOptions) {
+ so.ctx = ctx
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/concurrency/stm.go b/vendor/github.com/coreos/etcd/clientv3/concurrency/stm.go
new file mode 100644
index 0000000..d11023e
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/concurrency/stm.go
@@ -0,0 +1,387 @@
+// 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 concurrency
+
+import (
+ "context"
+ "math"
+
+ v3 "github.com/coreos/etcd/clientv3"
+)
+
+// STM is an interface for software transactional memory.
+type STM interface {
+ // Get returns the value for a key and inserts the key in the txn's read set.
+ // If Get fails, it aborts the transaction with an error, never returning.
+ Get(key ...string) string
+ // Put adds a value for a key to the write set.
+ Put(key, val string, opts ...v3.OpOption)
+ // Rev returns the revision of a key in the read set.
+ Rev(key string) int64
+ // Del deletes a key.
+ Del(key string)
+
+ // commit attempts to apply the txn's changes to the server.
+ commit() *v3.TxnResponse
+ reset()
+}
+
+// Isolation is an enumeration of transactional isolation levels which
+// describes how transactions should interfere and conflict.
+type Isolation int
+
+const (
+ // SerializableSnapshot provides serializable isolation and also checks
+ // for write conflicts.
+ SerializableSnapshot Isolation = iota
+ // Serializable reads within the same transaction attempt return data
+ // from the at the revision of the first read.
+ Serializable
+ // RepeatableReads reads within the same transaction attempt always
+ // return the same data.
+ RepeatableReads
+ // ReadCommitted reads keys from any committed revision.
+ ReadCommitted
+)
+
+// stmError safely passes STM errors through panic to the STM error channel.
+type stmError struct{ err error }
+
+type stmOptions struct {
+ iso Isolation
+ ctx context.Context
+ prefetch []string
+}
+
+type stmOption func(*stmOptions)
+
+// WithIsolation specifies the transaction isolation level.
+func WithIsolation(lvl Isolation) stmOption {
+ return func(so *stmOptions) { so.iso = lvl }
+}
+
+// WithAbortContext specifies the context for permanently aborting the transaction.
+func WithAbortContext(ctx context.Context) stmOption {
+ return func(so *stmOptions) { so.ctx = ctx }
+}
+
+// WithPrefetch is a hint to prefetch a list of keys before trying to apply.
+// If an STM transaction will unconditionally fetch a set of keys, prefetching
+// those keys will save the round-trip cost from requesting each key one by one
+// with Get().
+func WithPrefetch(keys ...string) stmOption {
+ return func(so *stmOptions) { so.prefetch = append(so.prefetch, keys...) }
+}
+
+// NewSTM initiates a new STM instance, using serializable snapshot isolation by default.
+func NewSTM(c *v3.Client, apply func(STM) error, so ...stmOption) (*v3.TxnResponse, error) {
+ opts := &stmOptions{ctx: c.Ctx()}
+ for _, f := range so {
+ f(opts)
+ }
+ if len(opts.prefetch) != 0 {
+ f := apply
+ apply = func(s STM) error {
+ s.Get(opts.prefetch...)
+ return f(s)
+ }
+ }
+ return runSTM(mkSTM(c, opts), apply)
+}
+
+func mkSTM(c *v3.Client, opts *stmOptions) STM {
+ switch opts.iso {
+ case SerializableSnapshot:
+ s := &stmSerializable{
+ stm: stm{client: c, ctx: opts.ctx},
+ prefetch: make(map[string]*v3.GetResponse),
+ }
+ s.conflicts = func() []v3.Cmp {
+ return append(s.rset.cmps(), s.wset.cmps(s.rset.first()+1)...)
+ }
+ return s
+ case Serializable:
+ s := &stmSerializable{
+ stm: stm{client: c, ctx: opts.ctx},
+ prefetch: make(map[string]*v3.GetResponse),
+ }
+ s.conflicts = func() []v3.Cmp { return s.rset.cmps() }
+ return s
+ case RepeatableReads:
+ s := &stm{client: c, ctx: opts.ctx, getOpts: []v3.OpOption{v3.WithSerializable()}}
+ s.conflicts = func() []v3.Cmp { return s.rset.cmps() }
+ return s
+ case ReadCommitted:
+ s := &stm{client: c, ctx: opts.ctx, getOpts: []v3.OpOption{v3.WithSerializable()}}
+ s.conflicts = func() []v3.Cmp { return nil }
+ return s
+ default:
+ panic("unsupported stm")
+ }
+}
+
+type stmResponse struct {
+ resp *v3.TxnResponse
+ err error
+}
+
+func runSTM(s STM, apply func(STM) error) (*v3.TxnResponse, error) {
+ outc := make(chan stmResponse, 1)
+ go func() {
+ defer func() {
+ if r := recover(); r != nil {
+ e, ok := r.(stmError)
+ if !ok {
+ // client apply panicked
+ panic(r)
+ }
+ outc <- stmResponse{nil, e.err}
+ }
+ }()
+ var out stmResponse
+ for {
+ s.reset()
+ if out.err = apply(s); out.err != nil {
+ break
+ }
+ if out.resp = s.commit(); out.resp != nil {
+ break
+ }
+ }
+ outc <- out
+ }()
+ r := <-outc
+ return r.resp, r.err
+}
+
+// stm implements repeatable-read software transactional memory over etcd
+type stm struct {
+ client *v3.Client
+ ctx context.Context
+ // rset holds read key values and revisions
+ rset readSet
+ // wset holds overwritten keys and their values
+ wset writeSet
+ // getOpts are the opts used for gets
+ getOpts []v3.OpOption
+ // conflicts computes the current conflicts on the txn
+ conflicts func() []v3.Cmp
+}
+
+type stmPut struct {
+ val string
+ op v3.Op
+}
+
+type readSet map[string]*v3.GetResponse
+
+func (rs readSet) add(keys []string, txnresp *v3.TxnResponse) {
+ for i, resp := range txnresp.Responses {
+ rs[keys[i]] = (*v3.GetResponse)(resp.GetResponseRange())
+ }
+}
+
+// first returns the store revision from the first fetch
+func (rs readSet) first() int64 {
+ ret := int64(math.MaxInt64 - 1)
+ for _, resp := range rs {
+ if rev := resp.Header.Revision; rev < ret {
+ ret = rev
+ }
+ }
+ return ret
+}
+
+// cmps guards the txn from updates to read set
+func (rs readSet) cmps() []v3.Cmp {
+ cmps := make([]v3.Cmp, 0, len(rs))
+ for k, rk := range rs {
+ cmps = append(cmps, isKeyCurrent(k, rk))
+ }
+ return cmps
+}
+
+type writeSet map[string]stmPut
+
+func (ws writeSet) get(keys ...string) *stmPut {
+ for _, key := range keys {
+ if wv, ok := ws[key]; ok {
+ return &wv
+ }
+ }
+ return nil
+}
+
+// cmps returns a cmp list testing no writes have happened past rev
+func (ws writeSet) cmps(rev int64) []v3.Cmp {
+ cmps := make([]v3.Cmp, 0, len(ws))
+ for key := range ws {
+ cmps = append(cmps, v3.Compare(v3.ModRevision(key), "<", rev))
+ }
+ return cmps
+}
+
+// puts is the list of ops for all pending writes
+func (ws writeSet) puts() []v3.Op {
+ puts := make([]v3.Op, 0, len(ws))
+ for _, v := range ws {
+ puts = append(puts, v.op)
+ }
+ return puts
+}
+
+func (s *stm) Get(keys ...string) string {
+ if wv := s.wset.get(keys...); wv != nil {
+ return wv.val
+ }
+ return respToValue(s.fetch(keys...))
+}
+
+func (s *stm) Put(key, val string, opts ...v3.OpOption) {
+ s.wset[key] = stmPut{val, v3.OpPut(key, val, opts...)}
+}
+
+func (s *stm) Del(key string) { s.wset[key] = stmPut{"", v3.OpDelete(key)} }
+
+func (s *stm) Rev(key string) int64 {
+ if resp := s.fetch(key); resp != nil && len(resp.Kvs) != 0 {
+ return resp.Kvs[0].ModRevision
+ }
+ return 0
+}
+
+func (s *stm) commit() *v3.TxnResponse {
+ txnresp, err := s.client.Txn(s.ctx).If(s.conflicts()...).Then(s.wset.puts()...).Commit()
+ if err != nil {
+ panic(stmError{err})
+ }
+ if txnresp.Succeeded {
+ return txnresp
+ }
+ return nil
+}
+
+func (s *stm) fetch(keys ...string) *v3.GetResponse {
+ if len(keys) == 0 {
+ return nil
+ }
+ ops := make([]v3.Op, len(keys))
+ for i, key := range keys {
+ if resp, ok := s.rset[key]; ok {
+ return resp
+ }
+ ops[i] = v3.OpGet(key, s.getOpts...)
+ }
+ txnresp, err := s.client.Txn(s.ctx).Then(ops...).Commit()
+ if err != nil {
+ panic(stmError{err})
+ }
+ s.rset.add(keys, txnresp)
+ return (*v3.GetResponse)(txnresp.Responses[0].GetResponseRange())
+}
+
+func (s *stm) reset() {
+ s.rset = make(map[string]*v3.GetResponse)
+ s.wset = make(map[string]stmPut)
+}
+
+type stmSerializable struct {
+ stm
+ prefetch map[string]*v3.GetResponse
+}
+
+func (s *stmSerializable) Get(keys ...string) string {
+ if wv := s.wset.get(keys...); wv != nil {
+ return wv.val
+ }
+ firstRead := len(s.rset) == 0
+ for _, key := range keys {
+ if resp, ok := s.prefetch[key]; ok {
+ delete(s.prefetch, key)
+ s.rset[key] = resp
+ }
+ }
+ resp := s.stm.fetch(keys...)
+ if firstRead {
+ // txn's base revision is defined by the first read
+ s.getOpts = []v3.OpOption{
+ v3.WithRev(resp.Header.Revision),
+ v3.WithSerializable(),
+ }
+ }
+ return respToValue(resp)
+}
+
+func (s *stmSerializable) Rev(key string) int64 {
+ s.Get(key)
+ return s.stm.Rev(key)
+}
+
+func (s *stmSerializable) gets() ([]string, []v3.Op) {
+ keys := make([]string, 0, len(s.rset))
+ ops := make([]v3.Op, 0, len(s.rset))
+ for k := range s.rset {
+ keys = append(keys, k)
+ ops = append(ops, v3.OpGet(k))
+ }
+ return keys, ops
+}
+
+func (s *stmSerializable) commit() *v3.TxnResponse {
+ keys, getops := s.gets()
+ txn := s.client.Txn(s.ctx).If(s.conflicts()...).Then(s.wset.puts()...)
+ // use Else to prefetch keys in case of conflict to save a round trip
+ txnresp, err := txn.Else(getops...).Commit()
+ if err != nil {
+ panic(stmError{err})
+ }
+ if txnresp.Succeeded {
+ return txnresp
+ }
+ // load prefetch with Else data
+ s.rset.add(keys, txnresp)
+ s.prefetch = s.rset
+ s.getOpts = nil
+ return nil
+}
+
+func isKeyCurrent(k string, r *v3.GetResponse) v3.Cmp {
+ if len(r.Kvs) != 0 {
+ return v3.Compare(v3.ModRevision(k), "=", r.Kvs[0].ModRevision)
+ }
+ return v3.Compare(v3.ModRevision(k), "=", 0)
+}
+
+func respToValue(resp *v3.GetResponse) string {
+ if resp == nil || len(resp.Kvs) == 0 {
+ return ""
+ }
+ return string(resp.Kvs[0].Value)
+}
+
+// NewSTMRepeatable is deprecated.
+func NewSTMRepeatable(ctx context.Context, c *v3.Client, apply func(STM) error) (*v3.TxnResponse, error) {
+ return NewSTM(c, apply, WithAbortContext(ctx), WithIsolation(RepeatableReads))
+}
+
+// NewSTMSerializable is deprecated.
+func NewSTMSerializable(ctx context.Context, c *v3.Client, apply func(STM) error) (*v3.TxnResponse, error) {
+ return NewSTM(c, apply, WithAbortContext(ctx), WithIsolation(Serializable))
+}
+
+// NewSTMReadCommitted is deprecated.
+func NewSTMReadCommitted(ctx context.Context, c *v3.Client, apply func(STM) error) (*v3.TxnResponse, error) {
+ return NewSTM(c, apply, WithAbortContext(ctx), WithIsolation(ReadCommitted))
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/config.go b/vendor/github.com/coreos/etcd/clientv3/config.go
new file mode 100644
index 0000000..79d6e2a
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/config.go
@@ -0,0 +1,75 @@
+// 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 clientv3
+
+import (
+ "context"
+ "crypto/tls"
+ "time"
+
+ "google.golang.org/grpc"
+)
+
+type Config struct {
+ // Endpoints is a list of URLs.
+ Endpoints []string `json:"endpoints"`
+
+ // AutoSyncInterval is the interval to update endpoints with its latest members.
+ // 0 disables auto-sync. By default auto-sync is disabled.
+ AutoSyncInterval time.Duration `json:"auto-sync-interval"`
+
+ // DialTimeout is the timeout for failing to establish a connection.
+ DialTimeout time.Duration `json:"dial-timeout"`
+
+ // DialKeepAliveTime is the time after which client pings the server to see if
+ // transport is alive.
+ DialKeepAliveTime time.Duration `json:"dial-keep-alive-time"`
+
+ // DialKeepAliveTimeout is the time that the client waits for a response for the
+ // keep-alive probe. If the response is not received in this time, the connection is closed.
+ DialKeepAliveTimeout time.Duration `json:"dial-keep-alive-timeout"`
+
+ // MaxCallSendMsgSize is the client-side request send limit in bytes.
+ // If 0, it defaults to 2.0 MiB (2 * 1024 * 1024).
+ // Make sure that "MaxCallSendMsgSize" < server-side default send/recv limit.
+ // ("--max-request-bytes" flag to etcd or "embed.Config.MaxRequestBytes").
+ MaxCallSendMsgSize int
+
+ // MaxCallRecvMsgSize is the client-side response receive limit.
+ // If 0, it defaults to "math.MaxInt32", because range response can
+ // easily exceed request send limits.
+ // Make sure that "MaxCallRecvMsgSize" >= server-side default send/recv limit.
+ // ("--max-request-bytes" flag to etcd or "embed.Config.MaxRequestBytes").
+ MaxCallRecvMsgSize int
+
+ // TLS holds the client secure credentials, if any.
+ TLS *tls.Config
+
+ // Username is a user name for authentication.
+ Username string `json:"username"`
+
+ // Password is a password for authentication.
+ Password string `json:"password"`
+
+ // RejectOldCluster when set will refuse to create a client against an outdated cluster.
+ RejectOldCluster bool `json:"reject-old-cluster"`
+
+ // DialOptions is a list of dial options for the grpc client (e.g., for interceptors).
+ DialOptions []grpc.DialOption
+
+ // Context is the default client context; it can be used to cancel grpc dial out and
+ // other operations that do not have an explicit context.
+ Context context.Context
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/doc.go b/vendor/github.com/coreos/etcd/clientv3/doc.go
new file mode 100644
index 0000000..717fbe4
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/doc.go
@@ -0,0 +1,97 @@
+// 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 clientv3 implements the official Go etcd client for v3.
+//
+// Create client using `clientv3.New`:
+//
+// // expect dial time-out on ipv4 blackhole
+// _, err := clientv3.New(clientv3.Config{
+// Endpoints: []string{"http://254.0.0.1:12345"},
+// DialTimeout: 2 * time.Second
+// })
+//
+// // etcd clientv3 >= v3.2.10, grpc/grpc-go >= v1.7.3
+// if err == context.DeadlineExceeded {
+// // handle errors
+// }
+//
+// // etcd clientv3 <= v3.2.9, grpc/grpc-go <= v1.2.1
+// if err == grpc.ErrClientConnTimeout {
+// // handle errors
+// }
+//
+// cli, err := clientv3.New(clientv3.Config{
+// Endpoints: []string{"localhost:2379", "localhost:22379", "localhost:32379"},
+// DialTimeout: 5 * time.Second,
+// })
+// if err != nil {
+// // handle error!
+// }
+// defer cli.Close()
+//
+// Make sure to close the client after using it. If the client is not closed, the
+// connection will have leaky goroutines.
+//
+// To specify a client request timeout, wrap the context with context.WithTimeout:
+//
+// ctx, cancel := context.WithTimeout(context.Background(), timeout)
+// resp, err := kvc.Put(ctx, "sample_key", "sample_value")
+// cancel()
+// if err != nil {
+// // handle error!
+// }
+// // use the response
+//
+// The Client has internal state (watchers and leases), so Clients should be reused instead of created as needed.
+// Clients are safe for concurrent use by multiple goroutines.
+//
+// etcd client returns 3 types of errors:
+//
+// 1. context error: canceled or deadline exceeded.
+// 2. gRPC status error: e.g. when clock drifts in server-side before client's context deadline exceeded.
+// 3. gRPC error: see https://github.com/coreos/etcd/blob/master/etcdserver/api/v3rpc/rpctypes/error.go
+//
+// Here is the example code to handle client errors:
+//
+// resp, err := kvc.Put(ctx, "", "")
+// if err != nil {
+// if err == context.Canceled {
+// // ctx is canceled by another routine
+// } else if err == context.DeadlineExceeded {
+// // ctx is attached with a deadline and it exceeded
+// } else if ev, ok := status.FromError(err); ok {
+// code := ev.Code()
+// if code == codes.DeadlineExceeded {
+// // server-side context might have timed-out first (due to clock skew)
+// // while original client-side context is not timed-out yet
+// }
+// } else if verr, ok := err.(*v3rpc.ErrEmptyKey); ok {
+// // process (verr.Errors)
+// } else {
+// // bad cluster endpoints, which are not etcd servers
+// }
+// }
+//
+// go func() { cli.Close() }()
+// _, err := kvc.Get(ctx, "a")
+// if err != nil {
+// if err == context.Canceled {
+// // grpc balancer calls 'Get' with an inflight client.Close
+// } else if err == grpc.ErrClientConnClosing {
+// // grpc balancer calls 'Get' after client.Close.
+// }
+// }
+//
+package clientv3
diff --git a/vendor/github.com/coreos/etcd/clientv3/health_balancer.go b/vendor/github.com/coreos/etcd/clientv3/health_balancer.go
new file mode 100644
index 0000000..5918cba
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/health_balancer.go
@@ -0,0 +1,609 @@
+// Copyright 2017 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 clientv3
+
+import (
+ "context"
+ "errors"
+ "net/url"
+ "strings"
+ "sync"
+ "time"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ healthpb "google.golang.org/grpc/health/grpc_health_v1"
+ "google.golang.org/grpc/status"
+)
+
+const (
+ minHealthRetryDuration = 3 * time.Second
+ unknownService = "unknown service grpc.health.v1.Health"
+)
+
+// ErrNoAddrAvilable is returned by Get() when the balancer does not have
+// any active connection to endpoints at the time.
+// This error is returned only when opts.BlockingWait is true.
+var ErrNoAddrAvilable = status.Error(codes.Unavailable, "there is no address available")
+
+type healthCheckFunc func(ep string) (bool, error)
+
+type notifyMsg int
+
+const (
+ notifyReset notifyMsg = iota
+ notifyNext
+)
+
+// healthBalancer does the bare minimum to expose multiple eps
+// to the grpc reconnection code path
+type healthBalancer struct {
+ // addrs are the client's endpoint addresses for grpc
+ addrs []grpc.Address
+
+ // eps holds the raw endpoints from the client
+ eps []string
+
+ // notifyCh notifies grpc of the set of addresses for connecting
+ notifyCh chan []grpc.Address
+
+ // readyc closes once the first connection is up
+ readyc chan struct{}
+ readyOnce sync.Once
+
+ // healthCheck checks an endpoint's health.
+ healthCheck healthCheckFunc
+ healthCheckTimeout time.Duration
+
+ unhealthyMu sync.RWMutex
+ unhealthyHostPorts map[string]time.Time
+
+ // mu protects all fields below.
+ mu sync.RWMutex
+
+ // upc closes when pinAddr transitions from empty to non-empty or the balancer closes.
+ upc chan struct{}
+
+ // downc closes when grpc calls down() on pinAddr
+ downc chan struct{}
+
+ // stopc is closed to signal updateNotifyLoop should stop.
+ stopc chan struct{}
+ stopOnce sync.Once
+ wg sync.WaitGroup
+
+ // donec closes when all goroutines are exited
+ donec chan struct{}
+
+ // updateAddrsC notifies updateNotifyLoop to update addrs.
+ updateAddrsC chan notifyMsg
+
+ // grpc issues TLS cert checks using the string passed into dial so
+ // that string must be the host. To recover the full scheme://host URL,
+ // have a map from hosts to the original endpoint.
+ hostPort2ep map[string]string
+
+ // pinAddr is the currently pinned address; set to the empty string on
+ // initialization and shutdown.
+ pinAddr string
+
+ closed bool
+}
+
+func newHealthBalancer(eps []string, timeout time.Duration, hc healthCheckFunc) *healthBalancer {
+ notifyCh := make(chan []grpc.Address)
+ addrs := eps2addrs(eps)
+ hb := &healthBalancer{
+ addrs: addrs,
+ eps: eps,
+ notifyCh: notifyCh,
+ readyc: make(chan struct{}),
+ healthCheck: hc,
+ unhealthyHostPorts: make(map[string]time.Time),
+ upc: make(chan struct{}),
+ stopc: make(chan struct{}),
+ downc: make(chan struct{}),
+ donec: make(chan struct{}),
+ updateAddrsC: make(chan notifyMsg),
+ hostPort2ep: getHostPort2ep(eps),
+ }
+ if timeout < minHealthRetryDuration {
+ timeout = minHealthRetryDuration
+ }
+ hb.healthCheckTimeout = timeout
+
+ close(hb.downc)
+ go hb.updateNotifyLoop()
+ hb.wg.Add(1)
+ go func() {
+ defer hb.wg.Done()
+ hb.updateUnhealthy()
+ }()
+ return hb
+}
+
+func (b *healthBalancer) Start(target string, config grpc.BalancerConfig) error { return nil }
+
+func (b *healthBalancer) ConnectNotify() <-chan struct{} {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.upc
+}
+
+func (b *healthBalancer) ready() <-chan struct{} { return b.readyc }
+
+func (b *healthBalancer) endpoint(hostPort string) string {
+ b.mu.RLock()
+ defer b.mu.RUnlock()
+ return b.hostPort2ep[hostPort]
+}
+
+func (b *healthBalancer) pinned() string {
+ b.mu.RLock()
+ defer b.mu.RUnlock()
+ return b.pinAddr
+}
+
+func (b *healthBalancer) hostPortError(hostPort string, err error) {
+ if b.endpoint(hostPort) == "" {
+ logger.Lvl(4).Infof("clientv3/balancer: %q is stale (skip marking as unhealthy on %q)", hostPort, err.Error())
+ return
+ }
+
+ b.unhealthyMu.Lock()
+ b.unhealthyHostPorts[hostPort] = time.Now()
+ b.unhealthyMu.Unlock()
+ logger.Lvl(4).Infof("clientv3/balancer: %q is marked unhealthy (%q)", hostPort, err.Error())
+}
+
+func (b *healthBalancer) removeUnhealthy(hostPort, msg string) {
+ if b.endpoint(hostPort) == "" {
+ logger.Lvl(4).Infof("clientv3/balancer: %q was not in unhealthy (%q)", hostPort, msg)
+ return
+ }
+
+ b.unhealthyMu.Lock()
+ delete(b.unhealthyHostPorts, hostPort)
+ b.unhealthyMu.Unlock()
+ logger.Lvl(4).Infof("clientv3/balancer: %q is removed from unhealthy (%q)", hostPort, msg)
+}
+
+func (b *healthBalancer) countUnhealthy() (count int) {
+ b.unhealthyMu.RLock()
+ count = len(b.unhealthyHostPorts)
+ b.unhealthyMu.RUnlock()
+ return count
+}
+
+func (b *healthBalancer) isUnhealthy(hostPort string) (unhealthy bool) {
+ b.unhealthyMu.RLock()
+ _, unhealthy = b.unhealthyHostPorts[hostPort]
+ b.unhealthyMu.RUnlock()
+ return unhealthy
+}
+
+func (b *healthBalancer) cleanupUnhealthy() {
+ b.unhealthyMu.Lock()
+ for k, v := range b.unhealthyHostPorts {
+ if time.Since(v) > b.healthCheckTimeout {
+ delete(b.unhealthyHostPorts, k)
+ logger.Lvl(4).Infof("clientv3/balancer: removed %q from unhealthy after %v", k, b.healthCheckTimeout)
+ }
+ }
+ b.unhealthyMu.Unlock()
+}
+
+func (b *healthBalancer) liveAddrs() ([]grpc.Address, map[string]struct{}) {
+ unhealthyCnt := b.countUnhealthy()
+
+ b.mu.RLock()
+ defer b.mu.RUnlock()
+
+ hbAddrs := b.addrs
+ if len(b.addrs) == 1 || unhealthyCnt == 0 || unhealthyCnt == len(b.addrs) {
+ liveHostPorts := make(map[string]struct{}, len(b.hostPort2ep))
+ for k := range b.hostPort2ep {
+ liveHostPorts[k] = struct{}{}
+ }
+ return hbAddrs, liveHostPorts
+ }
+
+ addrs := make([]grpc.Address, 0, len(b.addrs)-unhealthyCnt)
+ liveHostPorts := make(map[string]struct{}, len(addrs))
+ for _, addr := range b.addrs {
+ if !b.isUnhealthy(addr.Addr) {
+ addrs = append(addrs, addr)
+ liveHostPorts[addr.Addr] = struct{}{}
+ }
+ }
+ return addrs, liveHostPorts
+}
+
+func (b *healthBalancer) updateUnhealthy() {
+ for {
+ select {
+ case <-time.After(b.healthCheckTimeout):
+ b.cleanupUnhealthy()
+ pinned := b.pinned()
+ if pinned == "" || b.isUnhealthy(pinned) {
+ select {
+ case b.updateAddrsC <- notifyNext:
+ case <-b.stopc:
+ return
+ }
+ }
+ case <-b.stopc:
+ return
+ }
+ }
+}
+
+func (b *healthBalancer) updateAddrs(eps ...string) {
+ np := getHostPort2ep(eps)
+
+ b.mu.Lock()
+ defer b.mu.Unlock()
+
+ match := len(np) == len(b.hostPort2ep)
+ if match {
+ for k, v := range np {
+ if b.hostPort2ep[k] != v {
+ match = false
+ break
+ }
+ }
+ }
+ if match {
+ // same endpoints, so no need to update address
+ return
+ }
+
+ b.hostPort2ep = np
+ b.addrs, b.eps = eps2addrs(eps), eps
+
+ b.unhealthyMu.Lock()
+ b.unhealthyHostPorts = make(map[string]time.Time)
+ b.unhealthyMu.Unlock()
+}
+
+func (b *healthBalancer) next() {
+ b.mu.RLock()
+ downc := b.downc
+ b.mu.RUnlock()
+ select {
+ case b.updateAddrsC <- notifyNext:
+ case <-b.stopc:
+ }
+ // wait until disconnect so new RPCs are not issued on old connection
+ select {
+ case <-downc:
+ case <-b.stopc:
+ }
+}
+
+func (b *healthBalancer) updateNotifyLoop() {
+ defer close(b.donec)
+
+ for {
+ b.mu.RLock()
+ upc, downc, addr := b.upc, b.downc, b.pinAddr
+ b.mu.RUnlock()
+ // downc or upc should be closed
+ select {
+ case <-downc:
+ downc = nil
+ default:
+ }
+ select {
+ case <-upc:
+ upc = nil
+ default:
+ }
+ switch {
+ case downc == nil && upc == nil:
+ // stale
+ select {
+ case <-b.stopc:
+ return
+ default:
+ }
+ case downc == nil:
+ b.notifyAddrs(notifyReset)
+ select {
+ case <-upc:
+ case msg := <-b.updateAddrsC:
+ b.notifyAddrs(msg)
+ case <-b.stopc:
+ return
+ }
+ case upc == nil:
+ select {
+ // close connections that are not the pinned address
+ case b.notifyCh <- []grpc.Address{{Addr: addr}}:
+ case <-downc:
+ case <-b.stopc:
+ return
+ }
+ select {
+ case <-downc:
+ b.notifyAddrs(notifyReset)
+ case msg := <-b.updateAddrsC:
+ b.notifyAddrs(msg)
+ case <-b.stopc:
+ return
+ }
+ }
+ }
+}
+
+func (b *healthBalancer) notifyAddrs(msg notifyMsg) {
+ if msg == notifyNext {
+ select {
+ case b.notifyCh <- []grpc.Address{}:
+ case <-b.stopc:
+ return
+ }
+ }
+ b.mu.RLock()
+ pinAddr := b.pinAddr
+ downc := b.downc
+ b.mu.RUnlock()
+ addrs, hostPorts := b.liveAddrs()
+
+ var waitDown bool
+ if pinAddr != "" {
+ _, ok := hostPorts[pinAddr]
+ waitDown = !ok
+ }
+
+ select {
+ case b.notifyCh <- addrs:
+ if waitDown {
+ select {
+ case <-downc:
+ case <-b.stopc:
+ }
+ }
+ case <-b.stopc:
+ }
+}
+
+func (b *healthBalancer) Up(addr grpc.Address) func(error) {
+ if !b.mayPin(addr) {
+ return func(err error) {}
+ }
+
+ b.mu.Lock()
+ defer b.mu.Unlock()
+
+ // gRPC might call Up after it called Close. We add this check
+ // to "fix" it up at application layer. Otherwise, will panic
+ // if b.upc is already closed.
+ if b.closed {
+ return func(err error) {}
+ }
+
+ // gRPC might call Up on a stale address.
+ // Prevent updating pinAddr with a stale address.
+ if !hasAddr(b.addrs, addr.Addr) {
+ return func(err error) {}
+ }
+
+ if b.pinAddr != "" {
+ logger.Lvl(4).Infof("clientv3/balancer: %q is up but not pinned (already pinned %q)", addr.Addr, b.pinAddr)
+ return func(err error) {}
+ }
+
+ // notify waiting Get()s and pin first connected address
+ close(b.upc)
+ b.downc = make(chan struct{})
+ b.pinAddr = addr.Addr
+ logger.Lvl(4).Infof("clientv3/balancer: pin %q", addr.Addr)
+
+ // notify client that a connection is up
+ b.readyOnce.Do(func() { close(b.readyc) })
+
+ return func(err error) {
+ // If connected to a black hole endpoint or a killed server, the gRPC ping
+ // timeout will induce a network I/O error, and retrying until success;
+ // finding healthy endpoint on retry could take several timeouts and redials.
+ // To avoid wasting retries, gray-list unhealthy endpoints.
+ b.hostPortError(addr.Addr, err)
+
+ b.mu.Lock()
+ b.upc = make(chan struct{})
+ close(b.downc)
+ b.pinAddr = ""
+ b.mu.Unlock()
+ logger.Lvl(4).Infof("clientv3/balancer: unpin %q (%q)", addr.Addr, err.Error())
+ }
+}
+
+func (b *healthBalancer) mayPin(addr grpc.Address) bool {
+ if b.endpoint(addr.Addr) == "" { // stale host:port
+ return false
+ }
+
+ b.unhealthyMu.RLock()
+ unhealthyCnt := len(b.unhealthyHostPorts)
+ failedTime, bad := b.unhealthyHostPorts[addr.Addr]
+ b.unhealthyMu.RUnlock()
+
+ b.mu.RLock()
+ skip := len(b.addrs) == 1 || unhealthyCnt == 0 || len(b.addrs) == unhealthyCnt
+ b.mu.RUnlock()
+ if skip || !bad {
+ return true
+ }
+
+ // prevent isolated member's endpoint from being infinitely retried, as follows:
+ // 1. keepalive pings detects GoAway with http2.ErrCodeEnhanceYourCalm
+ // 2. balancer 'Up' unpins with grpc: failed with network I/O error
+ // 3. grpc-healthcheck still SERVING, thus retry to pin
+ // instead, return before grpc-healthcheck if failed within healthcheck timeout
+ if elapsed := time.Since(failedTime); elapsed < b.healthCheckTimeout {
+ logger.Lvl(4).Infof("clientv3/balancer: %q is up but not pinned (failed %v ago, require minimum %v after failure)", addr.Addr, elapsed, b.healthCheckTimeout)
+ return false
+ }
+
+ if ok, _ := b.healthCheck(addr.Addr); ok {
+ b.removeUnhealthy(addr.Addr, "health check success")
+ return true
+ }
+
+ b.hostPortError(addr.Addr, errors.New("health check failed"))
+ return false
+}
+
+func (b *healthBalancer) Get(ctx context.Context, opts grpc.BalancerGetOptions) (grpc.Address, func(), error) {
+ var (
+ addr string
+ closed bool
+ )
+
+ // If opts.BlockingWait is false (for fail-fast RPCs), it should return
+ // an address it has notified via Notify immediately instead of blocking.
+ if !opts.BlockingWait {
+ b.mu.RLock()
+ closed = b.closed
+ addr = b.pinAddr
+ b.mu.RUnlock()
+ if closed {
+ return grpc.Address{Addr: ""}, nil, grpc.ErrClientConnClosing
+ }
+ if addr == "" {
+ return grpc.Address{Addr: ""}, nil, ErrNoAddrAvilable
+ }
+ return grpc.Address{Addr: addr}, func() {}, nil
+ }
+
+ for {
+ b.mu.RLock()
+ ch := b.upc
+ b.mu.RUnlock()
+ select {
+ case <-ch:
+ case <-b.donec:
+ return grpc.Address{Addr: ""}, nil, grpc.ErrClientConnClosing
+ case <-ctx.Done():
+ return grpc.Address{Addr: ""}, nil, ctx.Err()
+ }
+ b.mu.RLock()
+ closed = b.closed
+ addr = b.pinAddr
+ b.mu.RUnlock()
+ // Close() which sets b.closed = true can be called before Get(), Get() must exit if balancer is closed.
+ if closed {
+ return grpc.Address{Addr: ""}, nil, grpc.ErrClientConnClosing
+ }
+ if addr != "" {
+ break
+ }
+ }
+ return grpc.Address{Addr: addr}, func() {}, nil
+}
+
+func (b *healthBalancer) Notify() <-chan []grpc.Address { return b.notifyCh }
+
+func (b *healthBalancer) Close() error {
+ b.mu.Lock()
+ // In case gRPC calls close twice. TODO: remove the checking
+ // when we are sure that gRPC wont call close twice.
+ if b.closed {
+ b.mu.Unlock()
+ <-b.donec
+ return nil
+ }
+ b.closed = true
+ b.stopOnce.Do(func() { close(b.stopc) })
+ b.pinAddr = ""
+
+ // In the case of following scenario:
+ // 1. upc is not closed; no pinned address
+ // 2. client issues an RPC, calling invoke(), which calls Get(), enters for loop, blocks
+ // 3. client.conn.Close() calls balancer.Close(); closed = true
+ // 4. for loop in Get() never exits since ctx is the context passed in by the client and may not be canceled
+ // we must close upc so Get() exits from blocking on upc
+ select {
+ case <-b.upc:
+ default:
+ // terminate all waiting Get()s
+ close(b.upc)
+ }
+
+ b.mu.Unlock()
+ b.wg.Wait()
+
+ // wait for updateNotifyLoop to finish
+ <-b.donec
+ close(b.notifyCh)
+
+ return nil
+}
+
+func grpcHealthCheck(client *Client, ep string) (bool, error) {
+ conn, err := client.dial(ep)
+ if err != nil {
+ return false, err
+ }
+ defer conn.Close()
+ cli := healthpb.NewHealthClient(conn)
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ resp, err := cli.Check(ctx, &healthpb.HealthCheckRequest{})
+ cancel()
+ if err != nil {
+ if s, ok := status.FromError(err); ok && s.Code() == codes.Unavailable {
+ if s.Message() == unknownService { // etcd < v3.3.0
+ return true, nil
+ }
+ }
+ return false, err
+ }
+ return resp.Status == healthpb.HealthCheckResponse_SERVING, nil
+}
+
+func hasAddr(addrs []grpc.Address, targetAddr string) bool {
+ for _, addr := range addrs {
+ if targetAddr == addr.Addr {
+ return true
+ }
+ }
+ return false
+}
+
+func getHost(ep string) string {
+ url, uerr := url.Parse(ep)
+ if uerr != nil || !strings.Contains(ep, "://") {
+ return ep
+ }
+ return url.Host
+}
+
+func eps2addrs(eps []string) []grpc.Address {
+ addrs := make([]grpc.Address, len(eps))
+ for i := range eps {
+ addrs[i].Addr = getHost(eps[i])
+ }
+ return addrs
+}
+
+func getHostPort2ep(eps []string) map[string]string {
+ hm := make(map[string]string, len(eps))
+ for i := range eps {
+ _, host, _ := parseEndpoint(eps[i])
+ hm[host] = eps[i]
+ }
+ return hm
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/kv.go b/vendor/github.com/coreos/etcd/clientv3/kv.go
new file mode 100644
index 0000000..5a7469b
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/kv.go
@@ -0,0 +1,177 @@
+// Copyright 2015 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 clientv3
+
+import (
+ "context"
+
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+ "google.golang.org/grpc"
+)
+
+type (
+ CompactResponse pb.CompactionResponse
+ PutResponse pb.PutResponse
+ GetResponse pb.RangeResponse
+ DeleteResponse pb.DeleteRangeResponse
+ TxnResponse pb.TxnResponse
+)
+
+type KV interface {
+ // Put puts a key-value pair into etcd.
+ // Note that key,value can be plain bytes array and string is
+ // an immutable representation of that bytes array.
+ // To get a string of bytes, do string([]byte{0x10, 0x20}).
+ Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error)
+
+ // Get retrieves keys.
+ // By default, Get will return the value for "key", if any.
+ // When passed WithRange(end), Get will return the keys in the range [key, end).
+ // When passed WithFromKey(), Get returns keys greater than or equal to key.
+ // When passed WithRev(rev) with rev > 0, Get retrieves keys at the given revision;
+ // if the required revision is compacted, the request will fail with ErrCompacted .
+ // When passed WithLimit(limit), the number of returned keys is bounded by limit.
+ // When passed WithSort(), the keys will be sorted.
+ Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error)
+
+ // Delete deletes a key, or optionally using WithRange(end), [key, end).
+ Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error)
+
+ // Compact compacts etcd KV history before the given rev.
+ Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error)
+
+ // Do applies a single Op on KV without a transaction.
+ // Do is useful when creating arbitrary operations to be issued at a
+ // later time; the user can range over the operations, calling Do to
+ // execute them. Get/Put/Delete, on the other hand, are best suited
+ // for when the operation should be issued at the time of declaration.
+ Do(ctx context.Context, op Op) (OpResponse, error)
+
+ // Txn creates a transaction.
+ Txn(ctx context.Context) Txn
+}
+
+type OpResponse struct {
+ put *PutResponse
+ get *GetResponse
+ del *DeleteResponse
+ txn *TxnResponse
+}
+
+func (op OpResponse) Put() *PutResponse { return op.put }
+func (op OpResponse) Get() *GetResponse { return op.get }
+func (op OpResponse) Del() *DeleteResponse { return op.del }
+func (op OpResponse) Txn() *TxnResponse { return op.txn }
+
+func (resp *PutResponse) OpResponse() OpResponse {
+ return OpResponse{put: resp}
+}
+func (resp *GetResponse) OpResponse() OpResponse {
+ return OpResponse{get: resp}
+}
+func (resp *DeleteResponse) OpResponse() OpResponse {
+ return OpResponse{del: resp}
+}
+func (resp *TxnResponse) OpResponse() OpResponse {
+ return OpResponse{txn: resp}
+}
+
+type kv struct {
+ remote pb.KVClient
+ callOpts []grpc.CallOption
+}
+
+func NewKV(c *Client) KV {
+ api := &kv{remote: RetryKVClient(c)}
+ if c != nil {
+ api.callOpts = c.callOpts
+ }
+ return api
+}
+
+func NewKVFromKVClient(remote pb.KVClient, c *Client) KV {
+ api := &kv{remote: remote}
+ if c != nil {
+ api.callOpts = c.callOpts
+ }
+ return api
+}
+
+func (kv *kv) Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error) {
+ r, err := kv.Do(ctx, OpPut(key, val, opts...))
+ return r.put, toErr(ctx, err)
+}
+
+func (kv *kv) Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error) {
+ r, err := kv.Do(ctx, OpGet(key, opts...))
+ return r.get, toErr(ctx, err)
+}
+
+func (kv *kv) Delete(ctx context.Context, key string, opts ...OpOption) (*DeleteResponse, error) {
+ r, err := kv.Do(ctx, OpDelete(key, opts...))
+ return r.del, toErr(ctx, err)
+}
+
+func (kv *kv) Compact(ctx context.Context, rev int64, opts ...CompactOption) (*CompactResponse, error) {
+ resp, err := kv.remote.Compact(ctx, OpCompact(rev, opts...).toRequest(), kv.callOpts...)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ return (*CompactResponse)(resp), err
+}
+
+func (kv *kv) Txn(ctx context.Context) Txn {
+ return &txn{
+ kv: kv,
+ ctx: ctx,
+ callOpts: kv.callOpts,
+ }
+}
+
+func (kv *kv) Do(ctx context.Context, op Op) (OpResponse, error) {
+ var err error
+ switch op.t {
+ case tRange:
+ var resp *pb.RangeResponse
+ resp, err = kv.remote.Range(ctx, op.toRangeRequest(), kv.callOpts...)
+ if err == nil {
+ return OpResponse{get: (*GetResponse)(resp)}, nil
+ }
+ case tPut:
+ var resp *pb.PutResponse
+ r := &pb.PutRequest{Key: op.key, Value: op.val, Lease: int64(op.leaseID), PrevKv: op.prevKV, IgnoreValue: op.ignoreValue, IgnoreLease: op.ignoreLease}
+ resp, err = kv.remote.Put(ctx, r, kv.callOpts...)
+ if err == nil {
+ return OpResponse{put: (*PutResponse)(resp)}, nil
+ }
+ case tDeleteRange:
+ var resp *pb.DeleteRangeResponse
+ r := &pb.DeleteRangeRequest{Key: op.key, RangeEnd: op.end, PrevKv: op.prevKV}
+ resp, err = kv.remote.DeleteRange(ctx, r, kv.callOpts...)
+ if err == nil {
+ return OpResponse{del: (*DeleteResponse)(resp)}, nil
+ }
+ case tTxn:
+ var resp *pb.TxnResponse
+ resp, err = kv.remote.Txn(ctx, op.toTxnRequest(), kv.callOpts...)
+ if err == nil {
+ return OpResponse{txn: (*TxnResponse)(resp)}, nil
+ }
+ default:
+ panic("Unknown op")
+ }
+ return OpResponse{}, toErr(ctx, err)
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/lease.go b/vendor/github.com/coreos/etcd/clientv3/lease.go
new file mode 100644
index 0000000..3729cf3
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/lease.go
@@ -0,0 +1,588 @@
+// 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 clientv3
+
+import (
+ "context"
+ "sync"
+ "time"
+
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/metadata"
+)
+
+type (
+ LeaseRevokeResponse pb.LeaseRevokeResponse
+ LeaseID int64
+)
+
+// LeaseGrantResponse wraps the protobuf message LeaseGrantResponse.
+type LeaseGrantResponse struct {
+ *pb.ResponseHeader
+ ID LeaseID
+ TTL int64
+ Error string
+}
+
+// LeaseKeepAliveResponse wraps the protobuf message LeaseKeepAliveResponse.
+type LeaseKeepAliveResponse struct {
+ *pb.ResponseHeader
+ ID LeaseID
+ TTL int64
+}
+
+// LeaseTimeToLiveResponse wraps the protobuf message LeaseTimeToLiveResponse.
+type LeaseTimeToLiveResponse struct {
+ *pb.ResponseHeader
+ ID LeaseID `json:"id"`
+
+ // TTL is the remaining TTL in seconds for the lease; the lease will expire in under TTL+1 seconds. Expired lease will return -1.
+ TTL int64 `json:"ttl"`
+
+ // GrantedTTL is the initial granted time in seconds upon lease creation/renewal.
+ GrantedTTL int64 `json:"granted-ttl"`
+
+ // Keys is the list of keys attached to this lease.
+ Keys [][]byte `json:"keys"`
+}
+
+// LeaseStatus represents a lease status.
+type LeaseStatus struct {
+ ID LeaseID `json:"id"`
+ // TODO: TTL int64
+}
+
+// LeaseLeasesResponse wraps the protobuf message LeaseLeasesResponse.
+type LeaseLeasesResponse struct {
+ *pb.ResponseHeader
+ Leases []LeaseStatus `json:"leases"`
+}
+
+const (
+ // defaultTTL is the assumed lease TTL used for the first keepalive
+ // deadline before the actual TTL is known to the client.
+ defaultTTL = 5 * time.Second
+ // NoLease is a lease ID for the absence of a lease.
+ NoLease LeaseID = 0
+
+ // retryConnWait is how long to wait before retrying request due to an error
+ retryConnWait = 500 * time.Millisecond
+)
+
+// LeaseResponseChSize is the size of buffer to store unsent lease responses.
+// WARNING: DO NOT UPDATE.
+// Only for testing purposes.
+var LeaseResponseChSize = 16
+
+// ErrKeepAliveHalted is returned if client keep alive loop halts with an unexpected error.
+//
+// This usually means that automatic lease renewal via KeepAlive is broken, but KeepAliveOnce will still work as expected.
+type ErrKeepAliveHalted struct {
+ Reason error
+}
+
+func (e ErrKeepAliveHalted) Error() string {
+ s := "etcdclient: leases keep alive halted"
+ if e.Reason != nil {
+ s += ": " + e.Reason.Error()
+ }
+ return s
+}
+
+type Lease interface {
+ // Grant creates a new lease.
+ Grant(ctx context.Context, ttl int64) (*LeaseGrantResponse, error)
+
+ // Revoke revokes the given lease.
+ Revoke(ctx context.Context, id LeaseID) (*LeaseRevokeResponse, error)
+
+ // TimeToLive retrieves the lease information of the given lease ID.
+ TimeToLive(ctx context.Context, id LeaseID, opts ...LeaseOption) (*LeaseTimeToLiveResponse, error)
+
+ // Leases retrieves all leases.
+ Leases(ctx context.Context) (*LeaseLeasesResponse, error)
+
+ // KeepAlive keeps the given lease alive forever. If the keepalive response
+ // posted to the channel is not consumed immediately, the lease client will
+ // continue sending keep alive requests to the etcd server at least every
+ // second until latest response is consumed.
+ //
+ // The returned "LeaseKeepAliveResponse" channel closes if underlying keep
+ // alive stream is interrupted in some way the client cannot handle itself;
+ // given context "ctx" is canceled or timed out. "LeaseKeepAliveResponse"
+ // from this closed channel is nil.
+ //
+ // If client keep alive loop halts with an unexpected error (e.g. "etcdserver:
+ // no leader") or canceled by the caller (e.g. context.Canceled), the error
+ // is returned. Otherwise, it retries.
+ //
+ // TODO(v4.0): post errors to last keep alive message before closing
+ // (see https://github.com/coreos/etcd/pull/7866)
+ KeepAlive(ctx context.Context, id LeaseID) (<-chan *LeaseKeepAliveResponse, error)
+
+ // KeepAliveOnce renews the lease once. The response corresponds to the
+ // first message from calling KeepAlive. If the response has a recoverable
+ // error, KeepAliveOnce will retry the RPC with a new keep alive message.
+ //
+ // In most of the cases, Keepalive should be used instead of KeepAliveOnce.
+ KeepAliveOnce(ctx context.Context, id LeaseID) (*LeaseKeepAliveResponse, error)
+
+ // Close releases all resources Lease keeps for efficient communication
+ // with the etcd server.
+ Close() error
+}
+
+type lessor struct {
+ mu sync.Mutex // guards all fields
+
+ // donec is closed and loopErr is set when recvKeepAliveLoop stops
+ donec chan struct{}
+ loopErr error
+
+ remote pb.LeaseClient
+
+ stream pb.Lease_LeaseKeepAliveClient
+ streamCancel context.CancelFunc
+
+ stopCtx context.Context
+ stopCancel context.CancelFunc
+
+ keepAlives map[LeaseID]*keepAlive
+
+ // firstKeepAliveTimeout is the timeout for the first keepalive request
+ // before the actual TTL is known to the lease client
+ firstKeepAliveTimeout time.Duration
+
+ // firstKeepAliveOnce ensures stream starts after first KeepAlive call.
+ firstKeepAliveOnce sync.Once
+
+ callOpts []grpc.CallOption
+}
+
+// keepAlive multiplexes a keepalive for a lease over multiple channels
+type keepAlive struct {
+ chs []chan<- *LeaseKeepAliveResponse
+ ctxs []context.Context
+ // deadline is the time the keep alive channels close if no response
+ deadline time.Time
+ // nextKeepAlive is when to send the next keep alive message
+ nextKeepAlive time.Time
+ // donec is closed on lease revoke, expiration, or cancel.
+ donec chan struct{}
+}
+
+func NewLease(c *Client) Lease {
+ return NewLeaseFromLeaseClient(RetryLeaseClient(c), c, c.cfg.DialTimeout+time.Second)
+}
+
+func NewLeaseFromLeaseClient(remote pb.LeaseClient, c *Client, keepAliveTimeout time.Duration) Lease {
+ l := &lessor{
+ donec: make(chan struct{}),
+ keepAlives: make(map[LeaseID]*keepAlive),
+ remote: remote,
+ firstKeepAliveTimeout: keepAliveTimeout,
+ }
+ if l.firstKeepAliveTimeout == time.Second {
+ l.firstKeepAliveTimeout = defaultTTL
+ }
+ if c != nil {
+ l.callOpts = c.callOpts
+ }
+ reqLeaderCtx := WithRequireLeader(context.Background())
+ l.stopCtx, l.stopCancel = context.WithCancel(reqLeaderCtx)
+ return l
+}
+
+func (l *lessor) Grant(ctx context.Context, ttl int64) (*LeaseGrantResponse, error) {
+ r := &pb.LeaseGrantRequest{TTL: ttl}
+ resp, err := l.remote.LeaseGrant(ctx, r, l.callOpts...)
+ if err == nil {
+ gresp := &LeaseGrantResponse{
+ ResponseHeader: resp.GetHeader(),
+ ID: LeaseID(resp.ID),
+ TTL: resp.TTL,
+ Error: resp.Error,
+ }
+ return gresp, nil
+ }
+ return nil, toErr(ctx, err)
+}
+
+func (l *lessor) Revoke(ctx context.Context, id LeaseID) (*LeaseRevokeResponse, error) {
+ r := &pb.LeaseRevokeRequest{ID: int64(id)}
+ resp, err := l.remote.LeaseRevoke(ctx, r, l.callOpts...)
+ if err == nil {
+ return (*LeaseRevokeResponse)(resp), nil
+ }
+ return nil, toErr(ctx, err)
+}
+
+func (l *lessor) TimeToLive(ctx context.Context, id LeaseID, opts ...LeaseOption) (*LeaseTimeToLiveResponse, error) {
+ r := toLeaseTimeToLiveRequest(id, opts...)
+ resp, err := l.remote.LeaseTimeToLive(ctx, r, l.callOpts...)
+ if err == nil {
+ gresp := &LeaseTimeToLiveResponse{
+ ResponseHeader: resp.GetHeader(),
+ ID: LeaseID(resp.ID),
+ TTL: resp.TTL,
+ GrantedTTL: resp.GrantedTTL,
+ Keys: resp.Keys,
+ }
+ return gresp, nil
+ }
+ return nil, toErr(ctx, err)
+}
+
+func (l *lessor) Leases(ctx context.Context) (*LeaseLeasesResponse, error) {
+ resp, err := l.remote.LeaseLeases(ctx, &pb.LeaseLeasesRequest{}, l.callOpts...)
+ if err == nil {
+ leases := make([]LeaseStatus, len(resp.Leases))
+ for i := range resp.Leases {
+ leases[i] = LeaseStatus{ID: LeaseID(resp.Leases[i].ID)}
+ }
+ return &LeaseLeasesResponse{ResponseHeader: resp.GetHeader(), Leases: leases}, nil
+ }
+ return nil, toErr(ctx, err)
+}
+
+func (l *lessor) KeepAlive(ctx context.Context, id LeaseID) (<-chan *LeaseKeepAliveResponse, error) {
+ ch := make(chan *LeaseKeepAliveResponse, LeaseResponseChSize)
+
+ l.mu.Lock()
+ // ensure that recvKeepAliveLoop is still running
+ select {
+ case <-l.donec:
+ err := l.loopErr
+ l.mu.Unlock()
+ close(ch)
+ return ch, ErrKeepAliveHalted{Reason: err}
+ default:
+ }
+ ka, ok := l.keepAlives[id]
+ if !ok {
+ // create fresh keep alive
+ ka = &keepAlive{
+ chs: []chan<- *LeaseKeepAliveResponse{ch},
+ ctxs: []context.Context{ctx},
+ deadline: time.Now().Add(l.firstKeepAliveTimeout),
+ nextKeepAlive: time.Now(),
+ donec: make(chan struct{}),
+ }
+ l.keepAlives[id] = ka
+ } else {
+ // add channel and context to existing keep alive
+ ka.ctxs = append(ka.ctxs, ctx)
+ ka.chs = append(ka.chs, ch)
+ }
+ l.mu.Unlock()
+
+ go l.keepAliveCtxCloser(id, ctx, ka.donec)
+ l.firstKeepAliveOnce.Do(func() {
+ go l.recvKeepAliveLoop()
+ go l.deadlineLoop()
+ })
+
+ return ch, nil
+}
+
+func (l *lessor) KeepAliveOnce(ctx context.Context, id LeaseID) (*LeaseKeepAliveResponse, error) {
+ for {
+ resp, err := l.keepAliveOnce(ctx, id)
+ if err == nil {
+ if resp.TTL <= 0 {
+ err = rpctypes.ErrLeaseNotFound
+ }
+ return resp, err
+ }
+ if isHaltErr(ctx, err) {
+ return nil, toErr(ctx, err)
+ }
+ }
+}
+
+func (l *lessor) Close() error {
+ l.stopCancel()
+ // close for synchronous teardown if stream goroutines never launched
+ l.firstKeepAliveOnce.Do(func() { close(l.donec) })
+ <-l.donec
+ return nil
+}
+
+func (l *lessor) keepAliveCtxCloser(id LeaseID, ctx context.Context, donec <-chan struct{}) {
+ select {
+ case <-donec:
+ return
+ case <-l.donec:
+ return
+ case <-ctx.Done():
+ }
+
+ l.mu.Lock()
+ defer l.mu.Unlock()
+
+ ka, ok := l.keepAlives[id]
+ if !ok {
+ return
+ }
+
+ // close channel and remove context if still associated with keep alive
+ for i, c := range ka.ctxs {
+ if c == ctx {
+ close(ka.chs[i])
+ ka.ctxs = append(ka.ctxs[:i], ka.ctxs[i+1:]...)
+ ka.chs = append(ka.chs[:i], ka.chs[i+1:]...)
+ break
+ }
+ }
+ // remove if no one more listeners
+ if len(ka.chs) == 0 {
+ delete(l.keepAlives, id)
+ }
+}
+
+// closeRequireLeader scans keepAlives for ctxs that have require leader
+// and closes the associated channels.
+func (l *lessor) closeRequireLeader() {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ for _, ka := range l.keepAlives {
+ reqIdxs := 0
+ // find all required leader channels, close, mark as nil
+ for i, ctx := range ka.ctxs {
+ md, ok := metadata.FromOutgoingContext(ctx)
+ if !ok {
+ continue
+ }
+ ks := md[rpctypes.MetadataRequireLeaderKey]
+ if len(ks) < 1 || ks[0] != rpctypes.MetadataHasLeader {
+ continue
+ }
+ close(ka.chs[i])
+ ka.chs[i] = nil
+ reqIdxs++
+ }
+ if reqIdxs == 0 {
+ continue
+ }
+ // remove all channels that required a leader from keepalive
+ newChs := make([]chan<- *LeaseKeepAliveResponse, len(ka.chs)-reqIdxs)
+ newCtxs := make([]context.Context, len(newChs))
+ newIdx := 0
+ for i := range ka.chs {
+ if ka.chs[i] == nil {
+ continue
+ }
+ newChs[newIdx], newCtxs[newIdx] = ka.chs[i], ka.ctxs[newIdx]
+ newIdx++
+ }
+ ka.chs, ka.ctxs = newChs, newCtxs
+ }
+}
+
+func (l *lessor) keepAliveOnce(ctx context.Context, id LeaseID) (*LeaseKeepAliveResponse, error) {
+ cctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+
+ stream, err := l.remote.LeaseKeepAlive(cctx, l.callOpts...)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+
+ err = stream.Send(&pb.LeaseKeepAliveRequest{ID: int64(id)})
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+
+ resp, rerr := stream.Recv()
+ if rerr != nil {
+ return nil, toErr(ctx, rerr)
+ }
+
+ karesp := &LeaseKeepAliveResponse{
+ ResponseHeader: resp.GetHeader(),
+ ID: LeaseID(resp.ID),
+ TTL: resp.TTL,
+ }
+ return karesp, nil
+}
+
+func (l *lessor) recvKeepAliveLoop() (gerr error) {
+ defer func() {
+ l.mu.Lock()
+ close(l.donec)
+ l.loopErr = gerr
+ for _, ka := range l.keepAlives {
+ ka.close()
+ }
+ l.keepAlives = make(map[LeaseID]*keepAlive)
+ l.mu.Unlock()
+ }()
+
+ for {
+ stream, err := l.resetRecv()
+ if err != nil {
+ if canceledByCaller(l.stopCtx, err) {
+ return err
+ }
+ } else {
+ for {
+ resp, err := stream.Recv()
+ if err != nil {
+ if canceledByCaller(l.stopCtx, err) {
+ return err
+ }
+
+ if toErr(l.stopCtx, err) == rpctypes.ErrNoLeader {
+ l.closeRequireLeader()
+ }
+ break
+ }
+
+ l.recvKeepAlive(resp)
+ }
+ }
+
+ select {
+ case <-time.After(retryConnWait):
+ continue
+ case <-l.stopCtx.Done():
+ return l.stopCtx.Err()
+ }
+ }
+}
+
+// resetRecv opens a new lease stream and starts sending keep alive requests.
+func (l *lessor) resetRecv() (pb.Lease_LeaseKeepAliveClient, error) {
+ sctx, cancel := context.WithCancel(l.stopCtx)
+ stream, err := l.remote.LeaseKeepAlive(sctx, l.callOpts...)
+ if err != nil {
+ cancel()
+ return nil, err
+ }
+
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ if l.stream != nil && l.streamCancel != nil {
+ l.streamCancel()
+ }
+
+ l.streamCancel = cancel
+ l.stream = stream
+
+ go l.sendKeepAliveLoop(stream)
+ return stream, nil
+}
+
+// recvKeepAlive updates a lease based on its LeaseKeepAliveResponse
+func (l *lessor) recvKeepAlive(resp *pb.LeaseKeepAliveResponse) {
+ karesp := &LeaseKeepAliveResponse{
+ ResponseHeader: resp.GetHeader(),
+ ID: LeaseID(resp.ID),
+ TTL: resp.TTL,
+ }
+
+ l.mu.Lock()
+ defer l.mu.Unlock()
+
+ ka, ok := l.keepAlives[karesp.ID]
+ if !ok {
+ return
+ }
+
+ if karesp.TTL <= 0 {
+ // lease expired; close all keep alive channels
+ delete(l.keepAlives, karesp.ID)
+ ka.close()
+ return
+ }
+
+ // send update to all channels
+ nextKeepAlive := time.Now().Add((time.Duration(karesp.TTL) * time.Second) / 3.0)
+ ka.deadline = time.Now().Add(time.Duration(karesp.TTL) * time.Second)
+ for _, ch := range ka.chs {
+ select {
+ case ch <- karesp:
+ default:
+ }
+ // still advance in order to rate-limit keep-alive sends
+ ka.nextKeepAlive = nextKeepAlive
+ }
+}
+
+// deadlineLoop reaps any keep alive channels that have not received a response
+// within the lease TTL
+func (l *lessor) deadlineLoop() {
+ for {
+ select {
+ case <-time.After(time.Second):
+ case <-l.donec:
+ return
+ }
+ now := time.Now()
+ l.mu.Lock()
+ for id, ka := range l.keepAlives {
+ if ka.deadline.Before(now) {
+ // waited too long for response; lease may be expired
+ ka.close()
+ delete(l.keepAlives, id)
+ }
+ }
+ l.mu.Unlock()
+ }
+}
+
+// sendKeepAliveLoop sends keep alive requests for the lifetime of the given stream.
+func (l *lessor) sendKeepAliveLoop(stream pb.Lease_LeaseKeepAliveClient) {
+ for {
+ var tosend []LeaseID
+
+ now := time.Now()
+ l.mu.Lock()
+ for id, ka := range l.keepAlives {
+ if ka.nextKeepAlive.Before(now) {
+ tosend = append(tosend, id)
+ }
+ }
+ l.mu.Unlock()
+
+ for _, id := range tosend {
+ r := &pb.LeaseKeepAliveRequest{ID: int64(id)}
+ if err := stream.Send(r); err != nil {
+ // TODO do something with this error?
+ return
+ }
+ }
+
+ select {
+ case <-time.After(500 * time.Millisecond):
+ case <-stream.Context().Done():
+ return
+ case <-l.donec:
+ return
+ case <-l.stopCtx.Done():
+ return
+ }
+ }
+}
+
+func (ka *keepAlive) close() {
+ close(ka.donec)
+ for _, ch := range ka.chs {
+ close(ch)
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/leasing/cache.go b/vendor/github.com/coreos/etcd/clientv3/leasing/cache.go
new file mode 100644
index 0000000..6903a78
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/leasing/cache.go
@@ -0,0 +1,306 @@
+// Copyright 2017 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 leasing
+
+import (
+ "context"
+ "strings"
+ "sync"
+ "time"
+
+ v3 "github.com/coreos/etcd/clientv3"
+ v3pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/mvcc/mvccpb"
+)
+
+const revokeBackoff = 2 * time.Second
+
+type leaseCache struct {
+ mu sync.RWMutex
+ entries map[string]*leaseKey
+ revokes map[string]time.Time
+ header *v3pb.ResponseHeader
+}
+
+type leaseKey struct {
+ response *v3.GetResponse
+ // rev is the leasing key revision.
+ rev int64
+ waitc chan struct{}
+}
+
+func (lc *leaseCache) Rev(key string) int64 {
+ lc.mu.RLock()
+ defer lc.mu.RUnlock()
+ if li := lc.entries[key]; li != nil {
+ return li.rev
+ }
+ return 0
+}
+
+func (lc *leaseCache) Lock(key string) (chan<- struct{}, int64) {
+ lc.mu.Lock()
+ defer lc.mu.Unlock()
+ if li := lc.entries[key]; li != nil {
+ li.waitc = make(chan struct{})
+ return li.waitc, li.rev
+ }
+ return nil, 0
+}
+
+func (lc *leaseCache) LockRange(begin, end string) (ret []chan<- struct{}) {
+ lc.mu.Lock()
+ defer lc.mu.Unlock()
+ for k, li := range lc.entries {
+ if inRange(k, begin, end) {
+ li.waitc = make(chan struct{})
+ ret = append(ret, li.waitc)
+ }
+ }
+ return ret
+}
+
+func inRange(k, begin, end string) bool {
+ if strings.Compare(k, begin) < 0 {
+ return false
+ }
+ if end != "\x00" && strings.Compare(k, end) >= 0 {
+ return false
+ }
+ return true
+}
+
+func (lc *leaseCache) LockWriteOps(ops []v3.Op) (ret []chan<- struct{}) {
+ for _, op := range ops {
+ if op.IsGet() {
+ continue
+ }
+ key := string(op.KeyBytes())
+ if end := string(op.RangeBytes()); end == "" {
+ if wc, _ := lc.Lock(key); wc != nil {
+ ret = append(ret, wc)
+ }
+ } else {
+ for k := range lc.entries {
+ if !inRange(k, key, end) {
+ continue
+ }
+ if wc, _ := lc.Lock(k); wc != nil {
+ ret = append(ret, wc)
+ }
+ }
+ }
+ }
+ return ret
+}
+
+func (lc *leaseCache) NotifyOps(ops []v3.Op) (wcs []<-chan struct{}) {
+ for _, op := range ops {
+ if op.IsGet() {
+ if _, wc := lc.notify(string(op.KeyBytes())); wc != nil {
+ wcs = append(wcs, wc)
+ }
+ }
+ }
+ return wcs
+}
+
+func (lc *leaseCache) MayAcquire(key string) bool {
+ lc.mu.RLock()
+ lr, ok := lc.revokes[key]
+ lc.mu.RUnlock()
+ return !ok || time.Since(lr) > revokeBackoff
+}
+
+func (lc *leaseCache) Add(key string, resp *v3.GetResponse, op v3.Op) *v3.GetResponse {
+ lk := &leaseKey{resp, resp.Header.Revision, closedCh}
+ lc.mu.Lock()
+ if lc.header == nil || lc.header.Revision < resp.Header.Revision {
+ lc.header = resp.Header
+ }
+ lc.entries[key] = lk
+ ret := lk.get(op)
+ lc.mu.Unlock()
+ return ret
+}
+
+func (lc *leaseCache) Update(key, val []byte, respHeader *v3pb.ResponseHeader) {
+ li := lc.entries[string(key)]
+ if li == nil {
+ return
+ }
+ cacheResp := li.response
+ if len(cacheResp.Kvs) == 0 {
+ kv := &mvccpb.KeyValue{
+ Key: key,
+ CreateRevision: respHeader.Revision,
+ }
+ cacheResp.Kvs = append(cacheResp.Kvs, kv)
+ cacheResp.Count = 1
+ }
+ cacheResp.Kvs[0].Version++
+ if cacheResp.Kvs[0].ModRevision < respHeader.Revision {
+ cacheResp.Header = respHeader
+ cacheResp.Kvs[0].ModRevision = respHeader.Revision
+ cacheResp.Kvs[0].Value = val
+ }
+}
+
+func (lc *leaseCache) Delete(key string, hdr *v3pb.ResponseHeader) {
+ lc.mu.Lock()
+ defer lc.mu.Unlock()
+ lc.delete(key, hdr)
+}
+
+func (lc *leaseCache) delete(key string, hdr *v3pb.ResponseHeader) {
+ if li := lc.entries[key]; li != nil && hdr.Revision >= li.response.Header.Revision {
+ li.response.Kvs = nil
+ li.response.Header = copyHeader(hdr)
+ }
+}
+
+func (lc *leaseCache) Evict(key string) (rev int64) {
+ lc.mu.Lock()
+ defer lc.mu.Unlock()
+ if li := lc.entries[key]; li != nil {
+ rev = li.rev
+ delete(lc.entries, key)
+ lc.revokes[key] = time.Now()
+ }
+ return rev
+}
+
+func (lc *leaseCache) EvictRange(key, end string) {
+ lc.mu.Lock()
+ defer lc.mu.Unlock()
+ for k := range lc.entries {
+ if inRange(k, key, end) {
+ delete(lc.entries, key)
+ lc.revokes[key] = time.Now()
+ }
+ }
+}
+
+func isBadOp(op v3.Op) bool { return op.Rev() > 0 || len(op.RangeBytes()) > 0 }
+
+func (lc *leaseCache) Get(ctx context.Context, op v3.Op) (*v3.GetResponse, bool) {
+ if isBadOp(op) {
+ return nil, false
+ }
+ key := string(op.KeyBytes())
+ li, wc := lc.notify(key)
+ if li == nil {
+ return nil, true
+ }
+ select {
+ case <-wc:
+ case <-ctx.Done():
+ return nil, true
+ }
+ lc.mu.RLock()
+ lk := *li
+ ret := lk.get(op)
+ lc.mu.RUnlock()
+ return ret, true
+}
+
+func (lk *leaseKey) get(op v3.Op) *v3.GetResponse {
+ ret := *lk.response
+ ret.Header = copyHeader(ret.Header)
+ empty := len(ret.Kvs) == 0 || op.IsCountOnly()
+ empty = empty || (op.MinModRev() > ret.Kvs[0].ModRevision)
+ empty = empty || (op.MaxModRev() != 0 && op.MaxModRev() < ret.Kvs[0].ModRevision)
+ empty = empty || (op.MinCreateRev() > ret.Kvs[0].CreateRevision)
+ empty = empty || (op.MaxCreateRev() != 0 && op.MaxCreateRev() < ret.Kvs[0].CreateRevision)
+ if empty {
+ ret.Kvs = nil
+ } else {
+ kv := *ret.Kvs[0]
+ kv.Key = make([]byte, len(kv.Key))
+ copy(kv.Key, ret.Kvs[0].Key)
+ if !op.IsKeysOnly() {
+ kv.Value = make([]byte, len(kv.Value))
+ copy(kv.Value, ret.Kvs[0].Value)
+ }
+ ret.Kvs = []*mvccpb.KeyValue{&kv}
+ }
+ return &ret
+}
+
+func (lc *leaseCache) notify(key string) (*leaseKey, <-chan struct{}) {
+ lc.mu.RLock()
+ defer lc.mu.RUnlock()
+ if li := lc.entries[key]; li != nil {
+ return li, li.waitc
+ }
+ return nil, nil
+}
+
+func (lc *leaseCache) clearOldRevokes(ctx context.Context) {
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(time.Second):
+ lc.mu.Lock()
+ for k, lr := range lc.revokes {
+ if time.Now().Sub(lr.Add(revokeBackoff)) > 0 {
+ delete(lc.revokes, k)
+ }
+ }
+ lc.mu.Unlock()
+ }
+ }
+}
+
+func (lc *leaseCache) evalCmp(cmps []v3.Cmp) (cmpVal bool, ok bool) {
+ for _, cmp := range cmps {
+ if len(cmp.RangeEnd) > 0 {
+ return false, false
+ }
+ lk := lc.entries[string(cmp.Key)]
+ if lk == nil {
+ return false, false
+ }
+ if !evalCmp(lk.response, cmp) {
+ return false, true
+ }
+ }
+ return true, true
+}
+
+func (lc *leaseCache) evalOps(ops []v3.Op) ([]*v3pb.ResponseOp, bool) {
+ resps := make([]*v3pb.ResponseOp, len(ops))
+ for i, op := range ops {
+ if !op.IsGet() || isBadOp(op) {
+ // TODO: support read-only Txn
+ return nil, false
+ }
+ lk := lc.entries[string(op.KeyBytes())]
+ if lk == nil {
+ return nil, false
+ }
+ resp := lk.get(op)
+ if resp == nil {
+ return nil, false
+ }
+ resps[i] = &v3pb.ResponseOp{
+ Response: &v3pb.ResponseOp_ResponseRange{
+ (*v3pb.RangeResponse)(resp),
+ },
+ }
+ }
+ return resps, true
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/leasing/doc.go b/vendor/github.com/coreos/etcd/clientv3/leasing/doc.go
new file mode 100644
index 0000000..fc97fc8
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/leasing/doc.go
@@ -0,0 +1,46 @@
+// Copyright 2017 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 leasing serves linearizable reads from a local cache by acquiring
+// exclusive write access to keys through a client-side leasing protocol. This
+// leasing layer can either directly wrap the etcd client or it can be exposed
+// through the etcd grpc proxy server, granting multiple clients write access.
+//
+// First, create a leasing KV from a clientv3.Client 'cli':
+//
+// lkv, err := leasing.NewKV(cli, "leasing-prefix")
+// if err != nil {
+// // handle error
+// }
+//
+// A range request for a key "abc" tries to acquire a leasing key so it can cache the range's
+// key locally. On the server, the leasing key is stored to "leasing-prefix/abc":
+//
+// resp, err := lkv.Get(context.TODO(), "abc")
+//
+// Future linearized read requests using 'lkv' will be served locally for the lease's lifetime:
+//
+// resp, err = lkv.Get(context.TODO(), "abc")
+//
+// If another leasing client writes to a leased key, then the owner relinquishes its exclusive
+// access, permitting the writer to modify the key:
+//
+// lkv2, err := leasing.NewKV(cli, "leasing-prefix")
+// if err != nil {
+// // handle error
+// }
+// lkv2.Put(context.TODO(), "abc", "456")
+// resp, err = lkv.Get("abc")
+//
+package leasing
diff --git a/vendor/github.com/coreos/etcd/clientv3/leasing/kv.go b/vendor/github.com/coreos/etcd/clientv3/leasing/kv.go
new file mode 100644
index 0000000..5a5e231
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/leasing/kv.go
@@ -0,0 +1,479 @@
+// Copyright 2017 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 leasing
+
+import (
+ "context"
+ "strings"
+ "sync"
+ "time"
+
+ v3 "github.com/coreos/etcd/clientv3"
+ "github.com/coreos/etcd/clientv3/concurrency"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/mvcc/mvccpb"
+
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+type leasingKV struct {
+ cl *v3.Client
+ kv v3.KV
+ pfx string
+ leases leaseCache
+
+ ctx context.Context
+ cancel context.CancelFunc
+ wg sync.WaitGroup
+
+ sessionOpts []concurrency.SessionOption
+ session *concurrency.Session
+ sessionc chan struct{}
+}
+
+var closedCh chan struct{}
+
+func init() {
+ closedCh = make(chan struct{})
+ close(closedCh)
+}
+
+// NewKV wraps a KV instance so that all requests are wired through a leasing protocol.
+func NewKV(cl *v3.Client, pfx string, opts ...concurrency.SessionOption) (v3.KV, func(), error) {
+ cctx, cancel := context.WithCancel(cl.Ctx())
+ lkv := &leasingKV{
+ cl: cl,
+ kv: cl.KV,
+ pfx: pfx,
+ leases: leaseCache{revokes: make(map[string]time.Time)},
+ ctx: cctx,
+ cancel: cancel,
+ sessionOpts: opts,
+ sessionc: make(chan struct{}),
+ }
+ lkv.wg.Add(2)
+ go func() {
+ defer lkv.wg.Done()
+ lkv.monitorSession()
+ }()
+ go func() {
+ defer lkv.wg.Done()
+ lkv.leases.clearOldRevokes(cctx)
+ }()
+ return lkv, lkv.Close, lkv.waitSession(cctx)
+}
+
+func (lkv *leasingKV) Close() {
+ lkv.cancel()
+ lkv.wg.Wait()
+}
+
+func (lkv *leasingKV) Get(ctx context.Context, key string, opts ...v3.OpOption) (*v3.GetResponse, error) {
+ return lkv.get(ctx, v3.OpGet(key, opts...))
+}
+
+func (lkv *leasingKV) Put(ctx context.Context, key, val string, opts ...v3.OpOption) (*v3.PutResponse, error) {
+ return lkv.put(ctx, v3.OpPut(key, val, opts...))
+}
+
+func (lkv *leasingKV) Delete(ctx context.Context, key string, opts ...v3.OpOption) (*v3.DeleteResponse, error) {
+ return lkv.delete(ctx, v3.OpDelete(key, opts...))
+}
+
+func (lkv *leasingKV) Do(ctx context.Context, op v3.Op) (v3.OpResponse, error) {
+ switch {
+ case op.IsGet():
+ resp, err := lkv.get(ctx, op)
+ return resp.OpResponse(), err
+ case op.IsPut():
+ resp, err := lkv.put(ctx, op)
+ return resp.OpResponse(), err
+ case op.IsDelete():
+ resp, err := lkv.delete(ctx, op)
+ return resp.OpResponse(), err
+ case op.IsTxn():
+ cmps, thenOps, elseOps := op.Txn()
+ resp, err := lkv.Txn(ctx).If(cmps...).Then(thenOps...).Else(elseOps...).Commit()
+ return resp.OpResponse(), err
+ }
+ return v3.OpResponse{}, nil
+}
+
+func (lkv *leasingKV) Compact(ctx context.Context, rev int64, opts ...v3.CompactOption) (*v3.CompactResponse, error) {
+ return lkv.kv.Compact(ctx, rev, opts...)
+}
+
+func (lkv *leasingKV) Txn(ctx context.Context) v3.Txn {
+ return &txnLeasing{Txn: lkv.kv.Txn(ctx), lkv: lkv, ctx: ctx}
+}
+
+func (lkv *leasingKV) monitorSession() {
+ for lkv.ctx.Err() == nil {
+ if lkv.session != nil {
+ select {
+ case <-lkv.session.Done():
+ case <-lkv.ctx.Done():
+ return
+ }
+ }
+ lkv.leases.mu.Lock()
+ select {
+ case <-lkv.sessionc:
+ lkv.sessionc = make(chan struct{})
+ default:
+ }
+ lkv.leases.entries = make(map[string]*leaseKey)
+ lkv.leases.mu.Unlock()
+
+ s, err := concurrency.NewSession(lkv.cl, lkv.sessionOpts...)
+ if err != nil {
+ continue
+ }
+
+ lkv.leases.mu.Lock()
+ lkv.session = s
+ close(lkv.sessionc)
+ lkv.leases.mu.Unlock()
+ }
+}
+
+func (lkv *leasingKV) monitorLease(ctx context.Context, key string, rev int64) {
+ cctx, cancel := context.WithCancel(lkv.ctx)
+ defer cancel()
+ for cctx.Err() == nil {
+ if rev == 0 {
+ resp, err := lkv.kv.Get(ctx, lkv.pfx+key)
+ if err != nil {
+ continue
+ }
+ rev = resp.Header.Revision
+ if len(resp.Kvs) == 0 || string(resp.Kvs[0].Value) == "REVOKE" {
+ lkv.rescind(cctx, key, rev)
+ return
+ }
+ }
+ wch := lkv.cl.Watch(cctx, lkv.pfx+key, v3.WithRev(rev+1))
+ for resp := range wch {
+ for _, ev := range resp.Events {
+ if string(ev.Kv.Value) != "REVOKE" {
+ continue
+ }
+ if v3.LeaseID(ev.Kv.Lease) == lkv.leaseID() {
+ lkv.rescind(cctx, key, ev.Kv.ModRevision)
+ }
+ return
+ }
+ }
+ rev = 0
+ }
+}
+
+// rescind releases a lease from this client.
+func (lkv *leasingKV) rescind(ctx context.Context, key string, rev int64) {
+ if lkv.leases.Evict(key) > rev {
+ return
+ }
+ cmp := v3.Compare(v3.CreateRevision(lkv.pfx+key), "<", rev)
+ op := v3.OpDelete(lkv.pfx + key)
+ for ctx.Err() == nil {
+ if _, err := lkv.kv.Txn(ctx).If(cmp).Then(op).Commit(); err == nil {
+ return
+ }
+ }
+}
+
+func (lkv *leasingKV) waitRescind(ctx context.Context, key string, rev int64) error {
+ cctx, cancel := context.WithCancel(ctx)
+ defer cancel()
+ wch := lkv.cl.Watch(cctx, lkv.pfx+key, v3.WithRev(rev+1))
+ for resp := range wch {
+ for _, ev := range resp.Events {
+ if ev.Type == v3.EventTypeDelete {
+ return ctx.Err()
+ }
+ }
+ }
+ return ctx.Err()
+}
+
+func (lkv *leasingKV) tryModifyOp(ctx context.Context, op v3.Op) (*v3.TxnResponse, chan<- struct{}, error) {
+ key := string(op.KeyBytes())
+ wc, rev := lkv.leases.Lock(key)
+ cmp := v3.Compare(v3.CreateRevision(lkv.pfx+key), "<", rev+1)
+ resp, err := lkv.kv.Txn(ctx).If(cmp).Then(op).Commit()
+ switch {
+ case err != nil:
+ lkv.leases.Evict(key)
+ fallthrough
+ case !resp.Succeeded:
+ if wc != nil {
+ close(wc)
+ }
+ return nil, nil, err
+ }
+ return resp, wc, nil
+}
+
+func (lkv *leasingKV) put(ctx context.Context, op v3.Op) (pr *v3.PutResponse, err error) {
+ if err := lkv.waitSession(ctx); err != nil {
+ return nil, err
+ }
+ for ctx.Err() == nil {
+ resp, wc, err := lkv.tryModifyOp(ctx, op)
+ if err != nil || wc == nil {
+ resp, err = lkv.revoke(ctx, string(op.KeyBytes()), op)
+ }
+ if err != nil {
+ return nil, err
+ }
+ if resp.Succeeded {
+ lkv.leases.mu.Lock()
+ lkv.leases.Update(op.KeyBytes(), op.ValueBytes(), resp.Header)
+ lkv.leases.mu.Unlock()
+ pr = (*v3.PutResponse)(resp.Responses[0].GetResponsePut())
+ pr.Header = resp.Header
+ }
+ if wc != nil {
+ close(wc)
+ }
+ if resp.Succeeded {
+ return pr, nil
+ }
+ }
+ return nil, ctx.Err()
+}
+
+func (lkv *leasingKV) acquire(ctx context.Context, key string, op v3.Op) (*v3.TxnResponse, error) {
+ for ctx.Err() == nil {
+ if err := lkv.waitSession(ctx); err != nil {
+ return nil, err
+ }
+ lcmp := v3.Cmp{Key: []byte(key), Target: pb.Compare_LEASE}
+ resp, err := lkv.kv.Txn(ctx).If(
+ v3.Compare(v3.CreateRevision(lkv.pfx+key), "=", 0),
+ v3.Compare(lcmp, "=", 0)).
+ Then(
+ op,
+ v3.OpPut(lkv.pfx+key, "", v3.WithLease(lkv.leaseID()))).
+ Else(
+ op,
+ v3.OpGet(lkv.pfx+key),
+ ).Commit()
+ if err == nil {
+ if !resp.Succeeded {
+ kvs := resp.Responses[1].GetResponseRange().Kvs
+ // if txn failed since already owner, lease is acquired
+ resp.Succeeded = len(kvs) > 0 && v3.LeaseID(kvs[0].Lease) == lkv.leaseID()
+ }
+ return resp, nil
+ }
+ // retry if transient error
+ if _, ok := err.(rpctypes.EtcdError); ok {
+ return nil, err
+ }
+ if ev, _ := status.FromError(err); ev.Code() != codes.Unavailable {
+ return nil, err
+ }
+ }
+ return nil, ctx.Err()
+}
+
+func (lkv *leasingKV) get(ctx context.Context, op v3.Op) (*v3.GetResponse, error) {
+ do := func() (*v3.GetResponse, error) {
+ r, err := lkv.kv.Do(ctx, op)
+ return r.Get(), err
+ }
+ if !lkv.readySession() {
+ return do()
+ }
+
+ if resp, ok := lkv.leases.Get(ctx, op); resp != nil {
+ return resp, nil
+ } else if !ok || op.IsSerializable() {
+ // must be handled by server or can skip linearization
+ return do()
+ }
+
+ key := string(op.KeyBytes())
+ if !lkv.leases.MayAcquire(key) {
+ resp, err := lkv.kv.Do(ctx, op)
+ return resp.Get(), err
+ }
+
+ resp, err := lkv.acquire(ctx, key, v3.OpGet(key))
+ if err != nil {
+ return nil, err
+ }
+ getResp := (*v3.GetResponse)(resp.Responses[0].GetResponseRange())
+ getResp.Header = resp.Header
+ if resp.Succeeded {
+ getResp = lkv.leases.Add(key, getResp, op)
+ lkv.wg.Add(1)
+ go func() {
+ defer lkv.wg.Done()
+ lkv.monitorLease(ctx, key, resp.Header.Revision)
+ }()
+ }
+ return getResp, nil
+}
+
+func (lkv *leasingKV) deleteRangeRPC(ctx context.Context, maxLeaseRev int64, key, end string) (*v3.DeleteResponse, error) {
+ lkey, lend := lkv.pfx+key, lkv.pfx+end
+ resp, err := lkv.kv.Txn(ctx).If(
+ v3.Compare(v3.CreateRevision(lkey).WithRange(lend), "<", maxLeaseRev+1),
+ ).Then(
+ v3.OpGet(key, v3.WithRange(end), v3.WithKeysOnly()),
+ v3.OpDelete(key, v3.WithRange(end)),
+ ).Commit()
+ if err != nil {
+ lkv.leases.EvictRange(key, end)
+ return nil, err
+ }
+ if !resp.Succeeded {
+ return nil, nil
+ }
+ for _, kv := range resp.Responses[0].GetResponseRange().Kvs {
+ lkv.leases.Delete(string(kv.Key), resp.Header)
+ }
+ delResp := (*v3.DeleteResponse)(resp.Responses[1].GetResponseDeleteRange())
+ delResp.Header = resp.Header
+ return delResp, nil
+}
+
+func (lkv *leasingKV) deleteRange(ctx context.Context, op v3.Op) (*v3.DeleteResponse, error) {
+ key, end := string(op.KeyBytes()), string(op.RangeBytes())
+ for ctx.Err() == nil {
+ maxLeaseRev, err := lkv.revokeRange(ctx, key, end)
+ if err != nil {
+ return nil, err
+ }
+ wcs := lkv.leases.LockRange(key, end)
+ delResp, err := lkv.deleteRangeRPC(ctx, maxLeaseRev, key, end)
+ closeAll(wcs)
+ if err != nil || delResp != nil {
+ return delResp, err
+ }
+ }
+ return nil, ctx.Err()
+}
+
+func (lkv *leasingKV) delete(ctx context.Context, op v3.Op) (dr *v3.DeleteResponse, err error) {
+ if err := lkv.waitSession(ctx); err != nil {
+ return nil, err
+ }
+ if len(op.RangeBytes()) > 0 {
+ return lkv.deleteRange(ctx, op)
+ }
+ key := string(op.KeyBytes())
+ for ctx.Err() == nil {
+ resp, wc, err := lkv.tryModifyOp(ctx, op)
+ if err != nil || wc == nil {
+ resp, err = lkv.revoke(ctx, key, op)
+ }
+ if err != nil {
+ // don't know if delete was processed
+ lkv.leases.Evict(key)
+ return nil, err
+ }
+ if resp.Succeeded {
+ dr = (*v3.DeleteResponse)(resp.Responses[0].GetResponseDeleteRange())
+ dr.Header = resp.Header
+ lkv.leases.Delete(key, dr.Header)
+ }
+ if wc != nil {
+ close(wc)
+ }
+ if resp.Succeeded {
+ return dr, nil
+ }
+ }
+ return nil, ctx.Err()
+}
+
+func (lkv *leasingKV) revoke(ctx context.Context, key string, op v3.Op) (*v3.TxnResponse, error) {
+ rev := lkv.leases.Rev(key)
+ txn := lkv.kv.Txn(ctx).If(v3.Compare(v3.CreateRevision(lkv.pfx+key), "<", rev+1)).Then(op)
+ resp, err := txn.Else(v3.OpPut(lkv.pfx+key, "REVOKE", v3.WithIgnoreLease())).Commit()
+ if err != nil || resp.Succeeded {
+ return resp, err
+ }
+ return resp, lkv.waitRescind(ctx, key, resp.Header.Revision)
+}
+
+func (lkv *leasingKV) revokeRange(ctx context.Context, begin, end string) (int64, error) {
+ lkey, lend := lkv.pfx+begin, ""
+ if len(end) > 0 {
+ lend = lkv.pfx + end
+ }
+ leaseKeys, err := lkv.kv.Get(ctx, lkey, v3.WithRange(lend))
+ if err != nil {
+ return 0, err
+ }
+ return lkv.revokeLeaseKvs(ctx, leaseKeys.Kvs)
+}
+
+func (lkv *leasingKV) revokeLeaseKvs(ctx context.Context, kvs []*mvccpb.KeyValue) (int64, error) {
+ maxLeaseRev := int64(0)
+ for _, kv := range kvs {
+ if rev := kv.CreateRevision; rev > maxLeaseRev {
+ maxLeaseRev = rev
+ }
+ if v3.LeaseID(kv.Lease) == lkv.leaseID() {
+ // don't revoke own keys
+ continue
+ }
+ key := strings.TrimPrefix(string(kv.Key), lkv.pfx)
+ if _, err := lkv.revoke(ctx, key, v3.OpGet(key)); err != nil {
+ return 0, err
+ }
+ }
+ return maxLeaseRev, nil
+}
+
+func (lkv *leasingKV) waitSession(ctx context.Context) error {
+ lkv.leases.mu.RLock()
+ sessionc := lkv.sessionc
+ lkv.leases.mu.RUnlock()
+ select {
+ case <-sessionc:
+ return nil
+ case <-lkv.ctx.Done():
+ return lkv.ctx.Err()
+ case <-ctx.Done():
+ return ctx.Err()
+ }
+}
+
+func (lkv *leasingKV) readySession() bool {
+ lkv.leases.mu.RLock()
+ defer lkv.leases.mu.RUnlock()
+ if lkv.session == nil {
+ return false
+ }
+ select {
+ case <-lkv.session.Done():
+ default:
+ return true
+ }
+ return false
+}
+
+func (lkv *leasingKV) leaseID() v3.LeaseID {
+ lkv.leases.mu.RLock()
+ defer lkv.leases.mu.RUnlock()
+ return lkv.session.Lease()
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/leasing/txn.go b/vendor/github.com/coreos/etcd/clientv3/leasing/txn.go
new file mode 100644
index 0000000..da5b83a
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/leasing/txn.go
@@ -0,0 +1,223 @@
+// Copyright 2017 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 leasing
+
+import (
+ "context"
+ "strings"
+
+ v3 "github.com/coreos/etcd/clientv3"
+ v3pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+type txnLeasing struct {
+ v3.Txn
+ lkv *leasingKV
+ ctx context.Context
+ cs []v3.Cmp
+ opst []v3.Op
+ opse []v3.Op
+}
+
+func (txn *txnLeasing) If(cs ...v3.Cmp) v3.Txn {
+ txn.cs = append(txn.cs, cs...)
+ txn.Txn = txn.Txn.If(cs...)
+ return txn
+}
+
+func (txn *txnLeasing) Then(ops ...v3.Op) v3.Txn {
+ txn.opst = append(txn.opst, ops...)
+ txn.Txn = txn.Txn.Then(ops...)
+ return txn
+}
+
+func (txn *txnLeasing) Else(ops ...v3.Op) v3.Txn {
+ txn.opse = append(txn.opse, ops...)
+ txn.Txn = txn.Txn.Else(ops...)
+ return txn
+}
+
+func (txn *txnLeasing) Commit() (*v3.TxnResponse, error) {
+ if resp, err := txn.eval(); resp != nil || err != nil {
+ return resp, err
+ }
+ return txn.serverTxn()
+}
+
+func (txn *txnLeasing) eval() (*v3.TxnResponse, error) {
+ // TODO: wait on keys in comparisons
+ thenOps, elseOps := gatherOps(txn.opst), gatherOps(txn.opse)
+ ops := make([]v3.Op, 0, len(thenOps)+len(elseOps))
+ ops = append(ops, thenOps...)
+ ops = append(ops, elseOps...)
+
+ for _, ch := range txn.lkv.leases.NotifyOps(ops) {
+ select {
+ case <-ch:
+ case <-txn.ctx.Done():
+ return nil, txn.ctx.Err()
+ }
+ }
+
+ txn.lkv.leases.mu.RLock()
+ defer txn.lkv.leases.mu.RUnlock()
+ succeeded, ok := txn.lkv.leases.evalCmp(txn.cs)
+ if !ok || txn.lkv.leases.header == nil {
+ return nil, nil
+ }
+ if ops = txn.opst; !succeeded {
+ ops = txn.opse
+ }
+
+ resps, ok := txn.lkv.leases.evalOps(ops)
+ if !ok {
+ return nil, nil
+ }
+ return &v3.TxnResponse{copyHeader(txn.lkv.leases.header), succeeded, resps}, nil
+}
+
+// fallback computes the ops to fetch all possible conflicting
+// leasing keys for a list of ops.
+func (txn *txnLeasing) fallback(ops []v3.Op) (fbOps []v3.Op) {
+ for _, op := range ops {
+ if op.IsGet() {
+ continue
+ }
+ lkey, lend := txn.lkv.pfx+string(op.KeyBytes()), ""
+ if len(op.RangeBytes()) > 0 {
+ lend = txn.lkv.pfx + string(op.RangeBytes())
+ }
+ fbOps = append(fbOps, v3.OpGet(lkey, v3.WithRange(lend)))
+ }
+ return fbOps
+}
+
+func (txn *txnLeasing) guardKeys(ops []v3.Op) (cmps []v3.Cmp) {
+ seen := make(map[string]bool)
+ for _, op := range ops {
+ key := string(op.KeyBytes())
+ if op.IsGet() || len(op.RangeBytes()) != 0 || seen[key] {
+ continue
+ }
+ rev := txn.lkv.leases.Rev(key)
+ cmps = append(cmps, v3.Compare(v3.CreateRevision(txn.lkv.pfx+key), "<", rev+1))
+ seen[key] = true
+ }
+ return cmps
+}
+
+func (txn *txnLeasing) guardRanges(ops []v3.Op) (cmps []v3.Cmp, err error) {
+ for _, op := range ops {
+ if op.IsGet() || len(op.RangeBytes()) == 0 {
+ continue
+ }
+
+ key, end := string(op.KeyBytes()), string(op.RangeBytes())
+ maxRevLK, err := txn.lkv.revokeRange(txn.ctx, key, end)
+ if err != nil {
+ return nil, err
+ }
+
+ opts := append(v3.WithLastRev(), v3.WithRange(end))
+ getResp, err := txn.lkv.kv.Get(txn.ctx, key, opts...)
+ if err != nil {
+ return nil, err
+ }
+ maxModRev := int64(0)
+ if len(getResp.Kvs) > 0 {
+ maxModRev = getResp.Kvs[0].ModRevision
+ }
+
+ noKeyUpdate := v3.Compare(v3.ModRevision(key).WithRange(end), "<", maxModRev+1)
+ noLeaseUpdate := v3.Compare(
+ v3.CreateRevision(txn.lkv.pfx+key).WithRange(txn.lkv.pfx+end),
+ "<",
+ maxRevLK+1)
+ cmps = append(cmps, noKeyUpdate, noLeaseUpdate)
+ }
+ return cmps, nil
+}
+
+func (txn *txnLeasing) guard(ops []v3.Op) ([]v3.Cmp, error) {
+ cmps := txn.guardKeys(ops)
+ rangeCmps, err := txn.guardRanges(ops)
+ return append(cmps, rangeCmps...), err
+}
+
+func (txn *txnLeasing) commitToCache(txnResp *v3pb.TxnResponse, userTxn v3.Op) {
+ ops := gatherResponseOps(txnResp.Responses, []v3.Op{userTxn})
+ txn.lkv.leases.mu.Lock()
+ for _, op := range ops {
+ key := string(op.KeyBytes())
+ if op.IsDelete() && len(op.RangeBytes()) > 0 {
+ end := string(op.RangeBytes())
+ for k := range txn.lkv.leases.entries {
+ if inRange(k, key, end) {
+ txn.lkv.leases.delete(k, txnResp.Header)
+ }
+ }
+ } else if op.IsDelete() {
+ txn.lkv.leases.delete(key, txnResp.Header)
+ }
+ if op.IsPut() {
+ txn.lkv.leases.Update(op.KeyBytes(), op.ValueBytes(), txnResp.Header)
+ }
+ }
+ txn.lkv.leases.mu.Unlock()
+}
+
+func (txn *txnLeasing) revokeFallback(fbResps []*v3pb.ResponseOp) error {
+ for _, resp := range fbResps {
+ _, err := txn.lkv.revokeLeaseKvs(txn.ctx, resp.GetResponseRange().Kvs)
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (txn *txnLeasing) serverTxn() (*v3.TxnResponse, error) {
+ if err := txn.lkv.waitSession(txn.ctx); err != nil {
+ return nil, err
+ }
+
+ userOps := gatherOps(append(txn.opst, txn.opse...))
+ userTxn := v3.OpTxn(txn.cs, txn.opst, txn.opse)
+ fbOps := txn.fallback(userOps)
+
+ defer closeAll(txn.lkv.leases.LockWriteOps(userOps))
+ for {
+ cmps, err := txn.guard(userOps)
+ if err != nil {
+ return nil, err
+ }
+ resp, err := txn.lkv.kv.Txn(txn.ctx).If(cmps...).Then(userTxn).Else(fbOps...).Commit()
+ if err != nil {
+ for _, cmp := range cmps {
+ txn.lkv.leases.Evict(strings.TrimPrefix(string(cmp.Key), txn.lkv.pfx))
+ }
+ return nil, err
+ }
+ if resp.Succeeded {
+ txn.commitToCache((*v3pb.TxnResponse)(resp), userTxn)
+ userResp := resp.Responses[0].GetResponseTxn()
+ userResp.Header = resp.Header
+ return (*v3.TxnResponse)(userResp), nil
+ }
+ if err := txn.revokeFallback(resp.Responses); err != nil {
+ return nil, err
+ }
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/leasing/util.go b/vendor/github.com/coreos/etcd/clientv3/leasing/util.go
new file mode 100644
index 0000000..61f6e8c
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/leasing/util.go
@@ -0,0 +1,108 @@
+// Copyright 2017 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 leasing
+
+import (
+ "bytes"
+
+ v3 "github.com/coreos/etcd/clientv3"
+ v3pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+func compareInt64(a, b int64) int {
+ switch {
+ case a < b:
+ return -1
+ case a > b:
+ return 1
+ default:
+ return 0
+ }
+}
+
+func evalCmp(resp *v3.GetResponse, tcmp v3.Cmp) bool {
+ var result int
+ if len(resp.Kvs) != 0 {
+ kv := resp.Kvs[0]
+ switch tcmp.Target {
+ case v3pb.Compare_VALUE:
+ if tv, _ := tcmp.TargetUnion.(*v3pb.Compare_Value); tv != nil {
+ result = bytes.Compare(kv.Value, tv.Value)
+ }
+ case v3pb.Compare_CREATE:
+ if tv, _ := tcmp.TargetUnion.(*v3pb.Compare_CreateRevision); tv != nil {
+ result = compareInt64(kv.CreateRevision, tv.CreateRevision)
+ }
+ case v3pb.Compare_MOD:
+ if tv, _ := tcmp.TargetUnion.(*v3pb.Compare_ModRevision); tv != nil {
+ result = compareInt64(kv.ModRevision, tv.ModRevision)
+ }
+ case v3pb.Compare_VERSION:
+ if tv, _ := tcmp.TargetUnion.(*v3pb.Compare_Version); tv != nil {
+ result = compareInt64(kv.Version, tv.Version)
+ }
+ }
+ }
+ switch tcmp.Result {
+ case v3pb.Compare_EQUAL:
+ return result == 0
+ case v3pb.Compare_NOT_EQUAL:
+ return result != 0
+ case v3pb.Compare_GREATER:
+ return result > 0
+ case v3pb.Compare_LESS:
+ return result < 0
+ }
+ return true
+}
+
+func gatherOps(ops []v3.Op) (ret []v3.Op) {
+ for _, op := range ops {
+ if !op.IsTxn() {
+ ret = append(ret, op)
+ continue
+ }
+ _, thenOps, elseOps := op.Txn()
+ ret = append(ret, gatherOps(append(thenOps, elseOps...))...)
+ }
+ return ret
+}
+
+func gatherResponseOps(resp []*v3pb.ResponseOp, ops []v3.Op) (ret []v3.Op) {
+ for i, op := range ops {
+ if !op.IsTxn() {
+ ret = append(ret, op)
+ continue
+ }
+ _, thenOps, elseOps := op.Txn()
+ if txnResp := resp[i].GetResponseTxn(); txnResp.Succeeded {
+ ret = append(ret, gatherResponseOps(txnResp.Responses, thenOps)...)
+ } else {
+ ret = append(ret, gatherResponseOps(txnResp.Responses, elseOps)...)
+ }
+ }
+ return ret
+}
+
+func copyHeader(hdr *v3pb.ResponseHeader) *v3pb.ResponseHeader {
+ h := *hdr
+ return &h
+}
+
+func closeAll(chs []chan<- struct{}) {
+ for _, ch := range chs {
+ close(ch)
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/logger.go b/vendor/github.com/coreos/etcd/clientv3/logger.go
new file mode 100644
index 0000000..782e313
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/logger.go
@@ -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 clientv3
+
+import (
+ "io/ioutil"
+ "sync"
+
+ "google.golang.org/grpc/grpclog"
+)
+
+// Logger is the logger used by client library.
+// It implements grpclog.LoggerV2 interface.
+type Logger interface {
+ grpclog.LoggerV2
+
+ // Lvl returns logger if logger's verbosity level >= "lvl".
+ // Otherwise, logger that discards all logs.
+ Lvl(lvl int) Logger
+
+ // to satisfy capnslog
+
+ Print(args ...interface{})
+ Printf(format string, args ...interface{})
+ Println(args ...interface{})
+}
+
+var (
+ loggerMu sync.RWMutex
+ logger Logger
+)
+
+type settableLogger struct {
+ l grpclog.LoggerV2
+ mu sync.RWMutex
+}
+
+func init() {
+ // disable client side logs by default
+ logger = &settableLogger{}
+ SetLogger(grpclog.NewLoggerV2(ioutil.Discard, ioutil.Discard, ioutil.Discard))
+}
+
+// SetLogger sets client-side Logger.
+func SetLogger(l grpclog.LoggerV2) {
+ loggerMu.Lock()
+ logger = NewLogger(l)
+ // override grpclog so that any changes happen with locking
+ grpclog.SetLoggerV2(logger)
+ loggerMu.Unlock()
+}
+
+// GetLogger returns the current logger.
+func GetLogger() Logger {
+ loggerMu.RLock()
+ l := logger
+ loggerMu.RUnlock()
+ return l
+}
+
+// NewLogger returns a new Logger with grpclog.LoggerV2.
+func NewLogger(gl grpclog.LoggerV2) Logger {
+ return &settableLogger{l: gl}
+}
+
+func (s *settableLogger) get() grpclog.LoggerV2 {
+ s.mu.RLock()
+ l := s.l
+ s.mu.RUnlock()
+ return l
+}
+
+// implement the grpclog.LoggerV2 interface
+
+func (s *settableLogger) Info(args ...interface{}) { s.get().Info(args...) }
+func (s *settableLogger) Infof(format string, args ...interface{}) { s.get().Infof(format, args...) }
+func (s *settableLogger) Infoln(args ...interface{}) { s.get().Infoln(args...) }
+func (s *settableLogger) Warning(args ...interface{}) { s.get().Warning(args...) }
+func (s *settableLogger) Warningf(format string, args ...interface{}) {
+ s.get().Warningf(format, args...)
+}
+func (s *settableLogger) Warningln(args ...interface{}) { s.get().Warningln(args...) }
+func (s *settableLogger) Error(args ...interface{}) { s.get().Error(args...) }
+func (s *settableLogger) Errorf(format string, args ...interface{}) {
+ s.get().Errorf(format, args...)
+}
+func (s *settableLogger) Errorln(args ...interface{}) { s.get().Errorln(args...) }
+func (s *settableLogger) Fatal(args ...interface{}) { s.get().Fatal(args...) }
+func (s *settableLogger) Fatalf(format string, args ...interface{}) { s.get().Fatalf(format, args...) }
+func (s *settableLogger) Fatalln(args ...interface{}) { s.get().Fatalln(args...) }
+func (s *settableLogger) Print(args ...interface{}) { s.get().Info(args...) }
+func (s *settableLogger) Printf(format string, args ...interface{}) { s.get().Infof(format, args...) }
+func (s *settableLogger) Println(args ...interface{}) { s.get().Infoln(args...) }
+func (s *settableLogger) V(l int) bool { return s.get().V(l) }
+func (s *settableLogger) Lvl(lvl int) Logger {
+ s.mu.RLock()
+ l := s.l
+ s.mu.RUnlock()
+ if l.V(lvl) {
+ return s
+ }
+ return &noLogger{}
+}
+
+type noLogger struct{}
+
+func (*noLogger) Info(args ...interface{}) {}
+func (*noLogger) Infof(format string, args ...interface{}) {}
+func (*noLogger) Infoln(args ...interface{}) {}
+func (*noLogger) Warning(args ...interface{}) {}
+func (*noLogger) Warningf(format string, args ...interface{}) {}
+func (*noLogger) Warningln(args ...interface{}) {}
+func (*noLogger) Error(args ...interface{}) {}
+func (*noLogger) Errorf(format string, args ...interface{}) {}
+func (*noLogger) Errorln(args ...interface{}) {}
+func (*noLogger) Fatal(args ...interface{}) {}
+func (*noLogger) Fatalf(format string, args ...interface{}) {}
+func (*noLogger) Fatalln(args ...interface{}) {}
+func (*noLogger) Print(args ...interface{}) {}
+func (*noLogger) Printf(format string, args ...interface{}) {}
+func (*noLogger) Println(args ...interface{}) {}
+func (*noLogger) V(l int) bool { return false }
+func (ng *noLogger) Lvl(lvl int) Logger { return ng }
diff --git a/vendor/github.com/coreos/etcd/clientv3/maintenance.go b/vendor/github.com/coreos/etcd/clientv3/maintenance.go
new file mode 100644
index 0000000..f60cfbe
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/maintenance.go
@@ -0,0 +1,226 @@
+// 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 clientv3
+
+import (
+ "context"
+ "io"
+
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+ "google.golang.org/grpc"
+)
+
+type (
+ DefragmentResponse pb.DefragmentResponse
+ AlarmResponse pb.AlarmResponse
+ AlarmMember pb.AlarmMember
+ StatusResponse pb.StatusResponse
+ HashKVResponse pb.HashKVResponse
+ MoveLeaderResponse pb.MoveLeaderResponse
+)
+
+type Maintenance interface {
+ // AlarmList gets all active alarms.
+ AlarmList(ctx context.Context) (*AlarmResponse, error)
+
+ // AlarmDisarm disarms a given alarm.
+ AlarmDisarm(ctx context.Context, m *AlarmMember) (*AlarmResponse, error)
+
+ // Defragment releases wasted space from internal fragmentation on a given etcd member.
+ // Defragment is only needed when deleting a large number of keys and want to reclaim
+ // the resources.
+ // Defragment is an expensive operation. User should avoid defragmenting multiple members
+ // at the same time.
+ // To defragment multiple members in the cluster, user need to call defragment multiple
+ // times with different endpoints.
+ Defragment(ctx context.Context, endpoint string) (*DefragmentResponse, error)
+
+ // Status gets the status of the endpoint.
+ Status(ctx context.Context, endpoint string) (*StatusResponse, error)
+
+ // HashKV returns a hash of the KV state at the time of the RPC.
+ // If revision is zero, the hash is computed on all keys. If the revision
+ // is non-zero, the hash is computed on all keys at or below the given revision.
+ HashKV(ctx context.Context, endpoint string, rev int64) (*HashKVResponse, error)
+
+ // Snapshot provides a reader for a point-in-time snapshot of etcd.
+ Snapshot(ctx context.Context) (io.ReadCloser, error)
+
+ // MoveLeader requests current leader to transfer its leadership to the transferee.
+ // Request must be made to the leader.
+ MoveLeader(ctx context.Context, transfereeID uint64) (*MoveLeaderResponse, error)
+}
+
+type maintenance struct {
+ dial func(endpoint string) (pb.MaintenanceClient, func(), error)
+ remote pb.MaintenanceClient
+ callOpts []grpc.CallOption
+}
+
+func NewMaintenance(c *Client) Maintenance {
+ api := &maintenance{
+ dial: func(endpoint string) (pb.MaintenanceClient, func(), error) {
+ conn, err := c.dial(endpoint)
+ if err != nil {
+ return nil, nil, err
+ }
+ cancel := func() { conn.Close() }
+ return RetryMaintenanceClient(c, conn), cancel, nil
+ },
+ remote: RetryMaintenanceClient(c, c.conn),
+ }
+ if c != nil {
+ api.callOpts = c.callOpts
+ }
+ return api
+}
+
+func NewMaintenanceFromMaintenanceClient(remote pb.MaintenanceClient, c *Client) Maintenance {
+ api := &maintenance{
+ dial: func(string) (pb.MaintenanceClient, func(), error) {
+ return remote, func() {}, nil
+ },
+ remote: remote,
+ }
+ if c != nil {
+ api.callOpts = c.callOpts
+ }
+ return api
+}
+
+func (m *maintenance) AlarmList(ctx context.Context) (*AlarmResponse, error) {
+ req := &pb.AlarmRequest{
+ Action: pb.AlarmRequest_GET,
+ MemberID: 0, // all
+ Alarm: pb.AlarmType_NONE, // all
+ }
+ resp, err := m.remote.Alarm(ctx, req, m.callOpts...)
+ if err == nil {
+ return (*AlarmResponse)(resp), nil
+ }
+ return nil, toErr(ctx, err)
+}
+
+func (m *maintenance) AlarmDisarm(ctx context.Context, am *AlarmMember) (*AlarmResponse, error) {
+ req := &pb.AlarmRequest{
+ Action: pb.AlarmRequest_DEACTIVATE,
+ MemberID: am.MemberID,
+ Alarm: am.Alarm,
+ }
+
+ if req.MemberID == 0 && req.Alarm == pb.AlarmType_NONE {
+ ar, err := m.AlarmList(ctx)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ ret := AlarmResponse{}
+ for _, am := range ar.Alarms {
+ dresp, derr := m.AlarmDisarm(ctx, (*AlarmMember)(am))
+ if derr != nil {
+ return nil, toErr(ctx, derr)
+ }
+ ret.Alarms = append(ret.Alarms, dresp.Alarms...)
+ }
+ return &ret, nil
+ }
+
+ resp, err := m.remote.Alarm(ctx, req, m.callOpts...)
+ if err == nil {
+ return (*AlarmResponse)(resp), nil
+ }
+ return nil, toErr(ctx, err)
+}
+
+func (m *maintenance) Defragment(ctx context.Context, endpoint string) (*DefragmentResponse, error) {
+ remote, cancel, err := m.dial(endpoint)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ defer cancel()
+ resp, err := remote.Defragment(ctx, &pb.DefragmentRequest{}, m.callOpts...)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ return (*DefragmentResponse)(resp), nil
+}
+
+func (m *maintenance) Status(ctx context.Context, endpoint string) (*StatusResponse, error) {
+ remote, cancel, err := m.dial(endpoint)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ defer cancel()
+ resp, err := remote.Status(ctx, &pb.StatusRequest{}, m.callOpts...)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ return (*StatusResponse)(resp), nil
+}
+
+func (m *maintenance) HashKV(ctx context.Context, endpoint string, rev int64) (*HashKVResponse, error) {
+ remote, cancel, err := m.dial(endpoint)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ defer cancel()
+ resp, err := remote.HashKV(ctx, &pb.HashKVRequest{Revision: rev}, m.callOpts...)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+ return (*HashKVResponse)(resp), nil
+}
+
+func (m *maintenance) Snapshot(ctx context.Context) (io.ReadCloser, error) {
+ ss, err := m.remote.Snapshot(ctx, &pb.SnapshotRequest{}, m.callOpts...)
+ if err != nil {
+ return nil, toErr(ctx, err)
+ }
+
+ pr, pw := io.Pipe()
+ go func() {
+ for {
+ resp, err := ss.Recv()
+ if err != nil {
+ pw.CloseWithError(err)
+ return
+ }
+ if resp == nil && err == nil {
+ break
+ }
+ if _, werr := pw.Write(resp.Blob); werr != nil {
+ pw.CloseWithError(werr)
+ return
+ }
+ }
+ pw.Close()
+ }()
+ return &snapshotReadCloser{ctx: ctx, ReadCloser: pr}, nil
+}
+
+type snapshotReadCloser struct {
+ ctx context.Context
+ io.ReadCloser
+}
+
+func (rc *snapshotReadCloser) Read(p []byte) (n int, err error) {
+ n, err = rc.ReadCloser.Read(p)
+ return n, toErr(rc.ctx, err)
+}
+
+func (m *maintenance) MoveLeader(ctx context.Context, transfereeID uint64) (*MoveLeaderResponse, error) {
+ resp, err := m.remote.MoveLeader(ctx, &pb.MoveLeaderRequest{TargetID: transfereeID}, m.callOpts...)
+ return (*MoveLeaderResponse)(resp), toErr(ctx, err)
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/namespace/doc.go b/vendor/github.com/coreos/etcd/clientv3/namespace/doc.go
new file mode 100644
index 0000000..3f88332
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/namespace/doc.go
@@ -0,0 +1,43 @@
+// Copyright 2017 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 namespace is a clientv3 wrapper that translates all keys to begin
+// with a given prefix.
+//
+// First, create a client:
+//
+// cli, err := clientv3.New(clientv3.Config{Endpoints: []string{"localhost:2379"}})
+// if err != nil {
+// // handle error!
+// }
+//
+// Next, override the client interfaces:
+//
+// unprefixedKV := cli.KV
+// cli.KV = namespace.NewKV(cli.KV, "my-prefix/")
+// cli.Watcher = namespace.NewWatcher(cli.Watcher, "my-prefix/")
+// cli.Lease = namespace.NewLease(cli.Lease, "my-prefix/")
+//
+// Now calls using 'cli' will namespace / prefix all keys with "my-prefix/":
+//
+// cli.Put(context.TODO(), "abc", "123")
+// resp, _ := unprefixedKV.Get(context.TODO(), "my-prefix/abc")
+// fmt.Printf("%s\n", resp.Kvs[0].Value)
+// // Output: 123
+// unprefixedKV.Put(context.TODO(), "my-prefix/abc", "456")
+// resp, _ = cli.Get("abc")
+// fmt.Printf("%s\n", resp.Kvs[0].Value)
+// // Output: 456
+//
+package namespace
diff --git a/vendor/github.com/coreos/etcd/clientv3/namespace/kv.go b/vendor/github.com/coreos/etcd/clientv3/namespace/kv.go
new file mode 100644
index 0000000..13dd83a
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/namespace/kv.go
@@ -0,0 +1,206 @@
+// Copyright 2017 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 namespace
+
+import (
+ "context"
+
+ "github.com/coreos/etcd/clientv3"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+)
+
+type kvPrefix struct {
+ clientv3.KV
+ pfx string
+}
+
+// NewKV wraps a KV instance so that all requests
+// are prefixed with a given string.
+func NewKV(kv clientv3.KV, prefix string) clientv3.KV {
+ return &kvPrefix{kv, prefix}
+}
+
+func (kv *kvPrefix) Put(ctx context.Context, key, val string, opts ...clientv3.OpOption) (*clientv3.PutResponse, error) {
+ if len(key) == 0 {
+ return nil, rpctypes.ErrEmptyKey
+ }
+ op := kv.prefixOp(clientv3.OpPut(key, val, opts...))
+ r, err := kv.KV.Do(ctx, op)
+ if err != nil {
+ return nil, err
+ }
+ put := r.Put()
+ kv.unprefixPutResponse(put)
+ return put, nil
+}
+
+func (kv *kvPrefix) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
+ if len(key) == 0 {
+ return nil, rpctypes.ErrEmptyKey
+ }
+ r, err := kv.KV.Do(ctx, kv.prefixOp(clientv3.OpGet(key, opts...)))
+ if err != nil {
+ return nil, err
+ }
+ get := r.Get()
+ kv.unprefixGetResponse(get)
+ return get, nil
+}
+
+func (kv *kvPrefix) Delete(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.DeleteResponse, error) {
+ if len(key) == 0 {
+ return nil, rpctypes.ErrEmptyKey
+ }
+ r, err := kv.KV.Do(ctx, kv.prefixOp(clientv3.OpDelete(key, opts...)))
+ if err != nil {
+ return nil, err
+ }
+ del := r.Del()
+ kv.unprefixDeleteResponse(del)
+ return del, nil
+}
+
+func (kv *kvPrefix) Do(ctx context.Context, op clientv3.Op) (clientv3.OpResponse, error) {
+ if len(op.KeyBytes()) == 0 && !op.IsTxn() {
+ return clientv3.OpResponse{}, rpctypes.ErrEmptyKey
+ }
+ r, err := kv.KV.Do(ctx, kv.prefixOp(op))
+ if err != nil {
+ return r, err
+ }
+ switch {
+ case r.Get() != nil:
+ kv.unprefixGetResponse(r.Get())
+ case r.Put() != nil:
+ kv.unprefixPutResponse(r.Put())
+ case r.Del() != nil:
+ kv.unprefixDeleteResponse(r.Del())
+ case r.Txn() != nil:
+ kv.unprefixTxnResponse(r.Txn())
+ }
+ return r, nil
+}
+
+type txnPrefix struct {
+ clientv3.Txn
+ kv *kvPrefix
+}
+
+func (kv *kvPrefix) Txn(ctx context.Context) clientv3.Txn {
+ return &txnPrefix{kv.KV.Txn(ctx), kv}
+}
+
+func (txn *txnPrefix) If(cs ...clientv3.Cmp) clientv3.Txn {
+ txn.Txn = txn.Txn.If(txn.kv.prefixCmps(cs)...)
+ return txn
+}
+
+func (txn *txnPrefix) Then(ops ...clientv3.Op) clientv3.Txn {
+ txn.Txn = txn.Txn.Then(txn.kv.prefixOps(ops)...)
+ return txn
+}
+
+func (txn *txnPrefix) Else(ops ...clientv3.Op) clientv3.Txn {
+ txn.Txn = txn.Txn.Else(txn.kv.prefixOps(ops)...)
+ return txn
+}
+
+func (txn *txnPrefix) Commit() (*clientv3.TxnResponse, error) {
+ resp, err := txn.Txn.Commit()
+ if err != nil {
+ return nil, err
+ }
+ txn.kv.unprefixTxnResponse(resp)
+ return resp, nil
+}
+
+func (kv *kvPrefix) prefixOp(op clientv3.Op) clientv3.Op {
+ if !op.IsTxn() {
+ begin, end := kv.prefixInterval(op.KeyBytes(), op.RangeBytes())
+ op.WithKeyBytes(begin)
+ op.WithRangeBytes(end)
+ return op
+ }
+ cmps, thenOps, elseOps := op.Txn()
+ return clientv3.OpTxn(kv.prefixCmps(cmps), kv.prefixOps(thenOps), kv.prefixOps(elseOps))
+}
+
+func (kv *kvPrefix) unprefixGetResponse(resp *clientv3.GetResponse) {
+ for i := range resp.Kvs {
+ resp.Kvs[i].Key = resp.Kvs[i].Key[len(kv.pfx):]
+ }
+}
+
+func (kv *kvPrefix) unprefixPutResponse(resp *clientv3.PutResponse) {
+ if resp.PrevKv != nil {
+ resp.PrevKv.Key = resp.PrevKv.Key[len(kv.pfx):]
+ }
+}
+
+func (kv *kvPrefix) unprefixDeleteResponse(resp *clientv3.DeleteResponse) {
+ for i := range resp.PrevKvs {
+ resp.PrevKvs[i].Key = resp.PrevKvs[i].Key[len(kv.pfx):]
+ }
+}
+
+func (kv *kvPrefix) unprefixTxnResponse(resp *clientv3.TxnResponse) {
+ for _, r := range resp.Responses {
+ switch tv := r.Response.(type) {
+ case *pb.ResponseOp_ResponseRange:
+ if tv.ResponseRange != nil {
+ kv.unprefixGetResponse((*clientv3.GetResponse)(tv.ResponseRange))
+ }
+ case *pb.ResponseOp_ResponsePut:
+ if tv.ResponsePut != nil {
+ kv.unprefixPutResponse((*clientv3.PutResponse)(tv.ResponsePut))
+ }
+ case *pb.ResponseOp_ResponseDeleteRange:
+ if tv.ResponseDeleteRange != nil {
+ kv.unprefixDeleteResponse((*clientv3.DeleteResponse)(tv.ResponseDeleteRange))
+ }
+ case *pb.ResponseOp_ResponseTxn:
+ if tv.ResponseTxn != nil {
+ kv.unprefixTxnResponse((*clientv3.TxnResponse)(tv.ResponseTxn))
+ }
+ default:
+ }
+ }
+}
+
+func (kv *kvPrefix) prefixInterval(key, end []byte) (pfxKey []byte, pfxEnd []byte) {
+ return prefixInterval(kv.pfx, key, end)
+}
+
+func (kv *kvPrefix) prefixCmps(cs []clientv3.Cmp) []clientv3.Cmp {
+ newCmps := make([]clientv3.Cmp, len(cs))
+ for i := range cs {
+ newCmps[i] = cs[i]
+ pfxKey, endKey := kv.prefixInterval(cs[i].KeyBytes(), cs[i].RangeEnd)
+ newCmps[i].WithKeyBytes(pfxKey)
+ if len(cs[i].RangeEnd) != 0 {
+ newCmps[i].RangeEnd = endKey
+ }
+ }
+ return newCmps
+}
+
+func (kv *kvPrefix) prefixOps(ops []clientv3.Op) []clientv3.Op {
+ newOps := make([]clientv3.Op, len(ops))
+ for i := range ops {
+ newOps[i] = kv.prefixOp(ops[i])
+ }
+ return newOps
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/namespace/lease.go b/vendor/github.com/coreos/etcd/clientv3/namespace/lease.go
new file mode 100644
index 0000000..f092106
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/namespace/lease.go
@@ -0,0 +1,57 @@
+// Copyright 2017 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 namespace
+
+import (
+ "bytes"
+ "context"
+
+ "github.com/coreos/etcd/clientv3"
+)
+
+type leasePrefix struct {
+ clientv3.Lease
+ pfx []byte
+}
+
+// NewLease wraps a Lease interface to filter for only keys with a prefix
+// and remove that prefix when fetching attached keys through TimeToLive.
+func NewLease(l clientv3.Lease, prefix string) clientv3.Lease {
+ return &leasePrefix{l, []byte(prefix)}
+}
+
+func (l *leasePrefix) TimeToLive(ctx context.Context, id clientv3.LeaseID, opts ...clientv3.LeaseOption) (*clientv3.LeaseTimeToLiveResponse, error) {
+ resp, err := l.Lease.TimeToLive(ctx, id, opts...)
+ if err != nil {
+ return nil, err
+ }
+ if len(resp.Keys) > 0 {
+ var outKeys [][]byte
+ for i := range resp.Keys {
+ if len(resp.Keys[i]) < len(l.pfx) {
+ // too short
+ continue
+ }
+ if !bytes.Equal(resp.Keys[i][:len(l.pfx)], l.pfx) {
+ // doesn't match prefix
+ continue
+ }
+ // strip prefix
+ outKeys = append(outKeys, resp.Keys[i][len(l.pfx):])
+ }
+ resp.Keys = outKeys
+ }
+ return resp, nil
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/namespace/util.go b/vendor/github.com/coreos/etcd/clientv3/namespace/util.go
new file mode 100644
index 0000000..ecf0404
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/namespace/util.go
@@ -0,0 +1,42 @@
+// Copyright 2017 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 namespace
+
+func prefixInterval(pfx string, key, end []byte) (pfxKey []byte, pfxEnd []byte) {
+ pfxKey = make([]byte, len(pfx)+len(key))
+ copy(pfxKey[copy(pfxKey, pfx):], key)
+
+ if len(end) == 1 && end[0] == 0 {
+ // the edge of the keyspace
+ pfxEnd = make([]byte, len(pfx))
+ copy(pfxEnd, pfx)
+ ok := false
+ for i := len(pfxEnd) - 1; i >= 0; i-- {
+ if pfxEnd[i]++; pfxEnd[i] != 0 {
+ ok = true
+ break
+ }
+ }
+ if !ok {
+ // 0xff..ff => 0x00
+ pfxEnd = []byte{0}
+ }
+ } else if len(end) >= 1 {
+ pfxEnd = make([]byte, len(pfx)+len(end))
+ copy(pfxEnd[copy(pfxEnd, pfx):], end)
+ }
+
+ return pfxKey, pfxEnd
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/namespace/watch.go b/vendor/github.com/coreos/etcd/clientv3/namespace/watch.go
new file mode 100644
index 0000000..5a9596d
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/namespace/watch.go
@@ -0,0 +1,83 @@
+// Copyright 2017 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 namespace
+
+import (
+ "context"
+ "sync"
+
+ "github.com/coreos/etcd/clientv3"
+)
+
+type watcherPrefix struct {
+ clientv3.Watcher
+ pfx string
+
+ wg sync.WaitGroup
+ stopc chan struct{}
+ stopOnce sync.Once
+}
+
+// NewWatcher wraps a Watcher instance so that all Watch requests
+// are prefixed with a given string and all Watch responses have
+// the prefix removed.
+func NewWatcher(w clientv3.Watcher, prefix string) clientv3.Watcher {
+ return &watcherPrefix{Watcher: w, pfx: prefix, stopc: make(chan struct{})}
+}
+
+func (w *watcherPrefix) Watch(ctx context.Context, key string, opts ...clientv3.OpOption) clientv3.WatchChan {
+ // since OpOption is opaque, determine range for prefixing through an OpGet
+ op := clientv3.OpGet(key, opts...)
+ end := op.RangeBytes()
+ pfxBegin, pfxEnd := prefixInterval(w.pfx, []byte(key), end)
+ if pfxEnd != nil {
+ opts = append(opts, clientv3.WithRange(string(pfxEnd)))
+ }
+
+ wch := w.Watcher.Watch(ctx, string(pfxBegin), opts...)
+
+ // translate watch events from prefixed to unprefixed
+ pfxWch := make(chan clientv3.WatchResponse)
+ w.wg.Add(1)
+ go func() {
+ defer func() {
+ close(pfxWch)
+ w.wg.Done()
+ }()
+ for wr := range wch {
+ for i := range wr.Events {
+ wr.Events[i].Kv.Key = wr.Events[i].Kv.Key[len(w.pfx):]
+ if wr.Events[i].PrevKv != nil {
+ wr.Events[i].PrevKv.Key = wr.Events[i].Kv.Key
+ }
+ }
+ select {
+ case pfxWch <- wr:
+ case <-ctx.Done():
+ return
+ case <-w.stopc:
+ return
+ }
+ }
+ }()
+ return pfxWch
+}
+
+func (w *watcherPrefix) Close() error {
+ err := w.Watcher.Close()
+ w.stopOnce.Do(func() { close(w.stopc) })
+ w.wg.Wait()
+ return err
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/naming/doc.go b/vendor/github.com/coreos/etcd/clientv3/naming/doc.go
new file mode 100644
index 0000000..71608cc
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/naming/doc.go
@@ -0,0 +1,56 @@
+// Copyright 2017 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 provides an etcd-backed gRPC resolver for discovering gRPC services.
+//
+// To use, first import the packages:
+//
+// import (
+// "github.com/coreos/etcd/clientv3"
+// etcdnaming "github.com/coreos/etcd/clientv3/naming"
+//
+// "google.golang.org/grpc"
+// "google.golang.org/grpc/naming"
+// )
+//
+// First, register new endpoint addresses for a service:
+//
+// func etcdAdd(c *clientv3.Client, service, addr string) error {
+// r := &etcdnaming.GRPCResolver{Client: c}
+// return r.Update(c.Ctx(), service, naming.Update{Op: naming.Add, Addr: addr})
+// }
+//
+// Dial an RPC service using the etcd gRPC resolver and a gRPC Balancer:
+//
+// func etcdDial(c *clientv3.Client, service string) (*grpc.ClientConn, error) {
+// r := &etcdnaming.GRPCResolver{Client: c}
+// b := grpc.RoundRobin(r)
+// return grpc.Dial(service, grpc.WithBalancer(b))
+// }
+//
+// Optionally, force delete an endpoint:
+//
+// func etcdDelete(c *clientv3, service, addr string) error {
+// r := &etcdnaming.GRPCResolver{Client: c}
+// return r.Update(c.Ctx(), "my-service", naming.Update{Op: naming.Delete, Addr: "1.2.3.4"})
+// }
+//
+// Or register an expiring endpoint with a lease:
+//
+// func etcdLeaseAdd(c *clientv3.Client, lid clientv3.LeaseID, service, addr string) error {
+// r := &etcdnaming.GRPCResolver{Client: c}
+// return r.Update(c.Ctx(), service, naming.Update{Op: naming.Add, Addr: addr}, clientv3.WithLease(lid))
+// }
+//
+package naming
diff --git a/vendor/github.com/coreos/etcd/clientv3/naming/grpc.go b/vendor/github.com/coreos/etcd/clientv3/naming/grpc.go
new file mode 100644
index 0000000..3c0e8e6
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/naming/grpc.go
@@ -0,0 +1,131 @@
+// 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"
+ "encoding/json"
+ "fmt"
+
+ etcd "github.com/coreos/etcd/clientv3"
+
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/naming"
+ "google.golang.org/grpc/status"
+)
+
+var ErrWatcherClosed = fmt.Errorf("naming: watch closed")
+
+// GRPCResolver creates a grpc.Watcher for a target to track its resolution changes.
+type GRPCResolver struct {
+ // Client is an initialized etcd client.
+ Client *etcd.Client
+}
+
+func (gr *GRPCResolver) Update(ctx context.Context, target string, nm naming.Update, opts ...etcd.OpOption) (err error) {
+ switch nm.Op {
+ case naming.Add:
+ var v []byte
+ if v, err = json.Marshal(nm); err != nil {
+ return status.Error(codes.InvalidArgument, err.Error())
+ }
+ _, err = gr.Client.KV.Put(ctx, target+"/"+nm.Addr, string(v), opts...)
+ case naming.Delete:
+ _, err = gr.Client.Delete(ctx, target+"/"+nm.Addr, opts...)
+ default:
+ return status.Error(codes.InvalidArgument, "naming: bad naming op")
+ }
+ return err
+}
+
+func (gr *GRPCResolver) Resolve(target string) (naming.Watcher, error) {
+ ctx, cancel := context.WithCancel(context.Background())
+ w := &gRPCWatcher{c: gr.Client, target: target + "/", ctx: ctx, cancel: cancel}
+ return w, nil
+}
+
+type gRPCWatcher struct {
+ c *etcd.Client
+ target string
+ ctx context.Context
+ cancel context.CancelFunc
+ wch etcd.WatchChan
+ err error
+}
+
+// Next gets the next set of updates from the etcd resolver.
+// Calls to Next should be serialized; concurrent calls are not safe since
+// there is no way to reconcile the update ordering.
+func (gw *gRPCWatcher) Next() ([]*naming.Update, error) {
+ if gw.wch == nil {
+ // first Next() returns all addresses
+ return gw.firstNext()
+ }
+ if gw.err != nil {
+ return nil, gw.err
+ }
+
+ // process new events on target/*
+ wr, ok := <-gw.wch
+ if !ok {
+ gw.err = status.Error(codes.Unavailable, ErrWatcherClosed.Error())
+ return nil, gw.err
+ }
+ if gw.err = wr.Err(); gw.err != nil {
+ return nil, gw.err
+ }
+
+ updates := make([]*naming.Update, 0, len(wr.Events))
+ for _, e := range wr.Events {
+ var jupdate naming.Update
+ var err error
+ switch e.Type {
+ case etcd.EventTypePut:
+ err = json.Unmarshal(e.Kv.Value, &jupdate)
+ jupdate.Op = naming.Add
+ case etcd.EventTypeDelete:
+ err = json.Unmarshal(e.PrevKv.Value, &jupdate)
+ jupdate.Op = naming.Delete
+ }
+ if err == nil {
+ updates = append(updates, &jupdate)
+ }
+ }
+ return updates, nil
+}
+
+func (gw *gRPCWatcher) firstNext() ([]*naming.Update, error) {
+ // Use serialized request so resolution still works if the target etcd
+ // server is partitioned away from the quorum.
+ resp, err := gw.c.Get(gw.ctx, gw.target, etcd.WithPrefix(), etcd.WithSerializable())
+ if gw.err = err; err != nil {
+ return nil, err
+ }
+
+ updates := make([]*naming.Update, 0, len(resp.Kvs))
+ for _, kv := range resp.Kvs {
+ var jupdate naming.Update
+ if err := json.Unmarshal(kv.Value, &jupdate); err != nil {
+ continue
+ }
+ updates = append(updates, &jupdate)
+ }
+
+ opts := []etcd.OpOption{etcd.WithRev(resp.Header.Revision + 1), etcd.WithPrefix(), etcd.WithPrevKV()}
+ gw.wch = gw.c.Watch(gw.ctx, gw.target, opts...)
+ return updates, nil
+}
+
+func (gw *gRPCWatcher) Close() { gw.cancel() }
diff --git a/vendor/github.com/coreos/etcd/clientv3/op.go b/vendor/github.com/coreos/etcd/clientv3/op.go
new file mode 100644
index 0000000..c6ec5bf
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/op.go
@@ -0,0 +1,513 @@
+// 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 clientv3
+
+import pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+type opType int
+
+const (
+ // A default Op has opType 0, which is invalid.
+ tRange opType = iota + 1
+ tPut
+ tDeleteRange
+ tTxn
+)
+
+var (
+ noPrefixEnd = []byte{0}
+)
+
+// Op represents an Operation that kv can execute.
+type Op struct {
+ t opType
+ key []byte
+ end []byte
+
+ // for range
+ limit int64
+ sort *SortOption
+ serializable bool
+ keysOnly bool
+ countOnly bool
+ minModRev int64
+ maxModRev int64
+ minCreateRev int64
+ maxCreateRev int64
+
+ // for range, watch
+ rev int64
+
+ // for watch, put, delete
+ prevKV bool
+
+ // for put
+ ignoreValue bool
+ ignoreLease bool
+
+ // progressNotify is for progress updates.
+ progressNotify bool
+ // createdNotify is for created event
+ createdNotify bool
+ // filters for watchers
+ filterPut bool
+ filterDelete bool
+
+ // for put
+ val []byte
+ leaseID LeaseID
+
+ // txn
+ cmps []Cmp
+ thenOps []Op
+ elseOps []Op
+}
+
+// accessors / mutators
+
+func (op Op) IsTxn() bool { return op.t == tTxn }
+func (op Op) Txn() ([]Cmp, []Op, []Op) { return op.cmps, op.thenOps, op.elseOps }
+
+// KeyBytes returns the byte slice holding the Op's key.
+func (op Op) KeyBytes() []byte { return op.key }
+
+// WithKeyBytes sets the byte slice for the Op's key.
+func (op *Op) WithKeyBytes(key []byte) { op.key = key }
+
+// RangeBytes returns the byte slice holding with the Op's range end, if any.
+func (op Op) RangeBytes() []byte { return op.end }
+
+// Rev returns the requested revision, if any.
+func (op Op) Rev() int64 { return op.rev }
+
+// IsPut returns true iff the operation is a Put.
+func (op Op) IsPut() bool { return op.t == tPut }
+
+// IsGet returns true iff the operation is a Get.
+func (op Op) IsGet() bool { return op.t == tRange }
+
+// IsDelete returns true iff the operation is a Delete.
+func (op Op) IsDelete() bool { return op.t == tDeleteRange }
+
+// IsSerializable returns true if the serializable field is true.
+func (op Op) IsSerializable() bool { return op.serializable == true }
+
+// IsKeysOnly returns whether keysOnly is set.
+func (op Op) IsKeysOnly() bool { return op.keysOnly == true }
+
+// IsCountOnly returns whether countOnly is set.
+func (op Op) IsCountOnly() bool { return op.countOnly == true }
+
+// MinModRev returns the operation's minimum modify revision.
+func (op Op) MinModRev() int64 { return op.minModRev }
+
+// MaxModRev returns the operation's maximum modify revision.
+func (op Op) MaxModRev() int64 { return op.maxModRev }
+
+// MinCreateRev returns the operation's minimum create revision.
+func (op Op) MinCreateRev() int64 { return op.minCreateRev }
+
+// MaxCreateRev returns the operation's maximum create revision.
+func (op Op) MaxCreateRev() int64 { return op.maxCreateRev }
+
+// WithRangeBytes sets the byte slice for the Op's range end.
+func (op *Op) WithRangeBytes(end []byte) { op.end = end }
+
+// ValueBytes returns the byte slice holding the Op's value, if any.
+func (op Op) ValueBytes() []byte { return op.val }
+
+// WithValueBytes sets the byte slice for the Op's value.
+func (op *Op) WithValueBytes(v []byte) { op.val = v }
+
+func (op Op) toRangeRequest() *pb.RangeRequest {
+ if op.t != tRange {
+ panic("op.t != tRange")
+ }
+ r := &pb.RangeRequest{
+ Key: op.key,
+ RangeEnd: op.end,
+ Limit: op.limit,
+ Revision: op.rev,
+ Serializable: op.serializable,
+ KeysOnly: op.keysOnly,
+ CountOnly: op.countOnly,
+ MinModRevision: op.minModRev,
+ MaxModRevision: op.maxModRev,
+ MinCreateRevision: op.minCreateRev,
+ MaxCreateRevision: op.maxCreateRev,
+ }
+ if op.sort != nil {
+ r.SortOrder = pb.RangeRequest_SortOrder(op.sort.Order)
+ r.SortTarget = pb.RangeRequest_SortTarget(op.sort.Target)
+ }
+ return r
+}
+
+func (op Op) toTxnRequest() *pb.TxnRequest {
+ thenOps := make([]*pb.RequestOp, len(op.thenOps))
+ for i, tOp := range op.thenOps {
+ thenOps[i] = tOp.toRequestOp()
+ }
+ elseOps := make([]*pb.RequestOp, len(op.elseOps))
+ for i, eOp := range op.elseOps {
+ elseOps[i] = eOp.toRequestOp()
+ }
+ cmps := make([]*pb.Compare, len(op.cmps))
+ for i := range op.cmps {
+ cmps[i] = (*pb.Compare)(&op.cmps[i])
+ }
+ return &pb.TxnRequest{Compare: cmps, Success: thenOps, Failure: elseOps}
+}
+
+func (op Op) toRequestOp() *pb.RequestOp {
+ switch op.t {
+ case tRange:
+ return &pb.RequestOp{Request: &pb.RequestOp_RequestRange{RequestRange: op.toRangeRequest()}}
+ case tPut:
+ r := &pb.PutRequest{Key: op.key, Value: op.val, Lease: int64(op.leaseID), PrevKv: op.prevKV, IgnoreValue: op.ignoreValue, IgnoreLease: op.ignoreLease}
+ return &pb.RequestOp{Request: &pb.RequestOp_RequestPut{RequestPut: r}}
+ case tDeleteRange:
+ r := &pb.DeleteRangeRequest{Key: op.key, RangeEnd: op.end, PrevKv: op.prevKV}
+ return &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{RequestDeleteRange: r}}
+ case tTxn:
+ return &pb.RequestOp{Request: &pb.RequestOp_RequestTxn{RequestTxn: op.toTxnRequest()}}
+ default:
+ panic("Unknown Op")
+ }
+}
+
+func (op Op) isWrite() bool {
+ if op.t == tTxn {
+ for _, tOp := range op.thenOps {
+ if tOp.isWrite() {
+ return true
+ }
+ }
+ for _, tOp := range op.elseOps {
+ if tOp.isWrite() {
+ return true
+ }
+ }
+ return false
+ }
+ return op.t != tRange
+}
+
+func OpGet(key string, opts ...OpOption) Op {
+ ret := Op{t: tRange, key: []byte(key)}
+ ret.applyOpts(opts)
+ return ret
+}
+
+func OpDelete(key string, opts ...OpOption) Op {
+ ret := Op{t: tDeleteRange, key: []byte(key)}
+ ret.applyOpts(opts)
+ switch {
+ case ret.leaseID != 0:
+ panic("unexpected lease in delete")
+ case ret.limit != 0:
+ panic("unexpected limit in delete")
+ case ret.rev != 0:
+ panic("unexpected revision in delete")
+ case ret.sort != nil:
+ panic("unexpected sort in delete")
+ case ret.serializable:
+ panic("unexpected serializable in delete")
+ case ret.countOnly:
+ panic("unexpected countOnly in delete")
+ case ret.minModRev != 0, ret.maxModRev != 0:
+ panic("unexpected mod revision filter in delete")
+ case ret.minCreateRev != 0, ret.maxCreateRev != 0:
+ panic("unexpected create revision filter in delete")
+ case ret.filterDelete, ret.filterPut:
+ panic("unexpected filter in delete")
+ case ret.createdNotify:
+ panic("unexpected createdNotify in delete")
+ }
+ return ret
+}
+
+func OpPut(key, val string, opts ...OpOption) Op {
+ ret := Op{t: tPut, key: []byte(key), val: []byte(val)}
+ ret.applyOpts(opts)
+ switch {
+ case ret.end != nil:
+ panic("unexpected range in put")
+ case ret.limit != 0:
+ panic("unexpected limit in put")
+ case ret.rev != 0:
+ panic("unexpected revision in put")
+ case ret.sort != nil:
+ panic("unexpected sort in put")
+ case ret.serializable:
+ panic("unexpected serializable in put")
+ case ret.countOnly:
+ panic("unexpected countOnly in put")
+ case ret.minModRev != 0, ret.maxModRev != 0:
+ panic("unexpected mod revision filter in put")
+ case ret.minCreateRev != 0, ret.maxCreateRev != 0:
+ panic("unexpected create revision filter in put")
+ case ret.filterDelete, ret.filterPut:
+ panic("unexpected filter in put")
+ case ret.createdNotify:
+ panic("unexpected createdNotify in put")
+ }
+ return ret
+}
+
+func OpTxn(cmps []Cmp, thenOps []Op, elseOps []Op) Op {
+ return Op{t: tTxn, cmps: cmps, thenOps: thenOps, elseOps: elseOps}
+}
+
+func opWatch(key string, opts ...OpOption) Op {
+ ret := Op{t: tRange, key: []byte(key)}
+ ret.applyOpts(opts)
+ switch {
+ case ret.leaseID != 0:
+ panic("unexpected lease in watch")
+ case ret.limit != 0:
+ panic("unexpected limit in watch")
+ case ret.sort != nil:
+ panic("unexpected sort in watch")
+ case ret.serializable:
+ panic("unexpected serializable in watch")
+ case ret.countOnly:
+ panic("unexpected countOnly in watch")
+ case ret.minModRev != 0, ret.maxModRev != 0:
+ panic("unexpected mod revision filter in watch")
+ case ret.minCreateRev != 0, ret.maxCreateRev != 0:
+ panic("unexpected create revision filter in watch")
+ }
+ return ret
+}
+
+func (op *Op) applyOpts(opts []OpOption) {
+ for _, opt := range opts {
+ opt(op)
+ }
+}
+
+// OpOption configures Operations like Get, Put, Delete.
+type OpOption func(*Op)
+
+// WithLease attaches a lease ID to a key in 'Put' request.
+func WithLease(leaseID LeaseID) OpOption {
+ return func(op *Op) { op.leaseID = leaseID }
+}
+
+// WithLimit limits the number of results to return from 'Get' request.
+// If WithLimit is given a 0 limit, it is treated as no limit.
+func WithLimit(n int64) OpOption { return func(op *Op) { op.limit = n } }
+
+// WithRev specifies the store revision for 'Get' request.
+// Or the start revision of 'Watch' request.
+func WithRev(rev int64) OpOption { return func(op *Op) { op.rev = rev } }
+
+// WithSort specifies the ordering in 'Get' request. It requires
+// 'WithRange' and/or 'WithPrefix' to be specified too.
+// 'target' specifies the target to sort by: key, version, revisions, value.
+// 'order' can be either 'SortNone', 'SortAscend', 'SortDescend'.
+func WithSort(target SortTarget, order SortOrder) OpOption {
+ return func(op *Op) {
+ if target == SortByKey && order == SortAscend {
+ // If order != SortNone, server fetches the entire key-space,
+ // and then applies the sort and limit, if provided.
+ // Since by default the server returns results sorted by keys
+ // in lexicographically ascending order, the client should ignore
+ // SortOrder if the target is SortByKey.
+ order = SortNone
+ }
+ op.sort = &SortOption{target, order}
+ }
+}
+
+// GetPrefixRangeEnd gets the range end of the prefix.
+// 'Get(foo, WithPrefix())' is equal to 'Get(foo, WithRange(GetPrefixRangeEnd(foo))'.
+func GetPrefixRangeEnd(prefix string) string {
+ return string(getPrefix([]byte(prefix)))
+}
+
+func getPrefix(key []byte) []byte {
+ end := make([]byte, len(key))
+ copy(end, key)
+ for i := len(end) - 1; i >= 0; i-- {
+ if end[i] < 0xff {
+ end[i] = end[i] + 1
+ end = end[:i+1]
+ return end
+ }
+ }
+ // next prefix does not exist (e.g., 0xffff);
+ // default to WithFromKey policy
+ return noPrefixEnd
+}
+
+// WithPrefix enables 'Get', 'Delete', or 'Watch' requests to operate
+// on the keys with matching prefix. For example, 'Get(foo, WithPrefix())'
+// can return 'foo1', 'foo2', and so on.
+func WithPrefix() OpOption {
+ return func(op *Op) {
+ if len(op.key) == 0 {
+ op.key, op.end = []byte{0}, []byte{0}
+ return
+ }
+ op.end = getPrefix(op.key)
+ }
+}
+
+// WithRange specifies the range of 'Get', 'Delete', 'Watch' requests.
+// For example, 'Get' requests with 'WithRange(end)' returns
+// the keys in the range [key, end).
+// endKey must be lexicographically greater than start key.
+func WithRange(endKey string) OpOption {
+ return func(op *Op) { op.end = []byte(endKey) }
+}
+
+// WithFromKey specifies the range of 'Get', 'Delete', 'Watch' requests
+// to be equal or greater than the key in the argument.
+func WithFromKey() OpOption { return WithRange("\x00") }
+
+// WithSerializable makes 'Get' request serializable. By default,
+// it's linearizable. Serializable requests are better for lower latency
+// requirement.
+func WithSerializable() OpOption {
+ return func(op *Op) { op.serializable = true }
+}
+
+// WithKeysOnly makes the 'Get' request return only the keys and the corresponding
+// values will be omitted.
+func WithKeysOnly() OpOption {
+ return func(op *Op) { op.keysOnly = true }
+}
+
+// WithCountOnly makes the 'Get' request return only the count of keys.
+func WithCountOnly() OpOption {
+ return func(op *Op) { op.countOnly = true }
+}
+
+// WithMinModRev filters out keys for Get with modification revisions less than the given revision.
+func WithMinModRev(rev int64) OpOption { return func(op *Op) { op.minModRev = rev } }
+
+// WithMaxModRev filters out keys for Get with modification revisions greater than the given revision.
+func WithMaxModRev(rev int64) OpOption { return func(op *Op) { op.maxModRev = rev } }
+
+// WithMinCreateRev filters out keys for Get with creation revisions less than the given revision.
+func WithMinCreateRev(rev int64) OpOption { return func(op *Op) { op.minCreateRev = rev } }
+
+// WithMaxCreateRev filters out keys for Get with creation revisions greater than the given revision.
+func WithMaxCreateRev(rev int64) OpOption { return func(op *Op) { op.maxCreateRev = rev } }
+
+// WithFirstCreate gets the key with the oldest creation revision in the request range.
+func WithFirstCreate() []OpOption { return withTop(SortByCreateRevision, SortAscend) }
+
+// WithLastCreate gets the key with the latest creation revision in the request range.
+func WithLastCreate() []OpOption { return withTop(SortByCreateRevision, SortDescend) }
+
+// WithFirstKey gets the lexically first key in the request range.
+func WithFirstKey() []OpOption { return withTop(SortByKey, SortAscend) }
+
+// WithLastKey gets the lexically last key in the request range.
+func WithLastKey() []OpOption { return withTop(SortByKey, SortDescend) }
+
+// WithFirstRev gets the key with the oldest modification revision in the request range.
+func WithFirstRev() []OpOption { return withTop(SortByModRevision, SortAscend) }
+
+// WithLastRev gets the key with the latest modification revision in the request range.
+func WithLastRev() []OpOption { return withTop(SortByModRevision, SortDescend) }
+
+// withTop gets the first key over the get's prefix given a sort order
+func withTop(target SortTarget, order SortOrder) []OpOption {
+ return []OpOption{WithPrefix(), WithSort(target, order), WithLimit(1)}
+}
+
+// WithProgressNotify makes watch server send periodic progress updates
+// every 10 minutes when there is no incoming events.
+// Progress updates have zero events in WatchResponse.
+func WithProgressNotify() OpOption {
+ return func(op *Op) {
+ op.progressNotify = true
+ }
+}
+
+// WithCreatedNotify makes watch server sends the created event.
+func WithCreatedNotify() OpOption {
+ return func(op *Op) {
+ op.createdNotify = true
+ }
+}
+
+// WithFilterPut discards PUT events from the watcher.
+func WithFilterPut() OpOption {
+ return func(op *Op) { op.filterPut = true }
+}
+
+// WithFilterDelete discards DELETE events from the watcher.
+func WithFilterDelete() OpOption {
+ return func(op *Op) { op.filterDelete = true }
+}
+
+// WithPrevKV gets the previous key-value pair before the event happens. If the previous KV is already compacted,
+// nothing will be returned.
+func WithPrevKV() OpOption {
+ return func(op *Op) {
+ op.prevKV = true
+ }
+}
+
+// WithIgnoreValue updates the key using its current value.
+// This option can not be combined with non-empty values.
+// Returns an error if the key does not exist.
+func WithIgnoreValue() OpOption {
+ return func(op *Op) {
+ op.ignoreValue = true
+ }
+}
+
+// WithIgnoreLease updates the key using its current lease.
+// This option can not be combined with WithLease.
+// Returns an error if the key does not exist.
+func WithIgnoreLease() OpOption {
+ return func(op *Op) {
+ op.ignoreLease = true
+ }
+}
+
+// LeaseOp represents an Operation that lease can execute.
+type LeaseOp struct {
+ id LeaseID
+
+ // for TimeToLive
+ attachedKeys bool
+}
+
+// LeaseOption configures lease operations.
+type LeaseOption func(*LeaseOp)
+
+func (op *LeaseOp) applyOpts(opts []LeaseOption) {
+ for _, opt := range opts {
+ opt(op)
+ }
+}
+
+// WithAttachedKeys makes TimeToLive list the keys attached to the given lease ID.
+func WithAttachedKeys() LeaseOption {
+ return func(op *LeaseOp) { op.attachedKeys = true }
+}
+
+func toLeaseTimeToLiveRequest(id LeaseID, opts ...LeaseOption) *pb.LeaseTimeToLiveRequest {
+ ret := &LeaseOp{id: id}
+ ret.applyOpts(opts)
+ return &pb.LeaseTimeToLiveRequest{ID: int64(id), Keys: ret.attachedKeys}
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/options.go b/vendor/github.com/coreos/etcd/clientv3/options.go
new file mode 100644
index 0000000..fa25811
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/options.go
@@ -0,0 +1,49 @@
+// Copyright 2017 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 clientv3
+
+import (
+ "math"
+
+ "google.golang.org/grpc"
+)
+
+var (
+ // Disable gRPC internal retrial logic
+ // TODO: enable when gRPC retry is stable (FailFast=false)
+ // Reference:
+ // - https://github.com/grpc/grpc-go/issues/1532
+ // - https://github.com/grpc/proposal/blob/master/A6-client-retries.md
+ defaultFailFast = grpc.FailFast(true)
+
+ // client-side request send limit, gRPC default is math.MaxInt32
+ // Make sure that "client-side send limit < server-side default send/recv limit"
+ // Same value as "embed.DefaultMaxRequestBytes" plus gRPC overhead bytes
+ defaultMaxCallSendMsgSize = grpc.MaxCallSendMsgSize(2 * 1024 * 1024)
+
+ // client-side response receive limit, gRPC default is 4MB
+ // Make sure that "client-side receive limit >= server-side default send/recv limit"
+ // because range response can easily exceed request send limits
+ // Default to math.MaxInt32; writes exceeding server-side send limit fails anyway
+ defaultMaxCallRecvMsgSize = grpc.MaxCallRecvMsgSize(math.MaxInt32)
+)
+
+// defaultCallOpts defines a list of default "gRPC.CallOption".
+// Some options are exposed to "clientv3.Config".
+// Defaults will be overridden by the settings in "clientv3.Config".
+var defaultCallOpts = []grpc.CallOption{defaultFailFast, defaultMaxCallSendMsgSize, defaultMaxCallRecvMsgSize}
+
+// MaxLeaseTTL is the maximum lease TTL value
+const MaxLeaseTTL = 9000000000
diff --git a/vendor/github.com/coreos/etcd/clientv3/ordering/doc.go b/vendor/github.com/coreos/etcd/clientv3/ordering/doc.go
new file mode 100644
index 0000000..856f330
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/ordering/doc.go
@@ -0,0 +1,42 @@
+// Copyright 2017 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 ordering is a clientv3 wrapper that caches response header revisions
+// to detect ordering violations from stale responses. Users may define a
+// policy on how to handle the ordering violation, but typically the client
+// should connect to another endpoint and reissue the request.
+//
+// The most common situation where an ordering violation happens is a client
+// reconnects to a partitioned member and issues a serializable read. Since the
+// partitioned member is likely behind the last member, it may return a Get
+// response based on a store revision older than the store revision used to
+// service a prior Get on the former endpoint.
+//
+// First, create a client:
+//
+// cli, err := clientv3.New(clientv3.Config{Endpoints: []string{"localhost:2379"}})
+// if err != nil {
+// // handle error!
+// }
+//
+// Next, override the client interface with the ordering wrapper:
+//
+// vf := func(op clientv3.Op, resp clientv3.OpResponse, prevRev int64) error {
+// return fmt.Errorf("ordering: issued %+v, got %+v, expected rev=%v", op, resp, prevRev)
+// }
+// cli.KV = ordering.NewKV(cli.KV, vf)
+//
+// Now calls using 'cli' will reject order violations with an error.
+//
+package ordering
diff --git a/vendor/github.com/coreos/etcd/clientv3/ordering/kv.go b/vendor/github.com/coreos/etcd/clientv3/ordering/kv.go
new file mode 100644
index 0000000..dc9926e
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/ordering/kv.go
@@ -0,0 +1,149 @@
+// Copyright 2017 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 ordering
+
+import (
+ "context"
+ "sync"
+
+ "github.com/coreos/etcd/clientv3"
+)
+
+// kvOrdering ensures that serialized requests do not return
+// get with revisions less than the previous
+// returned revision.
+type kvOrdering struct {
+ clientv3.KV
+ orderViolationFunc OrderViolationFunc
+ prevRev int64
+ revMu sync.RWMutex
+}
+
+func NewKV(kv clientv3.KV, orderViolationFunc OrderViolationFunc) *kvOrdering {
+ return &kvOrdering{kv, orderViolationFunc, 0, sync.RWMutex{}}
+}
+
+func (kv *kvOrdering) getPrevRev() int64 {
+ kv.revMu.RLock()
+ defer kv.revMu.RUnlock()
+ return kv.prevRev
+}
+
+func (kv *kvOrdering) setPrevRev(currRev int64) {
+ kv.revMu.Lock()
+ defer kv.revMu.Unlock()
+ if currRev > kv.prevRev {
+ kv.prevRev = currRev
+ }
+}
+
+func (kv *kvOrdering) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) {
+ // prevRev is stored in a local variable in order to record the prevRev
+ // at the beginning of the Get operation, because concurrent
+ // access to kvOrdering could change the prevRev field in the
+ // middle of the Get operation.
+ prevRev := kv.getPrevRev()
+ op := clientv3.OpGet(key, opts...)
+ for {
+ r, err := kv.KV.Do(ctx, op)
+ if err != nil {
+ return nil, err
+ }
+ resp := r.Get()
+ if resp.Header.Revision == prevRev {
+ return resp, nil
+ } else if resp.Header.Revision > prevRev {
+ kv.setPrevRev(resp.Header.Revision)
+ return resp, nil
+ }
+ err = kv.orderViolationFunc(op, r, prevRev)
+ if err != nil {
+ return nil, err
+ }
+ }
+}
+
+func (kv *kvOrdering) Txn(ctx context.Context) clientv3.Txn {
+ return &txnOrdering{
+ kv.KV.Txn(ctx),
+ kv,
+ ctx,
+ sync.Mutex{},
+ []clientv3.Cmp{},
+ []clientv3.Op{},
+ []clientv3.Op{},
+ }
+}
+
+// txnOrdering ensures that serialized requests do not return
+// txn responses with revisions less than the previous
+// returned revision.
+type txnOrdering struct {
+ clientv3.Txn
+ *kvOrdering
+ ctx context.Context
+ mu sync.Mutex
+ cmps []clientv3.Cmp
+ thenOps []clientv3.Op
+ elseOps []clientv3.Op
+}
+
+func (txn *txnOrdering) If(cs ...clientv3.Cmp) clientv3.Txn {
+ txn.mu.Lock()
+ defer txn.mu.Unlock()
+ txn.cmps = cs
+ txn.Txn.If(cs...)
+ return txn
+}
+
+func (txn *txnOrdering) Then(ops ...clientv3.Op) clientv3.Txn {
+ txn.mu.Lock()
+ defer txn.mu.Unlock()
+ txn.thenOps = ops
+ txn.Txn.Then(ops...)
+ return txn
+}
+
+func (txn *txnOrdering) Else(ops ...clientv3.Op) clientv3.Txn {
+ txn.mu.Lock()
+ defer txn.mu.Unlock()
+ txn.elseOps = ops
+ txn.Txn.Else(ops...)
+ return txn
+}
+
+func (txn *txnOrdering) Commit() (*clientv3.TxnResponse, error) {
+ // prevRev is stored in a local variable in order to record the prevRev
+ // at the beginning of the Commit operation, because concurrent
+ // access to txnOrdering could change the prevRev field in the
+ // middle of the Commit operation.
+ prevRev := txn.getPrevRev()
+ opTxn := clientv3.OpTxn(txn.cmps, txn.thenOps, txn.elseOps)
+ for {
+ opResp, err := txn.KV.Do(txn.ctx, opTxn)
+ if err != nil {
+ return nil, err
+ }
+ txnResp := opResp.Txn()
+ if txnResp.Header.Revision >= prevRev {
+ txn.setPrevRev(txnResp.Header.Revision)
+ return txnResp, nil
+ }
+ err = txn.orderViolationFunc(opTxn, opResp, prevRev)
+ if err != nil {
+ return nil, err
+ }
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/ordering/util.go b/vendor/github.com/coreos/etcd/clientv3/ordering/util.go
new file mode 100644
index 0000000..190a591
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/ordering/util.go
@@ -0,0 +1,49 @@
+// Copyright 2017 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 ordering
+
+import (
+ "errors"
+ "sync"
+ "time"
+
+ "github.com/coreos/etcd/clientv3"
+)
+
+type OrderViolationFunc func(op clientv3.Op, resp clientv3.OpResponse, prevRev int64) error
+
+var ErrNoGreaterRev = errors.New("etcdclient: no cluster members have a revision higher than the previously received revision")
+
+func NewOrderViolationSwitchEndpointClosure(c clientv3.Client) OrderViolationFunc {
+ var mu sync.Mutex
+ violationCount := 0
+ return func(op clientv3.Op, resp clientv3.OpResponse, prevRev int64) error {
+ if violationCount > len(c.Endpoints()) {
+ return ErrNoGreaterRev
+ }
+ mu.Lock()
+ defer mu.Unlock()
+ eps := c.Endpoints()
+ // force client to connect to given endpoint by limiting to a single endpoint
+ c.SetEndpoints(eps[violationCount%len(eps)])
+ // give enough time for operation
+ time.Sleep(1 * time.Second)
+ // set available endpoints back to all endpoints in to ensure
+ // the client has access to all the endpoints.
+ c.SetEndpoints(eps...)
+ violationCount++
+ return nil
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/ready_wait.go b/vendor/github.com/coreos/etcd/clientv3/ready_wait.go
new file mode 100644
index 0000000..c6ef585
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/ready_wait.go
@@ -0,0 +1,30 @@
+// Copyright 2017 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 clientv3
+
+import "context"
+
+// TODO: remove this when "FailFast=false" is fixed.
+// See https://github.com/grpc/grpc-go/issues/1532.
+func readyWait(rpcCtx, clientCtx context.Context, ready <-chan struct{}) error {
+ select {
+ case <-ready:
+ return nil
+ case <-rpcCtx.Done():
+ return rpcCtx.Err()
+ case <-clientCtx.Done():
+ return clientCtx.Err()
+ }
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/retry.go b/vendor/github.com/coreos/etcd/clientv3/retry.go
new file mode 100644
index 0000000..7f89ba6
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/retry.go
@@ -0,0 +1,496 @@
+// 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 clientv3
+
+import (
+ "context"
+
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/status"
+)
+
+type retryPolicy uint8
+
+const (
+ repeatable retryPolicy = iota
+ nonRepeatable
+)
+
+type rpcFunc func(ctx context.Context) error
+type retryRPCFunc func(context.Context, rpcFunc, retryPolicy) error
+type retryStopErrFunc func(error) bool
+
+// immutable requests (e.g. Get) should be retried unless it's
+// an obvious server-side error (e.g. rpctypes.ErrRequestTooLarge).
+//
+// "isRepeatableStopError" returns "true" when an immutable request
+// is interrupted by server-side or gRPC-side error and its status
+// code is not transient (!= codes.Unavailable).
+//
+// Returning "true" means retry should stop, since client cannot
+// handle itself even with retries.
+func isRepeatableStopError(err error) bool {
+ eErr := rpctypes.Error(err)
+ // always stop retry on etcd errors
+ if serverErr, ok := eErr.(rpctypes.EtcdError); ok && serverErr.Code() != codes.Unavailable {
+ return true
+ }
+ // only retry if unavailable
+ ev, _ := status.FromError(err)
+ return ev.Code() != codes.Unavailable
+}
+
+// mutable requests (e.g. Put, Delete, Txn) should only be retried
+// when the status code is codes.Unavailable when initial connection
+// has not been established (no pinned endpoint).
+//
+// "isNonRepeatableStopError" returns "true" when a mutable request
+// is interrupted by non-transient error that client cannot handle itself,
+// or transient error while the connection has already been established
+// (pinned endpoint exists).
+//
+// Returning "true" means retry should stop, otherwise it violates
+// write-at-most-once semantics.
+func isNonRepeatableStopError(err error) bool {
+ ev, _ := status.FromError(err)
+ if ev.Code() != codes.Unavailable {
+ return true
+ }
+ desc := rpctypes.ErrorDesc(err)
+ return desc != "there is no address available" && desc != "there is no connection available"
+}
+
+func (c *Client) newRetryWrapper() retryRPCFunc {
+ return func(rpcCtx context.Context, f rpcFunc, rp retryPolicy) error {
+ var isStop retryStopErrFunc
+ switch rp {
+ case repeatable:
+ isStop = isRepeatableStopError
+ case nonRepeatable:
+ isStop = isNonRepeatableStopError
+ }
+ for {
+ if err := readyWait(rpcCtx, c.ctx, c.balancer.ConnectNotify()); err != nil {
+ return err
+ }
+ pinned := c.balancer.pinned()
+ err := f(rpcCtx)
+ if err == nil {
+ return nil
+ }
+ logger.Lvl(4).Infof("clientv3/retry: error %q on pinned endpoint %q", err.Error(), pinned)
+
+ if s, ok := status.FromError(err); ok && (s.Code() == codes.Unavailable || s.Code() == codes.DeadlineExceeded || s.Code() == codes.Internal) {
+ // mark this before endpoint switch is triggered
+ c.balancer.hostPortError(pinned, err)
+ c.balancer.next()
+ logger.Lvl(4).Infof("clientv3/retry: switching from %q due to error %q", pinned, err.Error())
+ }
+
+ if isStop(err) {
+ return err
+ }
+ }
+ }
+}
+
+func (c *Client) newAuthRetryWrapper(retryf retryRPCFunc) retryRPCFunc {
+ return func(rpcCtx context.Context, f rpcFunc, rp retryPolicy) error {
+ for {
+ pinned := c.balancer.pinned()
+ err := retryf(rpcCtx, f, rp)
+ if err == nil {
+ return nil
+ }
+ logger.Lvl(4).Infof("clientv3/auth-retry: error %q on pinned endpoint %q", err.Error(), pinned)
+ // always stop retry on etcd errors other than invalid auth token
+ if rpctypes.Error(err) == rpctypes.ErrInvalidAuthToken {
+ gterr := c.getToken(rpcCtx)
+ if gterr != nil {
+ logger.Lvl(4).Infof("clientv3/auth-retry: cannot retry due to error %q(%q) on pinned endpoint %q", err.Error(), gterr.Error(), pinned)
+ return err // return the original error for simplicity
+ }
+ continue
+ }
+ return err
+ }
+ }
+}
+
+type retryKVClient struct {
+ kc pb.KVClient
+ retryf retryRPCFunc
+}
+
+// RetryKVClient implements a KVClient.
+func RetryKVClient(c *Client) pb.KVClient {
+ return &retryKVClient{
+ kc: pb.NewKVClient(c.conn),
+ retryf: c.newAuthRetryWrapper(c.newRetryWrapper()),
+ }
+}
+func (rkv *retryKVClient) Range(ctx context.Context, in *pb.RangeRequest, opts ...grpc.CallOption) (resp *pb.RangeResponse, err error) {
+ err = rkv.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rkv.kc.Range(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rkv *retryKVClient) Put(ctx context.Context, in *pb.PutRequest, opts ...grpc.CallOption) (resp *pb.PutResponse, err error) {
+ err = rkv.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rkv.kc.Put(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rkv *retryKVClient) DeleteRange(ctx context.Context, in *pb.DeleteRangeRequest, opts ...grpc.CallOption) (resp *pb.DeleteRangeResponse, err error) {
+ err = rkv.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rkv.kc.DeleteRange(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rkv *retryKVClient) Txn(ctx context.Context, in *pb.TxnRequest, opts ...grpc.CallOption) (resp *pb.TxnResponse, err error) {
+ // TODO: "repeatable" for read-only txn
+ err = rkv.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rkv.kc.Txn(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rkv *retryKVClient) Compact(ctx context.Context, in *pb.CompactionRequest, opts ...grpc.CallOption) (resp *pb.CompactionResponse, err error) {
+ err = rkv.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rkv.kc.Compact(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+type retryLeaseClient struct {
+ lc pb.LeaseClient
+ retryf retryRPCFunc
+}
+
+// RetryLeaseClient implements a LeaseClient.
+func RetryLeaseClient(c *Client) pb.LeaseClient {
+ return &retryLeaseClient{
+ lc: pb.NewLeaseClient(c.conn),
+ retryf: c.newAuthRetryWrapper(c.newRetryWrapper()),
+ }
+}
+
+func (rlc *retryLeaseClient) LeaseTimeToLive(ctx context.Context, in *pb.LeaseTimeToLiveRequest, opts ...grpc.CallOption) (resp *pb.LeaseTimeToLiveResponse, err error) {
+ err = rlc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rlc.lc.LeaseTimeToLive(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rlc *retryLeaseClient) LeaseLeases(ctx context.Context, in *pb.LeaseLeasesRequest, opts ...grpc.CallOption) (resp *pb.LeaseLeasesResponse, err error) {
+ err = rlc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rlc.lc.LeaseLeases(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rlc *retryLeaseClient) LeaseGrant(ctx context.Context, in *pb.LeaseGrantRequest, opts ...grpc.CallOption) (resp *pb.LeaseGrantResponse, err error) {
+ err = rlc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rlc.lc.LeaseGrant(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+
+}
+
+func (rlc *retryLeaseClient) LeaseRevoke(ctx context.Context, in *pb.LeaseRevokeRequest, opts ...grpc.CallOption) (resp *pb.LeaseRevokeResponse, err error) {
+ err = rlc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rlc.lc.LeaseRevoke(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rlc *retryLeaseClient) LeaseKeepAlive(ctx context.Context, opts ...grpc.CallOption) (stream pb.Lease_LeaseKeepAliveClient, err error) {
+ err = rlc.retryf(ctx, func(rctx context.Context) error {
+ stream, err = rlc.lc.LeaseKeepAlive(rctx, opts...)
+ return err
+ }, repeatable)
+ return stream, err
+}
+
+type retryClusterClient struct {
+ cc pb.ClusterClient
+ retryf retryRPCFunc
+}
+
+// RetryClusterClient implements a ClusterClient.
+func RetryClusterClient(c *Client) pb.ClusterClient {
+ return &retryClusterClient{
+ cc: pb.NewClusterClient(c.conn),
+ retryf: c.newRetryWrapper(),
+ }
+}
+
+func (rcc *retryClusterClient) MemberList(ctx context.Context, in *pb.MemberListRequest, opts ...grpc.CallOption) (resp *pb.MemberListResponse, err error) {
+ err = rcc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rcc.cc.MemberList(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rcc *retryClusterClient) MemberAdd(ctx context.Context, in *pb.MemberAddRequest, opts ...grpc.CallOption) (resp *pb.MemberAddResponse, err error) {
+ err = rcc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rcc.cc.MemberAdd(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rcc *retryClusterClient) MemberRemove(ctx context.Context, in *pb.MemberRemoveRequest, opts ...grpc.CallOption) (resp *pb.MemberRemoveResponse, err error) {
+ err = rcc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rcc.cc.MemberRemove(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rcc *retryClusterClient) MemberUpdate(ctx context.Context, in *pb.MemberUpdateRequest, opts ...grpc.CallOption) (resp *pb.MemberUpdateResponse, err error) {
+ err = rcc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rcc.cc.MemberUpdate(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+type retryMaintenanceClient struct {
+ mc pb.MaintenanceClient
+ retryf retryRPCFunc
+}
+
+// RetryMaintenanceClient implements a Maintenance.
+func RetryMaintenanceClient(c *Client, conn *grpc.ClientConn) pb.MaintenanceClient {
+ return &retryMaintenanceClient{
+ mc: pb.NewMaintenanceClient(conn),
+ retryf: c.newRetryWrapper(),
+ }
+}
+
+func (rmc *retryMaintenanceClient) Alarm(ctx context.Context, in *pb.AlarmRequest, opts ...grpc.CallOption) (resp *pb.AlarmResponse, err error) {
+ err = rmc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rmc.mc.Alarm(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rmc *retryMaintenanceClient) Status(ctx context.Context, in *pb.StatusRequest, opts ...grpc.CallOption) (resp *pb.StatusResponse, err error) {
+ err = rmc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rmc.mc.Status(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rmc *retryMaintenanceClient) Hash(ctx context.Context, in *pb.HashRequest, opts ...grpc.CallOption) (resp *pb.HashResponse, err error) {
+ err = rmc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rmc.mc.Hash(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rmc *retryMaintenanceClient) HashKV(ctx context.Context, in *pb.HashKVRequest, opts ...grpc.CallOption) (resp *pb.HashKVResponse, err error) {
+ err = rmc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rmc.mc.HashKV(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rmc *retryMaintenanceClient) Snapshot(ctx context.Context, in *pb.SnapshotRequest, opts ...grpc.CallOption) (stream pb.Maintenance_SnapshotClient, err error) {
+ err = rmc.retryf(ctx, func(rctx context.Context) error {
+ stream, err = rmc.mc.Snapshot(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return stream, err
+}
+
+func (rmc *retryMaintenanceClient) MoveLeader(ctx context.Context, in *pb.MoveLeaderRequest, opts ...grpc.CallOption) (resp *pb.MoveLeaderResponse, err error) {
+ err = rmc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rmc.mc.MoveLeader(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rmc *retryMaintenanceClient) Defragment(ctx context.Context, in *pb.DefragmentRequest, opts ...grpc.CallOption) (resp *pb.DefragmentResponse, err error) {
+ err = rmc.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rmc.mc.Defragment(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+type retryAuthClient struct {
+ ac pb.AuthClient
+ retryf retryRPCFunc
+}
+
+// RetryAuthClient implements a AuthClient.
+func RetryAuthClient(c *Client) pb.AuthClient {
+ return &retryAuthClient{
+ ac: pb.NewAuthClient(c.conn),
+ retryf: c.newRetryWrapper(),
+ }
+}
+
+func (rac *retryAuthClient) UserList(ctx context.Context, in *pb.AuthUserListRequest, opts ...grpc.CallOption) (resp *pb.AuthUserListResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.UserList(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) UserGet(ctx context.Context, in *pb.AuthUserGetRequest, opts ...grpc.CallOption) (resp *pb.AuthUserGetResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.UserGet(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) RoleGet(ctx context.Context, in *pb.AuthRoleGetRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleGetResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.RoleGet(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) RoleList(ctx context.Context, in *pb.AuthRoleListRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleListResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.RoleList(rctx, in, opts...)
+ return err
+ }, repeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) AuthEnable(ctx context.Context, in *pb.AuthEnableRequest, opts ...grpc.CallOption) (resp *pb.AuthEnableResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.AuthEnable(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) AuthDisable(ctx context.Context, in *pb.AuthDisableRequest, opts ...grpc.CallOption) (resp *pb.AuthDisableResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.AuthDisable(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) UserAdd(ctx context.Context, in *pb.AuthUserAddRequest, opts ...grpc.CallOption) (resp *pb.AuthUserAddResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.UserAdd(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) UserDelete(ctx context.Context, in *pb.AuthUserDeleteRequest, opts ...grpc.CallOption) (resp *pb.AuthUserDeleteResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.UserDelete(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) UserChangePassword(ctx context.Context, in *pb.AuthUserChangePasswordRequest, opts ...grpc.CallOption) (resp *pb.AuthUserChangePasswordResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.UserChangePassword(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) UserGrantRole(ctx context.Context, in *pb.AuthUserGrantRoleRequest, opts ...grpc.CallOption) (resp *pb.AuthUserGrantRoleResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.UserGrantRole(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) UserRevokeRole(ctx context.Context, in *pb.AuthUserRevokeRoleRequest, opts ...grpc.CallOption) (resp *pb.AuthUserRevokeRoleResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.UserRevokeRole(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) RoleAdd(ctx context.Context, in *pb.AuthRoleAddRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleAddResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.RoleAdd(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) RoleDelete(ctx context.Context, in *pb.AuthRoleDeleteRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleDeleteResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.RoleDelete(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) RoleGrantPermission(ctx context.Context, in *pb.AuthRoleGrantPermissionRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleGrantPermissionResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.RoleGrantPermission(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) RoleRevokePermission(ctx context.Context, in *pb.AuthRoleRevokePermissionRequest, opts ...grpc.CallOption) (resp *pb.AuthRoleRevokePermissionResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.RoleRevokePermission(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
+
+func (rac *retryAuthClient) Authenticate(ctx context.Context, in *pb.AuthenticateRequest, opts ...grpc.CallOption) (resp *pb.AuthenticateResponse, err error) {
+ err = rac.retryf(ctx, func(rctx context.Context) error {
+ resp, err = rac.ac.Authenticate(rctx, in, opts...)
+ return err
+ }, nonRepeatable)
+ return resp, err
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/sort.go b/vendor/github.com/coreos/etcd/clientv3/sort.go
new file mode 100644
index 0000000..2bb9d9a
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/sort.go
@@ -0,0 +1,37 @@
+// 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 clientv3
+
+type SortTarget int
+type SortOrder int
+
+const (
+ SortNone SortOrder = iota
+ SortAscend
+ SortDescend
+)
+
+const (
+ SortByKey SortTarget = iota
+ SortByVersion
+ SortByCreateRevision
+ SortByModRevision
+ SortByValue
+)
+
+type SortOption struct {
+ Target SortTarget
+ Order SortOrder
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/txn.go b/vendor/github.com/coreos/etcd/clientv3/txn.go
new file mode 100644
index 0000000..c3c2d24
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/txn.go
@@ -0,0 +1,151 @@
+// 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 clientv3
+
+import (
+ "context"
+ "sync"
+
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+
+ "google.golang.org/grpc"
+)
+
+// Txn is the interface that wraps mini-transactions.
+//
+// Txn(context.TODO()).If(
+// Compare(Value(k1), ">", v1),
+// Compare(Version(k1), "=", 2)
+// ).Then(
+// OpPut(k2,v2), OpPut(k3,v3)
+// ).Else(
+// OpPut(k4,v4), OpPut(k5,v5)
+// ).Commit()
+//
+type Txn interface {
+ // If takes a list of comparison. If all comparisons passed in succeed,
+ // the operations passed into Then() will be executed. Or the operations
+ // passed into Else() will be executed.
+ If(cs ...Cmp) Txn
+
+ // Then takes a list of operations. The Ops list will be executed, if the
+ // comparisons passed in If() succeed.
+ Then(ops ...Op) Txn
+
+ // Else takes a list of operations. The Ops list will be executed, if the
+ // comparisons passed in If() fail.
+ Else(ops ...Op) Txn
+
+ // Commit tries to commit the transaction.
+ Commit() (*TxnResponse, error)
+}
+
+type txn struct {
+ kv *kv
+ ctx context.Context
+
+ mu sync.Mutex
+ cif bool
+ cthen bool
+ celse bool
+
+ isWrite bool
+
+ cmps []*pb.Compare
+
+ sus []*pb.RequestOp
+ fas []*pb.RequestOp
+
+ callOpts []grpc.CallOption
+}
+
+func (txn *txn) If(cs ...Cmp) Txn {
+ txn.mu.Lock()
+ defer txn.mu.Unlock()
+
+ if txn.cif {
+ panic("cannot call If twice!")
+ }
+
+ if txn.cthen {
+ panic("cannot call If after Then!")
+ }
+
+ if txn.celse {
+ panic("cannot call If after Else!")
+ }
+
+ txn.cif = true
+
+ for i := range cs {
+ txn.cmps = append(txn.cmps, (*pb.Compare)(&cs[i]))
+ }
+
+ return txn
+}
+
+func (txn *txn) Then(ops ...Op) Txn {
+ txn.mu.Lock()
+ defer txn.mu.Unlock()
+
+ if txn.cthen {
+ panic("cannot call Then twice!")
+ }
+ if txn.celse {
+ panic("cannot call Then after Else!")
+ }
+
+ txn.cthen = true
+
+ for _, op := range ops {
+ txn.isWrite = txn.isWrite || op.isWrite()
+ txn.sus = append(txn.sus, op.toRequestOp())
+ }
+
+ return txn
+}
+
+func (txn *txn) Else(ops ...Op) Txn {
+ txn.mu.Lock()
+ defer txn.mu.Unlock()
+
+ if txn.celse {
+ panic("cannot call Else twice!")
+ }
+
+ txn.celse = true
+
+ for _, op := range ops {
+ txn.isWrite = txn.isWrite || op.isWrite()
+ txn.fas = append(txn.fas, op.toRequestOp())
+ }
+
+ return txn
+}
+
+func (txn *txn) Commit() (*TxnResponse, error) {
+ txn.mu.Lock()
+ defer txn.mu.Unlock()
+
+ r := &pb.TxnRequest{Compare: txn.cmps, Success: txn.sus, Failure: txn.fas}
+
+ var resp *pb.TxnResponse
+ var err error
+ resp, err = txn.kv.remote.Txn(txn.ctx, r, txn.callOpts...)
+ if err != nil {
+ return nil, toErr(txn.ctx, err)
+ }
+ return (*TxnResponse)(resp), nil
+}
diff --git a/vendor/github.com/coreos/etcd/clientv3/watch.go b/vendor/github.com/coreos/etcd/clientv3/watch.go
new file mode 100644
index 0000000..d763385
--- /dev/null
+++ b/vendor/github.com/coreos/etcd/clientv3/watch.go
@@ -0,0 +1,828 @@
+// 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 clientv3
+
+import (
+ "context"
+ "fmt"
+ "sync"
+ "time"
+
+ v3rpc "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
+ mvccpb "github.com/coreos/etcd/mvcc/mvccpb"
+
+ "google.golang.org/grpc"
+ "google.golang.org/grpc/codes"
+ "google.golang.org/grpc/metadata"
+ "google.golang.org/grpc/status"
+)
+
+const (
+ EventTypeDelete = mvccpb.DELETE
+ EventTypePut = mvccpb.PUT
+
+ closeSendErrTimeout = 250 * time.Millisecond
+)
+
+type Event mvccpb.Event
+
+type WatchChan <-chan WatchResponse
+
+type Watcher interface {
+ // Watch watches on a key or prefix. The watched events will be returned
+ // through the returned channel. If revisions waiting to be sent over the
+ // watch are compacted, then the watch will be canceled by the server, the
+ // client will post a compacted error watch response, and the channel will close.
+ Watch(ctx context.Context, key string, opts ...OpOption) WatchChan
+
+ // Close closes the watcher and cancels all watch requests.
+ Close() error
+}
+
+type WatchResponse struct {
+ Header pb.ResponseHeader
+ Events []*Event
+
+ // CompactRevision is the minimum revision the watcher may receive.
+ CompactRevision int64
+
+ // Canceled is used to indicate watch failure.
+ // If the watch failed and the stream was about to close, before the channel is closed,
+ // the channel sends a final response that has Canceled set to true with a non-nil Err().
+ Canceled bool
+
+ // Created is used to indicate the creation of the watcher.
+ Created bool
+
+ closeErr error
+
+ // cancelReason is a reason of canceling watch
+ cancelReason string
+}
+
+// IsCreate returns true if the event tells that the key is newly created.
+func (e *Event) IsCreate() bool {
+ return e.Type == EventTypePut && e.Kv.CreateRevision == e.Kv.ModRevision
+}
+
+// IsModify returns true if the event tells that a new value is put on existing key.
+func (e *Event) IsModify() bool {
+ return e.Type == EventTypePut && e.Kv.CreateRevision != e.Kv.ModRevision
+}
+
+// Err is the error value if this WatchResponse holds an error.
+func (wr *WatchResponse) Err() error {
+ switch {
+ case wr.closeErr != nil:
+ return v3rpc.Error(wr.closeErr)
+ case wr.CompactRevision != 0:
+ return v3rpc.ErrCompacted
+ case wr.Canceled:
+ if len(wr.cancelReason) != 0 {
+ return v3rpc.Error(status.Error(codes.FailedPrecondition, wr.cancelReason))
+ }
+ return v3rpc.ErrFutureRev
+ }
+ return nil
+}
+
+// IsProgressNotify returns true if the WatchResponse is progress notification.
+func (wr *WatchResponse) IsProgressNotify() bool {
+ return len(wr.Events) == 0 && !wr.Canceled && !wr.Created && wr.CompactRevision == 0 && wr.Header.Revision != 0
+}
+
+// watcher implements the Watcher interface
+type watcher struct {
+ remote pb.WatchClient
+ callOpts []grpc.CallOption
+
+ // mu protects the grpc streams map
+ mu sync.RWMutex
+
+ // streams holds all the active grpc streams keyed by ctx value.
+ streams map[string]*watchGrpcStream
+}
+
+// watchGrpcStream tracks all watch resources attached to a single grpc stream.
+type watchGrpcStream struct {
+ owner *watcher
+ remote pb.WatchClient
+ callOpts []grpc.CallOption
+
+ // ctx controls internal remote.Watch requests
+ ctx context.Context
+ // ctxKey is the key used when looking up this stream's context
+ ctxKey string
+ cancel context.CancelFunc
+
+ // substreams holds all active watchers on this grpc stream
+ substreams map[int64]*watcherStream
+ // resuming holds all resuming watchers on this grpc stream
+ resuming []*watcherStream
+
+ // reqc sends a watch request from Watch() to the main goroutine
+ reqc chan *watchRequest
+ // respc receives data from the watch client
+ respc chan *pb.WatchResponse
+ // donec closes to broadcast shutdown
+ donec chan struct{}
+ // errc transmits errors from grpc Recv to the watch stream reconnect logic
+ errc chan error
+ // closingc gets the watcherStream of closing watchers
+ closingc chan *watcherStream
+ // wg is Done when all substream goroutines have exited
+ wg sync.WaitGroup
+
+ // resumec closes to signal that all substreams should begin resuming
+ resumec chan struct{}
+ // closeErr is the error that closed the watch stream
+ closeErr error
+}
+
+// watchRequest is issued by the subscriber to start a new watcher
+type watchRequest struct {
+ ctx context.Context
+ key string
+ end string
+ rev int64
+ // send created notification event if this field is true
+ createdNotify bool
+ // progressNotify is for progress updates
+ progressNotify bool
+ // filters is the list of events to filter out
+ filters []pb.WatchCreateRequest_FilterType
+ // get the previous key-value pair before the event happens
+ prevKV bool
+ // retc receives a chan WatchResponse once the watcher is established
+ retc chan chan WatchResponse
+}
+
+// watcherStream represents a registered watcher
+type watcherStream struct {
+ // initReq is the request that initiated this request
+ initReq watchRequest
+
+ // outc publishes watch responses to subscriber
+ outc chan WatchResponse
+ // recvc buffers watch responses before publishing
+ recvc chan *WatchResponse
+ // donec closes when the watcherStream goroutine stops.
+ donec chan struct{}
+ // closing is set to true when stream should be scheduled to shutdown.
+ closing bool
+ // id is the registered watch id on the grpc stream
+ id int64
+
+ // buf holds all events received from etcd but not yet consumed by the client
+ buf []*WatchResponse
+}
+
+func NewWatcher(c *Client) Watcher {
+ return NewWatchFromWatchClient(pb.NewWatchClient(c.conn), c)
+}
+
+func NewWatchFromWatchClient(wc pb.WatchClient, c *Client) Watcher {
+ w := &watcher{
+ remote: wc,
+ streams: make(map[string]*watchGrpcStream),
+ }
+ if c != nil {
+ w.callOpts = c.callOpts
+ }
+ return w
+}
+
+// never closes
+var valCtxCh = make(chan struct{})
+var zeroTime = time.Unix(0, 0)
+
+// ctx with only the values; never Done
+type valCtx struct{ context.Context }
+
+func (vc *valCtx) Deadline() (time.Time, bool) { return zeroTime, false }
+func (vc *valCtx) Done() <-chan struct{} { return valCtxCh }
+func (vc *valCtx) Err() error { return nil }
+
+func (w *watcher) newWatcherGrpcStream(inctx context.Context) *watchGrpcStream {
+ ctx, cancel := context.WithCancel(&valCtx{inctx})
+ wgs := &watchGrpcStream{
+ owner: w,
+ remote: w.remote,
+ callOpts: w.callOpts,
+ ctx: ctx,
+ ctxKey: streamKeyFromCtx(inctx),
+ cancel: cancel,
+ substreams: make(map[int64]*watcherStream),
+ respc: make(chan *pb.WatchResponse),
+ reqc: make(chan *watchRequest),
+ donec: make(chan struct{}),
+ errc: make(chan error, 1),
+ closingc: make(chan *watcherStream),
+ resumec: make(chan struct{}),
+ }
+ go wgs.run()
+ return wgs
+}
+
+// Watch posts a watch request to run() and waits for a new watcher channel
+func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) WatchChan {
+ ow := opWatch(key, opts...)
+
+ var filters []pb.WatchCreateRequest_FilterType
+ if ow.filterPut {
+ filters = append(filters, pb.WatchCreateRequest_NOPUT)
+ }
+ if ow.filterDelete {
+ filters = append(filters, pb.WatchCreateRequest_NODELETE)
+ }
+
+ wr := &watchRequest{
+ ctx: ctx,
+ createdNotify: ow.createdNotify,
+ key: string(ow.key),
+ end: string(ow.end),
+ rev: ow.rev,
+ progressNotify: ow.progressNotify,
+ filters: filters,
+ prevKV: ow.prevKV,
+ retc: make(chan chan WatchResponse, 1),
+ }
+
+ ok := false
+ ctxKey := streamKeyFromCtx(ctx)
+
+ // find or allocate appropriate grpc watch stream
+ w.mu.Lock()
+ if w.streams == nil {
+ // closed
+ w.mu.Unlock()
+ ch := make(chan WatchResponse)
+ close(ch)
+ return ch
+ }
+ wgs := w.streams[ctxKey]
+ if wgs == nil {
+ wgs = w.newWatcherGrpcStream(ctx)
+ w.streams[ctxKey] = wgs
+ }
+ donec := wgs.donec
+ reqc := wgs.reqc
+ w.mu.Unlock()
+
+ // couldn't create channel; return closed channel
+ closeCh := make(chan WatchResponse, 1)
+
+ // submit request
+ select {
+ case reqc <- wr:
+ ok = true
+ case <-wr.ctx.Done():
+ case <-donec:
+ if wgs.closeErr != nil {
+ closeCh <- WatchResponse{closeErr: wgs.closeErr}
+ break
+ }
+ // retry; may have dropped stream from no ctxs
+ return w.Watch(ctx, key, opts...)
+ }
+
+ // receive channel
+ if ok {
+ select {
+ case ret := <-wr.retc:
+ return ret
+ case <-ctx.Done():
+ case <-donec:
+ if wgs.closeErr != nil {
+ closeCh <- WatchResponse{closeErr: wgs.closeErr}
+ break
+ }
+ // retry; may have dropped stream from no ctxs
+ return w.Watch(ctx, key, opts...)
+ }
+ }
+
+ close(closeCh)
+ return closeCh
+}
+
+func (w *watcher) Close() (err error) {
+ w.mu.Lock()
+ streams := w.streams
+ w.streams = nil
+ w.mu.Unlock()
+ for _, wgs := range streams {
+ if werr := wgs.close(); werr != nil {
+ err = werr
+ }
+ }
+ return err
+}
+
+func (w *watchGrpcStream) close() (err error) {
+ w.cancel()
+ <-w.donec
+ select {
+ case err = <-w.errc:
+ default:
+ }
+ return toErr(w.ctx, err)
+}
+
+func (w *watcher) closeStream(wgs *watchGrpcStream) {
+ w.mu.Lock()
+ close(wgs.donec)
+ wgs.cancel()
+ if w.streams != nil {
+ delete(w.streams, wgs.ctxKey)
+ }
+ w.mu.Unlock()
+}
+
+func (w *watchGrpcStream) addSubstream(resp *pb.WatchResponse, ws *watcherStream) {
+ if resp.WatchId == -1 {
+ // failed; no channel
+ close(ws.recvc)
+ return
+ }
+ ws.id = resp.WatchId
+ w.substreams[ws.id] = ws
+}
+
+func (w *watchGrpcStream) sendCloseSubstream(ws *watcherStream, resp *WatchResponse) {
+ select {
+ case ws.outc <- *resp:
+ case <-ws.initReq.ctx.Done():
+ case <-time.After(closeSendErrTimeout):
+ }
+ close(ws.outc)
+}
+
+func (w *watchGrpcStream) closeSubstream(ws *watcherStream) {
+ // send channel response in case stream was never established
+ select {
+ case ws.initReq.retc <- ws.outc:
+ default:
+ }
+ // close subscriber's channel
+ if closeErr := w.closeErr; closeErr != nil && ws.initReq.ctx.Err() == nil {
+ go w.sendCloseSubstream(ws, &WatchResponse{closeErr: w.closeErr})
+ } else if ws.outc != nil {
+ close(ws.outc)
+ }
+ if ws.id != -1 {
+ delete(w.substreams, ws.id)
+ return
+ }
+ for i := range w.resuming {
+ if w.resuming[i] == ws {
+ w.resuming[i] = nil
+ return
+ }
+ }
+}
+
+// run is the root of the goroutines for managing a watcher client
+func (w *watchGrpcStream) run() {
+ var wc pb.Watch_WatchClient
+ var closeErr error
+
+ // substreams marked to close but goroutine still running; needed for
+ // avoiding double-closing recvc on grpc stream teardown
+ closing := make(map[*watcherStream]struct{})
+
+ defer func() {
+ w.closeErr = closeErr
+ // shutdown substreams and resuming substreams
+ for _, ws := range w.substreams {
+ if _, ok := closing[ws]; !ok {
+ close(ws.recvc)
+ closing[ws] = struct{}{}
+ }
+ }
+ for _, ws := range w.resuming {
+ if _, ok := closing[ws]; ws != nil && !ok {
+ close(ws.recvc)
+ closing[ws] = struct{}{}
+ }
+ }
+ w.joinSubstreams()
+ for range closing {
+ w.closeSubstream(<-w.closingc)
+ }
+ w.wg.Wait()
+ w.owner.closeStream(w)
+ }()
+
+ // start a stream with the etcd grpc server
+ if wc, closeErr = w.newWatchClient(); closeErr != nil {
+ return
+ }
+
+ cancelSet := make(map[int64]struct{})
+
+ for {
+ select {
+ // Watch() requested
+ case wreq := <-w.reqc:
+ outc := make(chan WatchResponse, 1)
+ ws := &watcherStream{
+ initReq: *wreq,
+ id: -1,
+ outc: outc,
+ // unbuffered so resumes won't cause repeat events
+ recvc: make(chan *WatchResponse),
+ }
+
+ ws.donec = make(chan struct{})
+ w.wg.Add(1)
+ go w.serveSubstream(ws, w.resumec)
+
+ // queue up for watcher creation/resume
+ w.resuming = append(w.resuming, ws)
+ if len(w.resuming) == 1 {
+ // head of resume queue, can register a new watcher
+ wc.Send(ws.initReq.toPB())
+ }
+ // New events from the watch client
+ case pbresp := <-w.respc:
+ switch {
+ case pbresp.Created:
+ // response to head of queue creation
+ if ws := w.resuming[0]; ws != nil {
+ w.addSubstream(pbresp, ws)
+ w.dispatchEvent(pbresp)
+ w.resuming[0] = nil
+ }
+ if ws := w.nextResume(); ws != nil {
+ wc.Send(ws.initReq.toPB())
+ }
+ case pbresp.Canceled && pbresp.CompactRevision == 0:
+ delete(cancelSet, pbresp.WatchId)
+ if ws, ok := w.substreams[pbresp.WatchId]; ok {
+ // signal to stream goroutine to update closingc
+ close(ws.recvc)
+ closing[ws] = struct{}{}
+ }
+ default:
+ // dispatch to appropriate watch stream
+ if ok := w.dispatchEvent(pbresp); ok {
+ break
+ }
+ // watch response on unexpected watch id; cancel id
+ if _, ok := cancelSet[pbresp.WatchId]; ok {
+ break
+ }
+ cancelSet[pbresp.WatchId] = struct{}{}
+ cr := &pb.WatchRequest_CancelRequest{
+ CancelRequest: &pb.WatchCancelRequest{
+ WatchId: pbresp.WatchId,
+ },
+ }
+ req := &pb.WatchRequest{RequestUnion: cr}
+ wc.Send(req)
+ }
+ // watch client failed on Recv; spawn another if possible
+ case err := <-w.errc:
+ if isHaltErr(w.ctx, err) || toErr(w.ctx, err) == v3rpc.ErrNoLeader {
+ closeErr = err
+ return
+ }
+ if wc, closeErr = w.newWatchClient(); closeErr != nil {
+ return
+ }
+ if ws := w.nextResume(); ws != nil {
+ wc.Send(ws.initReq.toPB())
+ }
+ cancelSet = make(map[int64]struct{})
+ case <-w.ctx.Done():
+ return
+ case ws := <-w.closingc:
+ w.closeSubstream(ws)
+ delete(closing, ws)
+ if len(w.substreams)+len(w.resuming) == 0 {
+ // no more watchers on this stream, shutdown
+ return
+ }
+ }
+ }
+}
+
+// nextResume chooses the next resuming to register with the grpc stream. Abandoned
+// streams are marked as nil in the queue since the head must wait for its inflight registration.
+func (w *watchGrpcStream) nextResume() *watcherStream {
+ for len(w.resuming) != 0 {
+ if w.resuming[0] != nil {
+ return w.resuming[0]
+ }
+ w.resuming = w.resuming[1:len(w.resuming)]
+ }
+ return nil
+}
+
+// dispatchEvent sends a WatchResponse to the appropriate watcher stream
+func (w *watchGrpcStream) dispatchEvent(pbresp *pb.WatchResponse) bool {
+ events := make([]*Event, len(pbresp.Events))
+ for i, ev := range pbresp.Events {
+ events[i] = (*Event)(ev)
+ }
+ wr := &WatchResponse{
+ Header: *pbresp.Header,
+ Events: events,
+ CompactRevision: pbresp.CompactRevision,
+ Created: pbresp.Created,
+ Canceled: pbresp.Canceled,
+ cancelReason: pbresp.CancelReason,
+ }
+ ws, ok := w.substreams[pbresp.WatchId]
+ if !ok {
+ return false
+ }
+ select {
+ case ws.recvc <- wr:
+ case <-ws.donec:
+ return false
+ }
+ return true
+}
+
+// serveWatchClient forwards messages from the grpc stream to run()
+func (w *watchGrpcStream) serveWatchClient(wc pb.Watch_WatchClient) {
+ for {
+ resp, err := wc.Recv()
+ if err != nil {
+ select {
+ case w.errc <- err:
+ case <-w.donec:
+ }
+ return
+ }
+ select {
+ case w.respc <- resp:
+ case <-w.donec:
+ return
+ }
+ }
+}
+
+// serveSubstream forwards watch responses from run() to the subscriber
+func (w *watchGrpcStream) serveSubstream(ws *watcherStream, resumec chan struct{}) {
+ if ws.closing {
+ panic("created substream goroutine but substream is closing")
+ }
+
+ // nextRev is the minimum expected next revision
+ nextRev := ws.initReq.rev
+ resuming := false
+ defer func() {
+ if !resuming {
+ ws.closing = true
+ }
+ close(ws.donec)
+ if !resuming {
+ w.closingc <- ws
+ }
+ w.wg.Done()
+ }()
+
+ emptyWr := &WatchResponse{}
+ for {
+ curWr := emptyWr
+ outc := ws.outc
+
+ if len(ws.buf) > 0 {
+ curWr = ws.buf[0]
+ } else {
+ outc = nil
+ }
+ select {
+ case outc <- *curWr:
+ if ws.buf[0].Err() != nil {
+ return
+ }
+ ws.buf[0] = nil
+ ws.buf = ws.buf[1:]
+ case wr, ok := <-ws.recvc:
+ if !ok {
+ // shutdown from closeSubstream
+ return
+ }
+
+ if wr.Created {
+ if ws.initReq.retc != nil {
+ ws.initReq.retc <- ws.outc
+ // to prevent next write from taking the slot in buffered channel
+ // and posting duplicate create events
+ ws.initReq.retc = nil
+
+ // send first creation event only if requested
+ if ws.initReq.createdNotify {
+ ws.outc <- *wr
+ }
+ // once the watch channel is returned, a current revision
+ // watch must resume at the store revision. This is necessary
+ // for the following case to work as expected:
+ // wch := m1.Watch("a")
+ // m2.Put("a", "b")
+ // <-wch
+ // If the revision is only bound on the first observed event,
+ // if wch is disconnected before the Put is issued, then reconnects
+ // after it is committed, it'll miss the Put.
+ if ws.initReq.rev == 0 {
+ nextRev = wr.Header.Revision
+ }
+ }
+ } else {
+ // current progress of watch; <= store revision
+ nextRev = wr.Header.Revision
+ }
+
+ if len(wr.Events) > 0 {
+ nextRev = wr.Events[len(wr.Events)-1].Kv.ModRevision + 1
+ }
+ ws.initReq.rev = nextRev
+
+ // created event is already sent above,
+ // watcher should not post duplicate events
+ if wr.Created {
+ continue
+ }
+
+ // TODO pause channel if buffer gets too large
+ ws.buf = append(ws.buf, wr)
+ case <-w.ctx.Done():
+ return
+ case <-ws.initReq.ctx.Done():
+ return
+ case <-resumec:
+ resuming = true
+ return
+ }
+ }
+ // lazily send cancel message if events on missing id
+}
+
+func (w *watchGrpcStream) newWatchClient() (pb.Watch_WatchClient, error) {
+ // mark all substreams as resuming
+ close(w.resumec)
+ w.resumec = make(chan struct{})
+ w.joinSubstreams()
+ for _, ws := range w.substreams {
+ ws.id = -1
+ w.resuming = append(w.resuming, ws)
+ }
+ // strip out nils, if any
+ var resuming []*watcherStream
+ for _, ws := range w.resuming {
+ if ws != nil {
+ resuming = append(resuming, ws)
+ }
+ }
+ w.resuming = resuming
+ w.substreams = make(map[int64]*watcherStream)
+
+ // connect to grpc stream while accepting watcher cancelation
+ stopc := make(chan struct{})
+ donec := w.waitCancelSubstreams(stopc)
+ wc, err := w.openWatchClient()
+ close(stopc)
+ <-donec
+
+ // serve all non-closing streams, even if there's a client error
+ // so that the teardown path can shutdown the streams as expected.
+ for _, ws := range w.resuming {
+ if ws.closing {
+ continue
+ }
+ ws.donec = make(chan struct{})
+ w.wg.Add(1)
+ go w.serveSubstream(ws, w.resumec)
+ }
+
+ if err != nil {
+ return nil, v3rpc.Error(err)
+ }
+
+ // receive data from new grpc stream
+ go w.serveWatchClient(wc)
+ return wc, nil
+}
+
+func (w *watchGrpcStream) waitCancelSubstreams(stopc <-chan struct{}) <-chan struct{} {
+ var wg sync.WaitGroup
+ wg.Add(len(w.resuming))
+ donec := make(chan struct{})
+ for i := range w.resuming {
+ go func(ws *watcherStream) {
+ defer wg.Done()
+ if ws.closing {
+ if ws.initReq.ctx.Err() != nil && ws.outc != nil {
+ close(ws.outc)
+ ws.outc = nil
+ }
+ return
+ }
+ select {
+ case <-ws.initReq.ctx.Done():
+ // closed ws will be removed from resuming
+ ws.closing = true
+ close(ws.outc)
+ ws.outc = nil
+ w.wg.Add(1)
+ go func() {
+ defer w.wg.Done()
+ w.closingc <- ws
+ }()
+ case <-stopc:
+ }
+ }(w.resuming[i])
+ }
+ go func() {
+ defer close(donec)
+ wg.Wait()
+ }()
+ return donec
+}
+
+// joinSubstreams waits for all substream goroutines to complete.
+func (w *watchGrpcStream) joinSubstreams() {
+ for _, ws := range w.substreams {
+ <-ws.donec
+ }
+ for _, ws := range w.resuming {
+ if ws != nil {
+ <-ws.donec
+ }
+ }
+}
+
+var maxBackoff = 100 * time.Millisecond
+
+// openWatchClient retries opening a watch client until success or halt.
+// manually retry in case "ws==nil && err==nil"
+// TODO: remove FailFast=false
+func (w *watchGrpcStream) openWatchClient() (ws pb.Watch_WatchClient, err error) {
+ backoff := time.Millisecond
+ for {
+ select {
+ case <-w.ctx.Done():
+ if err == nil {
+ return nil, w.ctx.Err()
+ }
+ return nil, err
+ default:
+ }
+ if ws, err = w.remote.Watch(w.ctx, w.callOpts...); ws != nil && err == nil {
+ break
+ }
+ if isHaltErr(w.ctx, err) {
+ return nil, v3rpc.Error(err)
+ }
+ if isUnavailableErr(w.ctx, err) {
+ // retry, but backoff
+ if backoff < maxBackoff {
+ // 25% backoff factor
+ backoff = backoff + backoff/4
+ if backoff > maxBackoff {
+ backoff = maxBackoff
+ }
+ }
+ time.Sleep(backoff)
+ }
+ }
+ return ws, nil
+}
+
+// toPB converts an internal watch request structure to its protobuf WatchRequest structure.
+func (wr *watchRequest) toPB() *pb.WatchRequest {
+ req := &pb.WatchCreateRequest{
+ StartRevision: wr.rev,
+ Key: []byte(wr.key),
+ RangeEnd: []byte(wr.end),
+ ProgressNotify: wr.progressNotify,
+ Filters: wr.filters,
+ PrevKv: wr.prevKV,
+ }
+ cr := &pb.WatchRequest_CreateRequest{CreateRequest: req}
+ return &pb.WatchRequest{RequestUnion: cr}
+}
+
+func streamKeyFromCtx(ctx context.Context) string {
+ if md, ok := metadata.FromOutgoingContext(ctx); ok {
+ return fmt.Sprintf("%+v", md)
+ }
+ return ""
+}