Merge pull request #7732 from heyitsanthony/lease-err-ka

clientv3: don't halt lease client if there is a lease error
etcd-io · Apr 25, 2017 · fbbc4a4 · fbbc4a4
2 parents f254e38 + 2fd6df9
commit fbbc4a4
Show file tree

Hide file tree

Showing 9 changed files with 138 additions and 145 deletions.
diff --git a/clientv3/concurrency/session.go b/clientv3/concurrency/session.go
@@ -51,12 +51,9 @@ func NewSession(client *v3.Client, opts ...SessionOption) (*Session, error) {
 	}
 
 	ctx, cancel := context.WithCancel(ops.ctx)
-	keepAlive, err := client.KeepAlive(ctx, id)
-	if err != nil || keepAlive == nil {
-		return nil, err
-	}
-
+	keepAlive := client.KeepAlive(ctx, id)
 	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

diff --git a/clientv3/example_lease_test.go b/clientv3/example_lease_test.go
@@ -100,12 +100,13 @@ func ExampleLease_keepAlive() {
 	}
 
 	// the key 'foo' will be kept forever
-	ch, kaerr := cli.KeepAlive(context.TODO(), resp.ID)
-	if kaerr != nil {
-		log.Fatal(kaerr)
-	}
+	ch := cli.KeepAlive(context.TODO(), resp.ID)
 
 	ka := <-ch
+	if ka.Err != nil {
+		log.Fatal(ka.Err)
+	}
+
 	fmt.Println("ttl:", ka.TTL)
 	// Output: ttl: 5
 }
@@ -131,9 +132,9 @@ func ExampleLease_keepAliveOnce() {
 	}
 
 	// to renew the lease only once
