clientv3/integration: test lease keepalive

clientv3/client.go

@@ -15,6 +15,7 @@
 package clientv3
 
 import (
+	"errors"
 	"sync"
 	"time"
 
@@ -25,6 +26,10 @@ import (
 	"github.com/coreos/etcd/pkg/transport"
 )
 
+var (
+	ErrNoAvailableEndpoints = errors.New("etcdclient: no available endpoints")
+)
+
 // Client provides and manages an etcd v3 client session.
 type Client struct {
 	// KV is the keyvalue API for the client's connection.
@@ -182,6 +187,10 @@ func (c *Client) retryConnection(oldConn *grpc.ClientConn, err error) (*grpc.Cli
 // connection is established.
 func dialEndpointList(c *Client) (*grpc.ClientConn, error) {
 	var err error
+	if len(c.Endpoints()) == 0 {
+		return nil, ErrNoAvailableEndpoints
+	}
+
 	for _, ep := range c.Endpoints() {
 		conn, curErr := c.Dial(ep)
 		if curErr != nil {

clientv3/integration/lease_test.go

@@ -16,6 +16,7 @@ package integration
 
 import (
 	"testing"
+	"time"
 
 	"github.com/coreos/etcd/Godeps/_workspace/src/golang.org/x/net/context"
 	"github.com/coreos/etcd/clientv3"
@@ -50,7 +51,7 @@ func TestLeaseCreate(t *testing.T) {
 func TestLeaseRevoke(t *testing.T) {
 	defer testutil.AfterTest(t)
 
-	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 3})
 	defer clus.Terminate(t)
 
 	lapi := clientv3.NewLease(clus.RandClient())
@@ -73,3 +74,111 @@ func TestLeaseRevoke(t *testing.T) {
 		t.Fatalf("err = %v, want %v", err, v3rpc.ErrLeaseNotFound)
 	}
 }
+
+func TestLeaseKeepAliveOnce(t *testing.T) {
+	defer testutil.AfterTest(t)
+
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
+
+	lapi := clientv3.NewLease(clus.RandClient())
+	defer lapi.Close()
+
+	resp, err := lapi.Create(context.Background(), 10)
+	if err != nil {
+		t.Errorf("failed to create lease %v", err)
+	}
+
+	_, err = lapi.KeepAliveOnce(context.Background(), lease.LeaseID(resp.ID))
+	if err != nil {
+		t.Errorf("failed to keepalive lease", err)
+	}
+}
+
+func TestLeaseKeepAlive(t *testing.T) {
+	defer testutil.AfterTest(t)
+
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
+
+	lapi := clientv3.NewLease(clus.RandClient())
+
+	resp, err := lapi.Create(context.Background(), 10)
+	if err != nil {
+		t.Errorf("failed to create lease %v", err)
+	}
+
+	rc, kerr := lapi.KeepAlive(context.Background(), lease.LeaseID(resp.ID))
+	if kerr != nil {
+		t.Errorf("failed to keepalive lease %v", kerr)
+	}
+
+	kresp, ok := <-rc
+	if !ok {
+		t.Errorf("chan is closed, want not closed")
+	}
+
+	if kresp.ID != resp.ID {
+		t.Errorf("ID = %x, want %x", kresp.ID, resp.ID)
+	}
+
+	lapi.Close()
+
+	_, ok = <-rc
+	if ok {
+		t.Errorf("chan is not closed, want lease Close() closes chan")
+	}
+}
+
+// TODO: add a client that can connect to all the members of cluster via unix sock.
+// TODO: test handle more complicated failures.
+func TestLeaseKeepAliveHandleFailure(t *testing.T) {
+	t.Skip("test it when we have a cluster client")
+
+	defer testutil.AfterTest(t)
+
+	clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 3})
+	defer clus.Terminate(t)
+
+	// TODO: change this line to get a cluster client
+	lapi := clientv3.NewLease(clus.RandClient())
+
+	resp, err := lapi.Create(context.Background(), 10)
+	if err != nil {
+		t.Errorf("failed to create lease %v", err)
+	}
+
+	rc, kerr := lapi.KeepAlive(context.Background(), lease.LeaseID(resp.ID))
+	if kerr != nil {
+		t.Errorf("failed to keepalive lease %v", kerr)
+	}
+
+	kresp := <-rc
+	if kresp.ID != resp.ID {
+		t.Errorf("ID = %x, want %x", kresp.ID, resp.ID)
+	}
+
+	// restart the connected member.
+	clus.Members[0].Stop(t)
+
+	select {
+	case <-rc:
+		t.Fatalf("unexpected keepalive")
+	case <-time.After(10*time.Second/3 + 1):
+	}
+
+	// recover the member.
+	clus.Members[0].Restart(t)
+
+	kresp = <-rc
+	if kresp.ID != resp.ID {
+		t.Errorf("ID = %x, want %x", kresp.ID, resp.ID)
+	}
+
+	lapi.Close()
+
+	_, ok := <-rc
+	if ok {
+		t.Errorf("chan is not closed, want lease Close() closes chan")
+	}
+}

clientv3/lease.go

@@ -186,6 +186,7 @@ func (l *lessor) Close() error {
 
 	l.stopCancel()
 	l.stream = nil
+
 	return nil
 }
 
@@ -213,11 +214,25 @@ func (l *lessor) recvKeepAliveLoop() {
 		return
 	}
 
+	defer func() {
+		l.mu.Lock()
+		defer l.mu.Unlock()
+
+		for _, ch := range l.keepAlives {
+			close(ch)
+		}
+	}()
+
 	for {
 		stream := l.getKeepAliveStream()
 
 		resp, err := stream.Recv()
 		if err != nil {
+			if isRPCError(err) {
+				l.Close()
+				return
+			}
+
 			err = l.switchRemoteAndStream(err)
 			if err != nil {
 				l.Close()
@@ -280,6 +295,10 @@ func (l *lessor) sendKeepAliveLoop() {
 			r := &pb.LeaseKeepAliveRequest{ID: int64(id)}
 			err = stream.Send(r)
 			if err != nil {
+				if isRPCError(err) {
+					l.Close()
+					return
+				}
 				break
 			}
 		}