switch from godep to glide

vendor/github.com/coreos/etcd/clientv3/README.md

@@ -1,77 +0,0 @@
-# etcd/clientv3
-
-[![Godoc](http://img.shields.io/badge/go-documentation-blue.svg?style=flat-square)](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 := kvc.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 := kvc.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)
-	}
-}
-```
-
-## Examples
-
-More code examples can be found at [GoDoc](https://godoc.org/github.com/coreos/etcd/clientv3).

vendor/github.com/coreos/etcd/clientv3/auth.go

@@ -43,6 +43,7 @@ type (
 	AuthRoleListResponse             pb.AuthRoleListResponse
 
 	PermissionType authpb.Permission_Type
+	Permission     authpb.Permission
 )
 
 const (

vendor/github.com/coreos/etcd/clientv3/balancer.go

@@ -42,9 +42,16 @@ type simpleBalancer struct {
 	// upc closes when upEps transitions from empty to non-zero or the balancer closes.
 	upc chan struct{}
 
+	// 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.
+	host2ep map[string]string
+
 	// pinAddr is the currently pinned address; set to the empty string on
 	// intialization and shutdown.
 	pinAddr string
+
+	closed bool
 }
 
 func newSimpleBalancer(eps []string) *simpleBalancer {
@@ -60,11 +67,12 @@ func newSimpleBalancer(eps []string) *simpleBalancer {
 		readyc:   make(chan struct{}),
 		upEps:    make(map[string]struct{}),
 		upc:      make(chan struct{}),
+		host2ep:  getHost2ep(eps),
 	}
 	return sb
 }
 
-func (b *simpleBalancer) Start(target string) error { return nil }
+func (b *simpleBalancer) Start(target string, config grpc.BalancerConfig) error { return nil }
 
 func (b *simpleBalancer) ConnectNotify() <-chan struct{} {
 	b.mu.Lock()
@@ -72,17 +80,70 @@ func (b *simpleBalancer) ConnectNotify() <-chan struct{} {
 	return b.upc
 }
 
+func (b *simpleBalancer) getEndpoint(host string) string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.host2ep[host]
+}
+
+func getHost2ep(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
+}
+
+func (b *simpleBalancer) updateAddrs(eps []string) {
+	np := getHost2ep(eps)
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	match := len(np) == len(b.host2ep)
+	for k, v := range np {
+		if b.host2ep[k] != v {
+			match = false
+			break
+		}
+	}
+	if match {
+		// same endpoints, so no need to update address
+		return
+	}
+
+	b.host2ep = np
+
+	addrs := make([]grpc.Address, 0, len(eps))
+	for i := range eps {
+		addrs = append(addrs, grpc.Address{Addr: getHost(eps[i])})
+	}
+	b.addrs = addrs
+	b.notifyCh <- addrs
+}
+
 func (b *simpleBalancer) Up(addr grpc.Address) func(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. Or our simplerBalancer
+	// might panic since b.upc is closed.
+	if b.closed {
+		return func(err error) {}
+	}
+
 	if len(b.upEps) == 0 {
 		// notify waiting Get()s and pin first connected address
 		close(b.upc)
 		b.pinAddr = addr.Addr
 	}
 	b.upEps[addr.Addr] = struct{}{}
-	b.mu.Unlock()
+
 	// notify client that a connection is up
 	b.readyOnce.Do(func() { close(b.readyc) })
+
 	return func(err error) {
 		b.mu.Lock()
 		delete(b.upEps, addr.Addr)
@@ -128,13 +189,19 @@ func (b *simpleBalancer) Notify() <-chan []grpc.Address { return b.notifyCh }
 
 func (b *simpleBalancer) Close() error {
 	b.mu.Lock()
+	defer b.mu.Unlock()
+	// In case gRPC calls close twice. TODO: remove the checking
+	// when we are sure that gRPC wont call close twice.
+	if b.closed {
+		return nil
+	}
+	b.closed = true
 	close(b.notifyCh)
 	// terminate all waiting Get()s
 	b.pinAddr = ""
 	if len(b.upEps) == 0 {
 		close(b.upc)
 	}
-	b.mu.Unlock()
 	return nil
 }
 

vendor/github.com/coreos/etcd/clientv3/client.go

@@ -18,8 +18,6 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
-	"io/ioutil"
-	"log"
 	"net"
 	"net/url"
 	"strings"
@@ -29,6 +27,7 @@ import (
 
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/metadata"
 )
@@ -98,6 +97,44 @@ func (c *Client) Ctx() context.Context { return c.ctx }
 // Endpoints lists the registered endpoints for the client.
 func (c *Client) Endpoints() []string { return c.cfg.Endpoints }
 
+// SetEndpoints updates client's endpoints.
+func (c *Client) SetEndpoints(eps ...string) {
+	c.cfg.Endpoints = eps
+	c.balancer.updateAddrs(eps)
+}
+
+// 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, _ := context.WithTimeout(c.ctx, 5*time.Second)
+			if err := c.Sync(ctx); err != nil && err != c.ctx.Err() {
+				logger.Println("Auto sync endpoints failed:", err)
+			}
+		}
+	}
+}
+
 type authTokenCredential struct {
 	token string
 }
@@ -112,19 +149,31 @@ func (cred authTokenCredential) GetRequestMetadata(ctx context.Context, s ...str
 	}, nil
 }
 
-func (c *Client) dialTarget(endpoint string) (proto string, host string, creds *credentials.TransportCredentials) {
+func parseEndpoint(endpoint string) (proto string, host string, scheme string) {
 	proto = "tcp"
 	host = endpoint
-	creds = c.creds
 	url, uerr := url.Parse(endpoint)
 	if uerr != nil || !strings.Contains(endpoint, "://") {
 		return
 	}
+	scheme = url.Scheme
+
 	// strip scheme:// prefix since grpc dials by host
 	host = url.Host
 	switch url.Scheme {
+	case "http", "https":
 	case "unix":
 		proto = "unix"
+	default:
+		proto, host = "", ""
+	}
+	return
+}
+
+func (c *Client) processCreds(scheme string) (creds *credentials.TransportCredentials) {
+	creds = c.creds
+	switch scheme {
+	case "unix":
 	case "http":
 		creds = nil
 	case "https":
@@ -135,7 +184,7 @@ func (c *Client) dialTarget(endpoint string) (proto string, host string, creds *
 		emptyCreds := credentials.NewTLS(tlsconfig)
 		creds = &emptyCreds
 	default:
-		return "", "", nil
+		creds = nil
 	}
 	return
 }
@@ -147,17 +196,8 @@ func (c *Client) dialSetupOpts(endpoint string, dopts ...grpc.DialOption) (opts
 	}
 	opts = append(opts, dopts...)
 
-	// 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.
-	host2ep := make(map[string]string)
-	for i := range c.cfg.Endpoints {
-		_, host, _ := c.dialTarget(c.cfg.Endpoints[i])
-		host2ep[host] = c.cfg.Endpoints[i]
-	}
-
 	f := func(host string, t time.Duration) (net.Conn, error) {
-		proto, host, _ := c.dialTarget(host2ep[host])
+		proto, host, _ := parseEndpoint(c.balancer.getEndpoint(host))
 		if proto == "" {
 			return nil, fmt.Errorf("unknown scheme for %q", host)
 		}
@@ -170,7 +210,10 @@ func (c *Client) dialSetupOpts(endpoint string, dopts ...grpc.DialOption) (opts
 	}
 	opts = append(opts, grpc.WithDialer(f))
 
-	_, _, creds := c.dialTarget(endpoint)
+	creds := c.creds
+	if _, _, scheme := parseEndpoint(endpoint); len(scheme) != 0 {
+		creds = c.processCreds(scheme)
+	}
 	if creds != nil {
 		opts = append(opts, grpc.WithTransportCredentials(*creds))
 	} else {
@@ -272,13 +315,8 @@ func newClient(cfg *Config) (*Client, error) {
 	client.Watcher = NewWatcher(client)
 	client.Auth = NewAuth(client)
 	client.Maintenance = NewMaintenance(client)
-	if cfg.Logger != nil {
-		logger.Set(cfg.Logger)
-	} else {
-		// disable client side grpc by default
-		logger.Set(log.New(ioutil.Discard, "", 0))
-	}
 
+	go client.autoSync()
 	return client, nil
 }
 
@@ -294,17 +332,14 @@ func isHaltErr(ctx context.Context, err error) bool {
 	if err == nil {
 		return false
 	}
-	eErr := rpctypes.Error(err)
-	if _, ok := eErr.(rpctypes.EtcdError); ok {
-		return eErr != rpctypes.ErrStopped && eErr != rpctypes.ErrNoLeader
-	}
-	// treat etcdserver errors not recognized by the client as halting
-	return isConnClosing(err) || strings.Contains(err.Error(), "etcdserver:")
-}
-
-// isConnClosing returns true if the error matches a grpc client closing error
-func isConnClosing(err error) bool {
-	return strings.Contains(err.Error(), grpc.ErrClientConnClosing.Error())
+	code := grpc.Code(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 code != codes.Unavailable && code != codes.Internal
 }
 
 func toErr(ctx context.Context, err error) error {
@@ -312,12 +347,20 @@ func toErr(ctx context.Context, err error) error {
 		return nil
 	}
 	err = rpctypes.Error(err)
-	switch {
-	case ctx.Err() != nil && strings.Contains(err.Error(), "context"):
-		err = ctx.Err()
-	case strings.Contains(err.Error(), ErrNoAvailableEndpoints.Error()):
+	if _, ok := err.(rpctypes.EtcdError); ok {
+		return err
+	}
+	code := grpc.Code(err)
+	switch code {
+	case codes.DeadlineExceeded:
+		fallthrough
+	case codes.Canceled:
+		if ctx.Err() != nil {
+			err = ctx.Err()
+		}
+	case codes.Unavailable:
 		err = ErrNoAvailableEndpoints
-	case strings.Contains(err.Error(), grpc.ErrClientConnClosing.Error()):
+	case codes.FailedPrecondition:
 		err = grpc.ErrClientConnClosing
 	}
 	return err

vendor/github.com/coreos/etcd/clientv3/config.go

@@ -28,15 +28,16 @@ type Config struct {
 	// Endpoints is a list of URLs
 	Endpoints []string
 
+	// AutoSyncInterval is the interval to update endpoints with its latest members.
+	// 0 disables auto-sync. By default auto-sync is disabled.
+	AutoSyncInterval time.Duration
+
 	// DialTimeout is the timeout for failing to establish a connection.
 	DialTimeout time.Duration
 
 	// TLS holds the client secure credentials, if any.
 	TLS *tls.Config
 
-	// Logger is the logger used by client library.
-	Logger Logger
-
 	// Username is a username for authentication
 	Username string
 
@@ -46,6 +47,7 @@ type Config struct {
 
 type yamlConfig struct {
 	Endpoints             []string      `json:"endpoints"`
+	AutoSyncInterval      time.Duration `json:"auto-sync-interval"`
 	DialTimeout           time.Duration `json:"dial-timeout"`
 	InsecureTransport     bool          `json:"insecure-transport"`
 	InsecureSkipTLSVerify bool          `json:"insecure-skip-tls-verify"`
@@ -68,8 +70,9 @@ func configFromFile(fpath string) (*Config, error) {
 	}
 
 	cfg := &Config{
-		Endpoints:   yc.Endpoints,
-		DialTimeout: yc.DialTimeout,
+		Endpoints:        yc.Endpoints,
+		AutoSyncInterval: yc.AutoSyncInterval,
+		DialTimeout:      yc.DialTimeout,
 	}
 
 	if yc.InsecureTransport {

vendor/github.com/coreos/etcd/clientv3/kv.go

@@ -85,6 +85,10 @@ func NewKV(c *Client) KV {
 	return &kv{remote: RetryKVClient(c)}
 }
 
+func NewKVFromKVClient(remote pb.KVClient) KV {
+	return &kv{remote: remote}
+}
+
 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)
@@ -137,34 +141,20 @@ func (kv *kv) do(ctx context.Context, op Op) (OpResponse, error) {
 	// TODO: handle other ops
 	case tRange:
 		var resp *pb.RangeResponse
-		r := &pb.RangeRequest{
-			Key:          op.key,
-			RangeEnd:     op.end,
-			Limit:        op.limit,
-			Revision:     op.rev,
-			Serializable: op.serializable,
-			KeysOnly:     op.keysOnly,
-			CountOnly:    op.countOnly,
-		}
-		if op.sort != nil {
-			r.SortOrder = pb.RangeRequest_SortOrder(op.sort.Order)
-			r.SortTarget = pb.RangeRequest_SortTarget(op.sort.Target)
-		}
-
-		resp, err = kv.remote.Range(ctx, r, grpc.FailFast(false))
+		resp, err = kv.remote.Range(ctx, op.toRangeRequest(), grpc.FailFast(false))
 		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)}
+		r := &pb.PutRequest{Key: op.key, Value: op.val, Lease: int64(op.leaseID), PrevKv: op.prevKV}
 		resp, err = kv.remote.Put(ctx, r)
 		if err == nil {
 			return OpResponse{put: (*PutResponse)(resp)}, nil
 		}
 	case tDeleteRange:
 		var resp *pb.DeleteRangeResponse
-		r := &pb.DeleteRangeRequest{Key: op.key, RangeEnd: op.end}
+		r := &pb.DeleteRangeRequest{Key: op.key, RangeEnd: op.end, PrevKv: op.prevKV}
 		resp, err = kv.remote.DeleteRange(ctx, r)
 		if err == nil {
 			return OpResponse{del: (*DeleteResponse)(resp)}, nil

vendor/github.com/coreos/etcd/clientv3/lease.go

@@ -44,6 +44,21 @@ type LeaseKeepAliveResponse struct {
 	TTL int64
 }
 
+// LeaseTimeToLiveResponse is used to convert the protobuf lease timetolive response.
+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.
+	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"`
+}
+
 const (
 	// defaultTTL is the assumed lease TTL used for the first keepalive
 	// deadline before the actual TTL is known to the client.
@@ -61,6 +76,9 @@ type Lease interface {
 	// 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)
+
 	// KeepAlive keeps the given lease alive forever.
 	KeepAlive(ctx context.Context, id LeaseID) (<-chan *LeaseKeepAliveResponse, error)
 
@@ -141,7 +159,7 @@ func (l *lessor) Grant(ctx context.Context, ttl int64) (*LeaseGrantResponse, err
 			return gresp, nil
 		}
 		if isHaltErr(cctx, err) {
-			return nil, toErr(ctx, err)
+			return nil, toErr(cctx, err)
 		}
 		if nerr := l.newStream(); nerr != nil {
 			return nil, nerr
@@ -170,6 +188,30 @@ func (l *lessor) Revoke(ctx context.Context, id LeaseID) (*LeaseRevokeResponse,
 	}
 }
 
+func (l *lessor) TimeToLive(ctx context.Context, id LeaseID, opts ...LeaseOption) (*LeaseTimeToLiveResponse, error) {
+	cctx, cancel := context.WithCancel(ctx)
+	done := cancelWhenStop(cancel, l.stopCtx.Done())
+	defer close(done)
+
+	for {
+		r := toLeaseTimeToLiveRequest(id, opts...)
+		resp, err := l.remote.LeaseTimeToLive(cctx, r)
+		if err == nil {
+			gresp := &LeaseTimeToLiveResponse{
+				ResponseHeader: resp.GetHeader(),
+				ID:             LeaseID(resp.ID),
+				TTL:            resp.TTL,
+				GrantedTTL:     resp.GrantedTTL,
+				Keys:           resp.Keys,
+			}
+			return gresp, nil
+		}
+		if isHaltErr(cctx, err) {
+			return nil, toErr(cctx, err)
+		}
+	}
+}
+
 func (l *lessor) KeepAlive(ctx context.Context, id LeaseID) (<-chan *LeaseKeepAliveResponse, error) {
 	ch := make(chan *LeaseKeepAliveResponse, leaseResponseChSize)
 
@@ -390,7 +432,7 @@ func (l *lessor) sendKeepAliveLoop(stream pb.Lease_LeaseKeepAliveClient) {
 			return
 		}
 
-		tosend := make([]LeaseID, 0)
+		var tosend []LeaseID
 
 		now := time.Now()
 		l.mu.Lock()

vendor/github.com/coreos/etcd/clientv3/logger.go

@@ -15,13 +15,15 @@
 package clientv3
 
 import (
+	"io/ioutil"
 	"log"
-	"os"
 	"sync"
 
 	"google.golang.org/grpc/grpclog"
 )
 
+// Logger is the logger used by client library.
+// It implements grpclog.Logger interface.
 type Logger grpclog.Logger
 
 var (
@@ -34,20 +36,36 @@ type settableLogger struct {
 }
 
 func init() {
-	// use go's standard logger by default like grpc
+	// disable client side logs by default
 	logger.mu.Lock()
-	logger.l = log.New(os.Stderr, "", log.LstdFlags)
+	logger.l = log.New(ioutil.Discard, "", 0)
+
+	// logger has to override the grpclog at initialization so that
+	// any changes to the grpclog go through logger with locking
+	// instead of through SetLogger
+	//
+	// now updates only happen through settableLogger.set
 	grpclog.SetLogger(&logger)
 	logger.mu.Unlock()
 }
 
-func (s *settableLogger) Set(l Logger) {
+// SetLogger sets client-side Logger. By default, logs are disabled.
+func SetLogger(l Logger) {
+	logger.set(l)
+}
+
+// GetLogger returns the current logger.
+func GetLogger() Logger {
+	return logger.get()
+}
+
+func (s *settableLogger) set(l Logger) {
 	s.mu.Lock()
 	logger.l = l
 	s.mu.Unlock()
 }
 
-func (s *settableLogger) Get() Logger {
+func (s *settableLogger) get() Logger {
 	s.mu.RLock()
 	l := logger.l
 	s.mu.RUnlock()
@@ -56,9 +74,9 @@ func (s *settableLogger) Get() Logger {
 
 // implement the grpclog.Logger interface
 
-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().Print(args...) }
-func (s *settableLogger) Printf(format string, args ...interface{}) { s.Get().Printf(format, args...) }
-func (s *settableLogger) Println(args ...interface{})               { s.Get().Println(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().Print(args...) }
+func (s *settableLogger) Printf(format string, args ...interface{}) { s.get().Printf(format, args...) }
+func (s *settableLogger) Println(args ...interface{})               { s.get().Println(args...) }

vendor/github.com/coreos/etcd/clientv3/op.go

@@ -14,9 +14,7 @@
 
 package clientv3
 
-import (
-	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
-)
+import pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 
 type opType int
 
@@ -43,40 +41,63 @@ type Op struct {
 	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
+
 	// 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
 }
 
+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) toRequestOp() *pb.RequestOp {
 	switch op.t {
 	case 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,
-		}
-		if op.sort != nil {
-			r.SortOrder = pb.RangeRequest_SortOrder(op.sort.Order)
-			r.SortTarget = pb.RangeRequest_SortTarget(op.sort.Target)
-		}
-		return &pb.RequestOp{Request: &pb.RequestOp_RequestRange{RequestRange: r}}
+		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)}
+		r := &pb.PutRequest{Key: op.key, Value: op.val, Lease: int64(op.leaseID), PrevKv: op.prevKV}
 		return &pb.RequestOp{Request: &pb.RequestOp_RequestPut{RequestPut: r}}
 	case tDeleteRange:
-		r := &pb.DeleteRangeRequest{Key: op.key, RangeEnd: op.end}
+		r := &pb.DeleteRangeRequest{Key: op.key, RangeEnd: op.end, PrevKv: op.prevKV}
 		return &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{RequestDeleteRange: r}}
 	default:
 		panic("Unknown Op")
@@ -109,6 +130,14 @@ func OpDelete(key string, opts ...OpOption) Op {
 		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
 }
@@ -128,7 +157,15 @@ func OpPut(key, val string, opts ...OpOption) Op {
 	case ret.serializable:
 		panic("unexpected serializable in put")
 	case ret.countOnly:
-		panic("unexpected countOnly in delete")
+		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
 }
@@ -146,7 +183,11 @@ func opWatch(key string, opts ...OpOption) Op {
 	case ret.serializable:
 		panic("unexpected serializable in watch")
 	case ret.countOnly:
-		panic("unexpected countOnly in delete")
+		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
 }
@@ -178,6 +219,14 @@ func WithRev(rev int64) OpOption { return func(op *Op) { op.rev = rev } }
 // '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 current mvcc.Range implementation returns results
+			// sorted by keys in lexiographically ascending order,
+			// client should ignore SortOrder if the target is SortByKey.
+			order = SortNone
+		}
 		op.sort = &SortOption{target, order}
 	}
 }
@@ -241,6 +290,18 @@ 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) }
 
@@ -264,10 +325,65 @@ func withTop(target SortTarget, order SortOrder) []OpOption {
 	return []OpOption{WithPrefix(), WithSort(target, order), WithLimit(1)}
 }
 
-// WithProgressNotify makes watch server send periodic progress updates.
+// 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
+	}
+}
+
+// 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 requests lease timetolive API to return
+// attached keys of 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}
+}

vendor/github.com/coreos/etcd/clientv3/retry.go

@@ -15,9 +15,11 @@
 package clientv3
 
 import (
+	"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
 	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 )
 
 type rpcFunc func(ctx context.Context) error
@@ -27,8 +29,16 @@ func (c *Client) newRetryWrapper() retryRpcFunc {
 	return func(rpcCtx context.Context, f rpcFunc) {
 		for {
 			err := f(rpcCtx)
-			// ignore grpc conn closing on fail-fast calls; they are transient errors
-			if err == nil || !isConnClosing(err) {
+			if err == nil {
+				return
+			}
+			// only retry if unavailable
+			if grpc.Code(err) != codes.Unavailable {
+				return
+			}
+			// always stop retry on etcd errors
+			eErr := rpctypes.Error(err)
+			if _, ok := eErr.(rpctypes.EtcdError); ok {
 				return
 			}
 			select {

vendor/github.com/coreos/etcd/clientv3/txn.go

@@ -19,6 +19,7 @@ import (
 
 	pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
 	"golang.org/x/net/context"
+	"google.golang.org/grpc"
 )
 
 // Txn is the interface that wraps mini-transactions.
@@ -152,7 +153,12 @@ func (txn *txn) Commit() (*TxnResponse, error) {
 
 func (txn *txn) commit() (*TxnResponse, error) {
 	r := &pb.TxnRequest{Compare: txn.cmps, Success: txn.sus, Failure: txn.fas}
-	resp, err := txn.kv.remote.Txn(txn.ctx, r)
+
+	var opts []grpc.CallOption
+	if !txn.isWrite {
+		opts = []grpc.CallOption{grpc.FailFast(false)}
+	}
+	resp, err := txn.kv.remote.Txn(txn.ctx, r, opts...)
 	if err != nil {
 		return nil, err
 	}

vendor/github.com/coreos/etcd/clientv3/watch.go

@@ -61,6 +61,9 @@ type WatchResponse struct {
 	// 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
 }
 
@@ -89,7 +92,7 @@ func (wr *WatchResponse) Err() error {
 
 // IsProgressNotify returns true if the WatchResponse is progress notification.
 func (wr *WatchResponse) IsProgressNotify() bool {
-	return len(wr.Events) == 0 && !wr.Canceled
+	return len(wr.Events) == 0 && !wr.Canceled && !wr.Created && wr.CompactRevision == 0 && wr.Header.Revision != 0
 }
 
 // watcher implements the Watcher interface
@@ -98,10 +101,12 @@ type watcher struct {
 
 	// 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
@@ -112,10 +117,10 @@ type watchGrpcStream struct {
 	ctxKey string
 	cancel context.CancelFunc
 
-	// mu protects the streams map
-	mu sync.RWMutex
-	// streams holds all active watchers
-	streams map[int64]*watcherStream
+	// 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
@@ -127,8 +132,12 @@ type watchGrpcStream struct {
 	donec chan struct{}
 	// errc transmits errors from grpc Recv to the watch stream reconn logic
 	errc chan error
+	// closingc gets the watcherStream of closing watchers
+	closingc chan *watcherStream
 
-	// the error that closed the watch stream
+	// resumec closes to signal that all substreams should begin resuming
+	resumec chan struct{}
+	// closeErr is the error that closed the watch stream
 	closeErr error
 }
 
@@ -138,8 +147,14 @@ type watchRequest struct {
 	key string
 	end string
 	rev int64
-	// progressNotify is for progress updates.
+	// 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
 }
@@ -150,20 +165,27 @@ type watcherStream struct {
 	initReq watchRequest
 
 	// outc publishes watch responses to subscriber
-	outc chan<- WatchResponse
+	outc chan WatchResponse
 	// recvc buffers watch responses before publishing
 	recvc chan *WatchResponse
-	id    int64
+	// 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
 
-	// lastRev is revision last successfully sent over outc
-	lastRev int64
-	// resumec indicates the stream must recover at a given revision
-	resumec chan 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))
+}
+
+func NewWatchFromWatchClient(wc pb.WatchClient) Watcher {
 	return &watcher{
-		remote:  pb.NewWatchClient(c.conn),
+		remote:  wc,
 		streams: make(map[string]*watchGrpcStream),
 	}
 }
@@ -182,18 +204,20 @@ 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,
-		ctx:     ctx,
-		ctxKey:  fmt.Sprintf("%v", inctx),
-		cancel:  cancel,
-		streams: make(map[int64]*watcherStream),
+		owner:      w,
+		remote:     w.remote,
+		ctx:        ctx,
+		ctxKey:     fmt.Sprintf("%v", inctx),
+		cancel:     cancel,
+		substreams: make(map[int64]*watcherStream),
 
-		respc: make(chan *pb.WatchResponse),
-		reqc:  make(chan *watchRequest),
-		stopc: make(chan struct{}),
-		donec: make(chan struct{}),
-		errc:  make(chan error, 1),
+		respc:    make(chan *pb.WatchResponse),
+		reqc:     make(chan *watchRequest),
+		stopc:    make(chan struct{}),
+		donec:    make(chan struct{}),
+		errc:     make(chan error, 1),
+		closingc: make(chan *watcherStream),
+		resumec:  make(chan struct{}),
 	}
 	go wgs.run()
 	return wgs
@@ -203,14 +227,24 @@ func (w *watcher) newWatcherGrpcStream(inctx context.Context) *watchGrpcStream {
 func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) WatchChan {
 	ow := opWatch(key, opts...)
 
-	retc := make(chan chan WatchResponse, 1)
+	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,
-		retc:           retc,
+		filters:        filters,
+		prevKV:         ow.prevKV,
+		retc:           make(chan chan WatchResponse, 1),
 	}
 
 	ok := false
@@ -242,7 +276,6 @@ func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) Watch
 	case reqc <- wr:
 		ok = true
 	case <-wr.ctx.Done():
-		wgs.stopIfEmpty()
 	case <-donec:
 		if wgs.closeErr != nil {
 			closeCh <- WatchResponse{closeErr: wgs.closeErr}
@@ -255,7 +288,7 @@ func (w *watcher) Watch(ctx context.Context, key string, opts ...OpOption) Watch
 	// receive channel
 	if ok {
 		select {
-		case ret := <-retc:
+		case ret := <-wr.retc:
 			return ret
 		case <-ctx.Done():
 		case <-donec:
@@ -286,12 +319,7 @@ func (w *watcher) Close() (err error) {
 }
 
 func (w *watchGrpcStream) Close() (err error) {
-	w.mu.Lock()
-	if w.stopc != nil {
-		close(w.stopc)
-		w.stopc = nil
-	}
-	w.mu.Unlock()
+	close(w.stopc)
 	<-w.donec
 	select {
 	case err = <-w.errc:
@@ -300,67 +328,57 @@ func (w *watchGrpcStream) Close() (err error) {
 	return toErr(w.ctx, err)
 }
 
-func (w *watchGrpcStream) addStream(resp *pb.WatchResponse, pendingReq *watchRequest) {
-	if pendingReq == nil {
-		// no pending request; ignore
-		return
-	}
-	if resp.Canceled || resp.CompactRevision != 0 {
-		// a cancel at id creation time means the start revision has
-		// been compacted out of the store
-		ret := make(chan WatchResponse, 1)
-		ret <- WatchResponse{
-			Header:          *resp.Header,
-			CompactRevision: resp.CompactRevision,
-			Canceled:        true}
-		close(ret)
-		pendingReq.retc <- ret
-		return
-	}
-
-	ret := make(chan WatchResponse)
-	if resp.WatchId == -1 {
-		// failed; no channel
-		close(ret)
-		pendingReq.retc <- ret
-		return
-	}
-
-	ws := &watcherStream{
-		initReq: *pendingReq,
-		id:      resp.WatchId,
-		outc:    ret,
-		// buffered so unlikely to block on sending while holding mu
-		recvc:   make(chan *WatchResponse, 4),
-		resumec: make(chan int64),
-	}
-
-	if pendingReq.rev == 0 {
-		// note the header revision so that a put following a current watcher
-		// disconnect will arrive on the watcher channel after reconnect
-		ws.initReq.rev = resp.Header.Revision
-	}
-
+func (w *watcher) closeStream(wgs *watchGrpcStream) {
 	w.mu.Lock()
-	w.streams[ws.id] = ws
+	close(wgs.donec)
+	wgs.cancel()
+	if w.streams != nil {
+		delete(w.streams, wgs.ctxKey)
+	}
 	w.mu.Unlock()
-
-	// pass back the subscriber channel for the watcher
-	pendingReq.retc <- ret
-
-	// send messages to subscriber
-	go w.serveStream(ws)
 }
 
-// closeStream closes the watcher resources and removes it
-func (w *watchGrpcStream) closeStream(ws *watcherStream) {
-	w.mu.Lock()
-	// cancels request stream; subscriber receives nil channel
-	close(ws.initReq.retc)
-	// close subscriber's channel
+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)
-	delete(w.streams, ws.id)
-	w.mu.Unlock()
+}
+
+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 {
+		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
@@ -368,66 +386,79 @@ func (w *watchGrpcStream) run() {
 	var wc pb.Watch_WatchClient
 	var closeErr error
 
-	defer func() {
-		w.owner.mu.Lock()
-		w.closeErr = closeErr
-		if w.owner.streams != nil {
-			delete(w.owner.streams, w.ctxKey)
-		}
-		close(w.donec)
-		w.owner.mu.Unlock()
-		w.cancel()
-	}()
+	// substreams marked to close but goroutine still running; needed for
+	// avoiding double-closing recvc on grpc stream teardown
+	closing := make(map[*watcherStream]struct{})
 
-	// already stopped?
-	w.mu.RLock()
-	stopc := w.stopc
-	w.mu.RUnlock()
-	if stopc == nil {
-		return
-	}
+	defer func() {
+		w.closeErr = closeErr
+		// shutdown substreams and resuming substreams
+		for _, ws := range w.substreams {
+			if _, ok := closing[ws]; !ok {
+				close(ws.recvc)
+			}
+		}
+		for _, ws := range w.resuming {
+			if _, ok := closing[ws]; ws != nil && !ok {
+				close(ws.recvc)
+			}
+		}
+		w.joinSubstreams()
+		for toClose := len(w.substreams) + len(w.resuming); toClose > 0; toClose-- {
+			w.closeSubstream(<-w.closingc)
+		}
+
+		w.owner.closeStream(w)
+	}()
 
 	// start a stream with the etcd grpc server
 	if wc, closeErr = w.newWatchClient(); closeErr != nil {
 		return
 	}
 
-	var pendingReq, failedReq *watchRequest
-	curReqC := w.reqc
 	cancelSet := make(map[int64]struct{})
 
 	for {
 		select {
 		// Watch() requested
-		case pendingReq = <-curReqC:
-			// no more watch requests until there's a response
-			curReqC = nil
-			if err := wc.Send(pendingReq.toPB()); err == nil {
-				// pendingReq now waits on w.respc
-				break
+		case wreq := <-w.reqc:
+			outc := make(chan WatchResponse, 1)
+			ws := &watcherStream{
+				initReq: *wreq,
+				id:      -1,
+				outc:    outc,
+				// unbufffered so resumes won't cause repeat events
+				recvc: make(chan *WatchResponse),
+			}
+
+			ws.donec = make(chan struct{})
+			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())
 			}
-			failedReq = pendingReq
 		// New events from the watch client
 		case pbresp := <-w.respc:
 			switch {
 			case pbresp.Created:
-				// response to pending req, try to add
-				w.addStream(pbresp, pendingReq)
-				pendingReq = nil
-				curReqC = w.reqc
+				// 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:
 				delete(cancelSet, pbresp.WatchId)
-				// shutdown serveStream, if any
-				w.mu.Lock()
-				if ws, ok := w.streams[pbresp.WatchId]; ok {
+				if ws, ok := w.substreams[pbresp.WatchId]; ok {
+					// signal to stream goroutine to update closingc
 					close(ws.recvc)
-					delete(w.streams, ws.id)
-				}
-				numStreams := len(w.streams)
-				w.mu.Unlock()
-				if numStreams == 0 {
-					// don't leak watcher streams
-					return
+					closing[ws] = struct{}{}
 				}
 			default:
 				// dispatch to appropriate watch stream
@@ -448,57 +479,66 @@ func (w *watchGrpcStream) run() {
 				wc.Send(req)
 			}
 		// watch client failed to recv; spawn another if possible
-		// TODO report watch client errors from errc?
 		case err := <-w.errc:
-			if toErr(w.ctx, err) == v3rpc.ErrNoLeader {
+			if isHaltErr(w.ctx, err) || toErr(w.ctx, err) == v3rpc.ErrNoLeader {
 				closeErr = err
 				return
 			}
 			if wc, closeErr = w.newWatchClient(); closeErr != nil {
 				return
 			}
-			curReqC = w.reqc
-			if pendingReq != nil {
-				failedReq = pendingReq
+			if ws := w.nextResume(); ws != nil {
+				wc.Send(ws.initReq.toPB())
 			}
 			cancelSet = make(map[int64]struct{})
-		case <-stopc:
+		case <-w.stopc:
 			return
-		}
-
-		// send failed; queue for retry
-		if failedReq != nil {
-			go func(wr *watchRequest) {
-				select {
-				case w.reqc <- wr:
-				case <-wr.ctx.Done():
-				case <-w.donec:
-				}
-			}(pendingReq)
-			failedReq = nil
-			pendingReq = nil
+		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 {
-	w.mu.RLock()
-	defer w.mu.RUnlock()
-	ws, ok := w.streams[pbresp.WatchId]
+	ws, ok := w.substreams[pbresp.WatchId]
+	if !ok {
+		return false
+	}
 	events := make([]*Event, len(pbresp.Events))
 	for i, ev := range pbresp.Events {
 		events[i] = (*Event)(ev)
 	}
-	if ok {
-		wr := &WatchResponse{
-			Header:          *pbresp.Header,
-			Events:          events,
-			CompactRevision: pbresp.CompactRevision,
-			Canceled:        pbresp.Canceled}
-		ws.recvc <- wr
+	wr := &WatchResponse{
+		Header:          *pbresp.Header,
+		Events:          events,
+		CompactRevision: pbresp.CompactRevision,
+		Created:         pbresp.Created,
+		Canceled:        pbresp.Canceled,
 	}
-	return ok
+	select {
+	case ws.recvc <- wr:
+	case <-ws.donec:
+		return false
+	}
+	return true
 }
 
 // serveWatchClient forwards messages from the grpc stream to run()
@@ -520,134 +560,126 @@ func (w *watchGrpcStream) serveWatchClient(wc pb.Watch_WatchClient) {
 	}
 }
 
-// serveStream forwards watch responses from run() to the subscriber
-func (w *watchGrpcStream) serveStream(ws *watcherStream) {
-	var closeErr error
-	emptyWr := &WatchResponse{}
-	wrs := []*WatchResponse{}
+// 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
-	closing := false
-	for !closing {
+	defer func() {
+		if !resuming {
+			ws.closing = true
+		}
+		close(ws.donec)
+		if !resuming {
+			w.closingc <- ws
+		}
+	}()
+
+	emptyWr := &WatchResponse{}
+	for {
 		curWr := emptyWr
 		outc := ws.outc
-		if len(wrs) > 0 {
-			curWr = wrs[0]
+
+		if len(ws.buf) > 0 && ws.buf[0].Created {
+			select {
+			case ws.initReq.retc <- ws.outc:
+				// send first creation event and only if requested
+				if !ws.initReq.createdNotify {
+					ws.buf = ws.buf[1:]
+				}
+			default:
+			}
+		}
+
+		if len(ws.buf) > 0 {
+			curWr = ws.buf[0]
 		} else {
 			outc = nil
 		}
 		select {
 		case outc <- *curWr:
-			if wrs[0].Err() != nil {
-				closing = true
-				break
-			}
-			var newRev int64
-			if len(wrs[0].Events) > 0 {
-				newRev = wrs[0].Events[len(wrs[0].Events)-1].Kv.ModRevision
-			} else {
-				newRev = wrs[0].Header.Revision
-			}
-			if newRev != ws.lastRev {
-				ws.lastRev = newRev
-			}
-			wrs[0] = nil
-			wrs = wrs[1:]
-		case wr, ok := <-ws.recvc:
-			if !ok {
-				// shutdown from closeStream
+			if ws.buf[0].Err() != nil {
 				return
 			}
-			// resume up to last seen event if disconnected
-			if resuming && wr.Err() == nil {
-				resuming = false
-				// trim events already seen
-				for i := 0; i < len(wr.Events); i++ {
-					if wr.Events[i].Kv.ModRevision > ws.lastRev {
-						wr.Events = wr.Events[i:]
-						break
-					}
-				}
-				// only forward new events
-				if wr.Events[0].Kv.ModRevision == ws.lastRev {
-					break
-				}
+			ws.buf[0] = nil
+			ws.buf = ws.buf[1:]
+		case wr, ok := <-ws.recvc:
+			if !ok {
+				// shutdown from closeSubstream
+				return
 			}
-			resuming = false
-			// TODO don't keep buffering if subscriber stops reading
-			wrs = append(wrs, wr)
-		case resumeRev := <-ws.resumec:
-			wrs = nil
-			resuming = true
-			if resumeRev == -1 {
-				// pause serving stream while resume gets set up
-				break
+			// TODO pause channel if buffer gets too large
+			ws.buf = append(ws.buf, wr)
+			nextRev = wr.Header.Revision
+			if len(wr.Events) > 0 {
+				nextRev = wr.Events[len(wr.Events)-1].Kv.ModRevision + 1
 			}
-			if resumeRev != ws.lastRev {
-				panic("unexpected resume revision")
-			}
-		case <-w.donec:
-			closing = true
-			closeErr = w.closeErr
+			ws.initReq.rev = nextRev
 		case <-ws.initReq.ctx.Done():
-			closing = true
+			return
+		case <-resumec:
+			resuming = true
+			return
 		}
 	}
-
-	// try to send off close error
-	if closeErr != nil {
-		select {
-		case ws.outc <- WatchResponse{closeErr: w.closeErr}:
-		case <-w.donec:
-		case <-time.After(closeSendErrTimeout):
-		}
-	}
-
-	w.closeStream(ws)
-	w.stopIfEmpty()
 	// lazily send cancel message if events on missing id
 }
 
-func (wgs *watchGrpcStream) stopIfEmpty() {
-	wgs.mu.Lock()
-	if len(wgs.streams) == 0 && wgs.stopc != nil {
-		close(wgs.stopc)
-		wgs.stopc = nil
-	}
-	wgs.mu.Unlock()
-}
-
 func (w *watchGrpcStream) newWatchClient() (pb.Watch_WatchClient, error) {
-	ws, rerr := w.resume()
-	if rerr != nil {
-		return nil, rerr
+	// connect to grpc stream
+	wc, err := w.openWatchClient()
+	if err != nil {
+		return nil, v3rpc.Error(err)
 	}
-	go w.serveWatchClient(ws)
-	return ws, nil
+	// mark all substreams as resuming
+	if len(w.substreams)+len(w.resuming) > 0 {
+		close(w.resumec)
+		w.resumec = make(chan struct{})
+		w.joinSubstreams()
+		for _, ws := range w.substreams {
+			ws.id = -1
+			w.resuming = append(w.resuming, ws)
+		}
+		for _, ws := range w.resuming {
+			if ws == nil || ws.closing {
+				continue
+			}
+			ws.donec = make(chan struct{})
+			go w.serveSubstream(ws, w.resumec)
+		}
+	}
+	w.substreams = make(map[int64]*watcherStream)
+	// receive data from new grpc stream
+	go w.serveWatchClient(wc)
+	return wc, nil
 }
 
-// resume creates a new WatchClient with all current watchers reestablished
-func (w *watchGrpcStream) resume() (ws pb.Watch_WatchClient, err error) {
-	for {
-		if ws, err = w.openWatchClient(); err != nil {
-			break
-		} else if err = w.resumeWatchers(ws); err == nil {
-			break
+// joinSubstream 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
 		}
 	}
-	return ws, v3rpc.Error(err)
 }
 
 // openWatchClient retries opening a watchclient until retryConnection fails
 func (w *watchGrpcStream) openWatchClient() (ws pb.Watch_WatchClient, err error) {
 	for {
-		w.mu.Lock()
-		stopc := w.stopc
-		w.mu.Unlock()
-		if stopc == nil {
+		select {
+		case <-w.stopc:
 			if err == nil {
-				err = context.Canceled
+				return nil, context.Canceled
 			}
 			return nil, err
+		default:
 		}
 		if ws, err = w.remote.Watch(w.ctx, grpc.FailFast(false)); ws != nil && err == nil {
 			break
@@ -659,48 +691,6 @@ func (w *watchGrpcStream) openWatchClient() (ws pb.Watch_WatchClient, err error)
 	return ws, nil
 }
 
-// resumeWatchers rebuilds every registered watcher on a new client
-func (w *watchGrpcStream) resumeWatchers(wc pb.Watch_WatchClient) error {
-	w.mu.RLock()
-	streams := make([]*watcherStream, 0, len(w.streams))
-	for _, ws := range w.streams {
-		streams = append(streams, ws)
-	}
-	w.mu.RUnlock()
-
-	for _, ws := range streams {
-		// pause serveStream
-		ws.resumec <- -1
-
-		// reconstruct watcher from initial request
-		if ws.lastRev != 0 {
-			ws.initReq.rev = ws.lastRev
-		}
-		if err := wc.Send(ws.initReq.toPB()); err != nil {
-			return err
-		}
-
-		// wait for request ack
-		resp, err := wc.Recv()
-		if err != nil {
-			return err
-		} else if len(resp.Events) != 0 || !resp.Created {
-			return fmt.Errorf("watcher: unexpected response (%+v)", resp)
-		}
-
-		// id may be different since new remote watcher; update map
-		w.mu.Lock()
-		delete(w.streams, ws.id)
-		ws.id = resp.WatchId
-		w.streams[ws.id] = ws
-		w.mu.Unlock()
-
-		// unpause serveStream
-		ws.resumec <- ws.lastRev
-	}
-	return nil
-}
-
 // toPB converts an internal watch request structure to its protobuf messagefunc (wr *watchRequest)
 func (wr *watchRequest) toPB() *pb.WatchRequest {
 	req := &pb.WatchCreateRequest{
@@ -708,6 +698,8 @@ func (wr *watchRequest) toPB() *pb.WatchRequest {
 		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}