-	ka, kaerr := cli.KeepAliveOnce(context.TODO(), resp.ID)
-	if kaerr != nil {
-		log.Fatal(kaerr)
+	ka := cli.KeepAliveOnce(context.TODO(), resp.ID)
+	if ka.Err != nil {
+		log.Fatal(ka.Err)
 	}
 
 	fmt.Println("ttl:", ka.TTL)

diff --git a/clientv3/integration/lease_test.go b/clientv3/integration/lease_test.go
@@ -104,14 +104,14 @@ func TestLeaseKeepAliveOnce(t *testing.T) {
 		t.Errorf("failed to create lease %v", err)
 	}
 
-	_, err = lapi.KeepAliveOnce(context.Background(), resp.ID)
-	if err != nil {
-		t.Errorf("failed to keepalive lease %v", err)
+	ka := lapi.KeepAliveOnce(context.Background(), resp.ID)
+	if ka.Err != nil {
+		t.Errorf("failed to keepalive lease %v", ka.Err)
 	}
 
-	_, err = lapi.KeepAliveOnce(context.Background(), clientv3.LeaseID(0))
-	if err != rpctypes.ErrLeaseNotFound {
-		t.Errorf("expected %v, got %v", rpctypes.ErrLeaseNotFound, err)
+	ka = lapi.KeepAliveOnce(context.Background(), clientv3.LeaseID(0))
+	if ka.Err != rpctypes.ErrLeaseNotFound {
+		t.Errorf("expected %v, got %v", rpctypes.ErrLeaseNotFound, ka.Err)
 	}
 }
 
@@ -129,10 +129,7 @@ func TestLeaseKeepAlive(t *testing.T) {
 		t.Errorf("failed to create lease %v", err)
 	}
 
-	rc, kerr := lapi.KeepAlive(context.Background(), resp.ID)
-	if kerr != nil {
-		t.Errorf("failed to keepalive lease %v", kerr)
-	}
+	rc := lapi.KeepAlive(context.Background(), resp.ID)
 
 	kresp, ok := <-rc
 	if !ok {
@@ -163,11 +160,7 @@ func TestLeaseKeepAliveOneSecond(t *testing.T) {
 	if err != nil {
 		t.Errorf("failed to create lease %v", err)
 	}
-	rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
-	if kerr != nil {
-		t.Errorf("failed to keepalive lease %v", kerr)
-	}
-
+	rc := cli.KeepAlive(context.Background(), resp.ID)
 	for i := 0; i < 3; i++ {
 		if _, ok := <-rc; !ok {
 			t.Errorf("chan is closed, want not closed")
@@ -193,10 +186,7 @@ func TestLeaseKeepAliveHandleFailure(t *testing.T) {
 		t.Errorf("failed to create lease %v", err)
 	}
 
-	rc, kerr := lapi.KeepAlive(context.Background(), resp.ID)
-	if kerr != nil {
-		t.Errorf("failed to keepalive lease %v", kerr)
-	}
+	rc := lapi.KeepAlive(context.Background(), resp.ID)
 
 	kresp := <-rc
 	if kresp.ID != resp.ID {
@@ -230,7 +220,7 @@ func TestLeaseKeepAliveHandleFailure(t *testing.T) {
 
 type leaseCh struct {
 	lid clientv3.LeaseID
-	ch  <-chan *clientv3.LeaseKeepAliveResponse
+	ch  clientv3.LeaseKeepAliveChan
 }
 
 // TestLeaseKeepAliveNotFound ensures a revoked lease won't stop other keep alives
@@ -247,10 +237,7 @@ func TestLeaseKeepAliveNotFound(t *testing.T) {
 		if rerr != nil {
 			t.Fatal(rerr)
 		}
-		kach, kaerr := cli.KeepAlive(context.Background(), resp.ID)
-		if kaerr != nil {
-			t.Fatal(kaerr)
-		}
+		kach := cli.KeepAlive(context.Background(), resp.ID)
 		lchs = append(lchs, leaseCh{resp.ID, kach})
 	}
 
@@ -375,10 +362,7 @@ func TestLeaseKeepAliveCloseAfterDisconnectRevoke(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
-	if kerr != nil {
-		t.Fatal(kerr)
-	}
+	rc := cli.KeepAlive(context.Background(), resp.ID)
 	kresp := <-rc
 	if kresp.ID != resp.ID {
 		t.Fatalf("ID = %x, want %x", kresp.ID, resp.ID)
@@ -397,9 +381,10 @@ func TestLeaseKeepAliveCloseAfterDisconnectRevoke(t *testing.T) {
 
 	// some keep-alives may still be buffered; drain until close
 	timer := time.After(time.Duration(kresp.TTL) * time.Second)
-	for kresp != nil {
+	loop := true
+	for loop {
 		select {
-		case kresp = <-rc:
+		case _, loop = <-rc:
 		case <-timer:
 			t.Fatalf("keepalive channel did not close")
 		}
@@ -423,10 +408,7 @@ func TestLeaseKeepAliveInitTimeout(t *testing.T) {
 	}
 	// keep client disconnected
 	clus.Members[0].Stop(t)
-	rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
-	if kerr != nil {
-		t.Fatal(kerr)
-	}
+	rc := cli.KeepAlive(context.Background(), resp.ID)
 	select {
 	case ka, ok := <-rc:
 		if ok {
@@ -454,10 +436,7 @@ func TestLeaseKeepAliveTTLTimeout(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	rc, kerr := cli.KeepAlive(context.Background(), resp.ID)
-	if kerr != nil {
-		t.Fatal(kerr)
-	}
+	rc := cli.KeepAlive(context.Background(), resp.ID)
 	if kresp := <-rc; kresp.ID != resp.ID {
 		t.Fatalf("ID = %x, want %x", kresp.ID, resp.ID)
 	}
@@ -580,10 +559,7 @@ func TestLeaseRenewLostQuorum(t *testing.T) {
 
 	kctx, kcancel := context.WithCancel(context.Background())
 	defer kcancel()
-	ka, err := cli.KeepAlive(kctx, r.ID)
-	if err != nil {
-		t.Fatal(err)
-	}
+	ka := cli.KeepAlive(kctx, r.ID)
 	// consume first keepalive so next message sends when cluster is down
 	<-ka
 	lastKa := time.Now()
@@ -630,9 +606,9 @@ func TestLeaseKeepAliveLoopExit(t *testing.T) {
 	}
 	cli.Close()
 
-	_, err = cli.KeepAlive(ctx, resp.ID)
-	if _, ok := err.(clientv3.ErrKeepAliveHalted); !ok {
-		t.Fatalf("expected %T, got %v(%T)", clientv3.ErrKeepAliveHalted{}, err, err)
+	ka := cli.KeepAlive(ctx, resp.ID)
+	if resp, ok := <-ka; ok {
+		t.Fatalf("expected closed channel, got response %+v", resp)
 	}
 }
 
@@ -707,15 +683,9 @@ func TestLeaseWithRequireLeader(t *testing.T) {
 		t.Fatal(err2)
 	}
 	// kaReqLeader close if the leader is lost
-	kaReqLeader, kerr1 := c.KeepAlive(clientv3.WithRequireLeader(context.TODO()), lid1.ID)
-	if kerr1 != nil {
-		t.Fatal(kerr1)
-	}
+	kaReqLeader := c.KeepAlive(clientv3.WithRequireLeader(context.TODO()), lid1.ID)
 	// kaWait will wait even if the leader is lost
-	kaWait, kerr2 := c.KeepAlive(context.TODO(), lid2.ID)
-	if kerr2 != nil {
-		t.Fatal(kerr2)
-	}
+	kaWait := c.KeepAlive(context.TODO(), lid2.ID)
 
 	select {
 	case <-kaReqLeader: