[VOL-4291] Rw-core updates for gRPC migration
Change-Id: I8d5a554409115b29318089671ca4e1ab3fa98810
diff --git a/vendor/go.etcd.io/etcd/clientv3/lease.go b/vendor/go.etcd.io/etcd/clientv3/lease.go
index c2796fc..3729cf3 100644
--- a/vendor/go.etcd.io/etcd/clientv3/lease.go
+++ b/vendor/go.etcd.io/etcd/clientv3/lease.go
@@ -19,10 +19,9 @@
"sync"
"time"
- "go.etcd.io/etcd/etcdserver/api/v3rpc/rpctypes"
- pb "go.etcd.io/etcd/etcdserver/etcdserverpb"
+ "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
+ pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
- "go.uber.org/zap"
"google.golang.org/grpc"
"google.golang.org/grpc/metadata"
)
@@ -118,21 +117,22 @@
// Leases retrieves all leases.
Leases(ctx context.Context) (*LeaseLeasesResponse, error)
- // KeepAlive attempts to keep the given lease alive forever. If the keepalive responses posted
- // to the channel are not consumed promptly the channel may become full. When full, the lease
- // client will continue sending keep alive requests to the etcd server, but will drop responses
- // until there is capacity on the channel to send more responses.
- //
- // If client keep alive loop halts with an unexpected error (e.g. "etcdserver: no leader") or
- // canceled by the caller (e.g. context.Canceled), KeepAlive returns a ErrKeepAliveHalted error
- // containing the error reason.
+ // 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.
+ // 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/etcd-io/etcd/pull/7866)
+ // (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
@@ -172,8 +172,6 @@
firstKeepAliveOnce sync.Once
callOpts []grpc.CallOption
-
- lg *zap.Logger
}
// keepAlive multiplexes a keepalive for a lease over multiple channels
@@ -198,7 +196,6 @@
keepAlives: make(map[LeaseID]*keepAlive),
remote: remote,
firstKeepAliveTimeout: keepAliveTimeout,
- lg: c.lg,
}
if l.firstKeepAliveTimeout == time.Second {
l.firstKeepAliveTimeout = defaultTTL
@@ -294,7 +291,7 @@
}
l.mu.Unlock()
- go l.keepAliveCtxCloser(ctx, id, ka.donec)
+ go l.keepAliveCtxCloser(id, ctx, ka.donec)
l.firstKeepAliveOnce.Do(func() {
go l.recvKeepAliveLoop()
go l.deadlineLoop()
@@ -326,7 +323,7 @@
return nil
}
-func (l *lessor) keepAliveCtxCloser(ctx context.Context, id LeaseID, donec <-chan struct{}) {
+func (l *lessor) keepAliveCtxCloser(id LeaseID, ctx context.Context, donec <-chan struct{}) {
select {
case <-donec:
return
@@ -462,6 +459,7 @@
select {
case <-time.After(retryConnWait):
+ continue
case <-l.stopCtx.Done():
return l.stopCtx.Err()
}
@@ -471,7 +469,7 @@
// 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, append(l.callOpts, withMax(0))...)
+ stream, err := l.remote.LeaseKeepAlive(sctx, l.callOpts...)
if err != nil {
cancel()
return nil, err
@@ -520,12 +518,6 @@
select {
case ch <- karesp:
default:
- if l.lg != nil {
- l.lg.Warn("lease keepalive response queue is full; dropping response send",
- zap.Int("queue-size", len(ch)),
- zap.Int("queue-capacity", cap(ch)),
- )
- }
}
// still advance in order to rate-limit keep-alive sends
ka.nextKeepAlive = nextKeepAlive
@@ -577,7 +569,7 @@
}
select {
- case <-time.After(retryConnWait):
+ case <-time.After(500 * time.Millisecond):
case <-stream.Context().Done():
return
case <-l.donec: