1// Copyright 2016 The etcd Authors 2// 3// Licensed under the Apache License, Version 2.0 (the "License"); 4// you may not use this file except in compliance with the License. 5// You may obtain a copy of the License at 6// 7// http://www.apache.org/licenses/LICENSE-2.0 8// 9// Unless required by applicable law or agreed to in writing, software 10// distributed under the License is distributed on an "AS IS" BASIS, 11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 12// See the License for the specific language governing permissions and 13// limitations under the License. 14 15package integration 16 17import ( 18 "context" 19 "fmt" 20 "testing" 21 "time" 22 23 "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes" 24 pb "github.com/coreos/etcd/etcdserver/etcdserverpb" 25 "github.com/coreos/etcd/mvcc/mvccpb" 26 "github.com/coreos/etcd/pkg/testutil" 27 28 "google.golang.org/grpc/metadata" 29) 30 31// TestV3LeasePrmote ensures the newly elected leader can promote itself 32// to the primary lessor, refresh the leases and start to manage leases. 33// TODO: use customized clock to make this test go faster? 34func TestV3LeasePrmote(t *testing.T) { 35 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 36 defer clus.Terminate(t) 37 38 // create lease 39 lresp, err := toGRPC(clus.RandClient()).Lease.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: 3}) 40 ttl := time.Duration(lresp.TTL) * time.Second 41 afterGrant := time.Now() 42 if err != nil { 43 t.Fatal(err) 44 } 45 if lresp.Error != "" { 46 t.Fatal(lresp.Error) 47 } 48 49 // wait until the lease is going to expire. 50 time.Sleep(time.Until(afterGrant.Add(ttl - time.Second))) 51 52 // kill the current leader, all leases should be refreshed. 53 toStop := clus.waitLeader(t, clus.Members) 54 beforeStop := time.Now() 55 clus.Members[toStop].Stop(t) 56 57 var toWait []*member 58 for i, m := range clus.Members { 59 if i != toStop { 60 toWait = append(toWait, m) 61 } 62 } 63 clus.waitLeader(t, toWait) 64 clus.Members[toStop].Restart(t) 65 clus.waitLeader(t, clus.Members) 66 afterReelect := time.Now() 67 68 // ensure lease is refreshed by waiting for a "long" time. 69 // it was going to expire anyway. 70 time.Sleep(time.Until(beforeStop.Add(ttl - time.Second))) 71 72 if !leaseExist(t, clus, lresp.ID) { 73 t.Error("unexpected lease not exists") 74 } 75 76 // wait until the renewed lease is expected to expire. 77 time.Sleep(time.Until(afterReelect.Add(ttl))) 78 79 // wait for up to 10 seconds for lease to expire. 80 expiredCondition := func() (bool, error) { 81 return !leaseExist(t, clus, lresp.ID), nil 82 } 83 expired, err := testutil.Poll(100*time.Millisecond, 10*time.Second, expiredCondition) 84 if err != nil { 85 t.Error(err) 86 } 87 88 if !expired { 89 t.Error("unexpected lease exists") 90 } 91} 92 93// TestV3LeaseRevoke ensures a key is deleted once its lease is revoked. 94func TestV3LeaseRevoke(t *testing.T) { 95 defer testutil.AfterTest(t) 96 testLeaseRemoveLeasedKey(t, func(clus *ClusterV3, leaseID int64) error { 97 lc := toGRPC(clus.RandClient()).Lease 98 _, err := lc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: leaseID}) 99 return err 100 }) 101} 102 103// TestV3LeaseGrantById ensures leases may be created by a given id. 104func TestV3LeaseGrantByID(t *testing.T) { 105 defer testutil.AfterTest(t) 106 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 107 defer clus.Terminate(t) 108 109 // create fixed lease 110 lresp, err := toGRPC(clus.RandClient()).Lease.LeaseGrant( 111 context.TODO(), 112 &pb.LeaseGrantRequest{ID: 1, TTL: 1}) 113 if err != nil { 114 t.Errorf("could not create lease 1 (%v)", err) 115 } 116 if lresp.ID != 1 { 117 t.Errorf("got id %v, wanted id %v", lresp.ID, 1) 118 } 119 120 // create duplicate fixed lease 121 _, err = toGRPC(clus.RandClient()).Lease.LeaseGrant( 122 context.TODO(), 123 &pb.LeaseGrantRequest{ID: 1, TTL: 1}) 124 if !eqErrGRPC(err, rpctypes.ErrGRPCLeaseExist) { 125 t.Error(err) 126 } 127 128 // create fresh fixed lease 129 lresp, err = toGRPC(clus.RandClient()).Lease.LeaseGrant( 130 context.TODO(), 131 &pb.LeaseGrantRequest{ID: 2, TTL: 1}) 132 if err != nil { 133 t.Errorf("could not create lease 2 (%v)", err) 134 } 135 if lresp.ID != 2 { 136 t.Errorf("got id %v, wanted id %v", lresp.ID, 2) 137 } 138} 139 140// TestV3LeaseExpire ensures a key is deleted once a key expires. 141func TestV3LeaseExpire(t *testing.T) { 142 defer testutil.AfterTest(t) 143 testLeaseRemoveLeasedKey(t, func(clus *ClusterV3, leaseID int64) error { 144 // let lease lapse; wait for deleted key 145 146 ctx, cancel := context.WithCancel(context.Background()) 147 defer cancel() 148 wStream, err := toGRPC(clus.RandClient()).Watch.Watch(ctx) 149 if err != nil { 150 return err 151 } 152 153 wreq := &pb.WatchRequest{RequestUnion: &pb.WatchRequest_CreateRequest{ 154 CreateRequest: &pb.WatchCreateRequest{ 155 Key: []byte("foo"), StartRevision: 1}}} 156 if err := wStream.Send(wreq); err != nil { 157 return err 158 } 159 if _, err := wStream.Recv(); err != nil { 160 // the 'created' message 161 return err 162 } 163 if _, err := wStream.Recv(); err != nil { 164 // the 'put' message 165 return err 166 } 167 168 errc := make(chan error, 1) 169 go func() { 170 resp, err := wStream.Recv() 171 switch { 172 case err != nil: 173 errc <- err 174 case len(resp.Events) != 1: 175 fallthrough 176 case resp.Events[0].Type != mvccpb.DELETE: 177 errc <- fmt.Errorf("expected key delete, got %v", resp) 178 default: 179 errc <- nil 180 } 181 }() 182 183 select { 184 case <-time.After(15 * time.Second): 185 return fmt.Errorf("lease expiration too slow") 186 case err := <-errc: 187 return err 188 } 189 }) 190} 191 192// TestV3LeaseKeepAlive ensures keepalive keeps the lease alive. 193func TestV3LeaseKeepAlive(t *testing.T) { 194 defer testutil.AfterTest(t) 195 testLeaseRemoveLeasedKey(t, func(clus *ClusterV3, leaseID int64) error { 196 lc := toGRPC(clus.RandClient()).Lease 197 lreq := &pb.LeaseKeepAliveRequest{ID: leaseID} 198 ctx, cancel := context.WithCancel(context.Background()) 199 defer cancel() 200 lac, err := lc.LeaseKeepAlive(ctx) 201 if err != nil { 202 return err 203 } 204 defer lac.CloseSend() 205 206 // renew long enough so lease would've expired otherwise 207 for i := 0; i < 3; i++ { 208 if err = lac.Send(lreq); err != nil { 209 return err 210 } 211 lresp, rxerr := lac.Recv() 212 if rxerr != nil { 213 return rxerr 214 } 215 if lresp.ID != leaseID { 216 return fmt.Errorf("expected lease ID %v, got %v", leaseID, lresp.ID) 217 } 218 time.Sleep(time.Duration(lresp.TTL/2) * time.Second) 219 } 220 _, err = lc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: leaseID}) 221 return err 222 }) 223} 224 225// TestV3LeaseExists creates a lease on a random client and confirms it exists in the cluster. 226func TestV3LeaseExists(t *testing.T) { 227 defer testutil.AfterTest(t) 228 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 229 defer clus.Terminate(t) 230 231 // create lease 232 ctx0, cancel0 := context.WithCancel(context.Background()) 233 defer cancel0() 234 lresp, err := toGRPC(clus.RandClient()).Lease.LeaseGrant( 235 ctx0, 236 &pb.LeaseGrantRequest{TTL: 30}) 237 if err != nil { 238 t.Fatal(err) 239 } 240 if lresp.Error != "" { 241 t.Fatal(lresp.Error) 242 } 243 244 if !leaseExist(t, clus, lresp.ID) { 245 t.Error("unexpected lease not exists") 246 } 247} 248 249// TestV3LeaseLeases creates leases and confirms list RPC fetches created ones. 250func TestV3LeaseLeases(t *testing.T) { 251 defer testutil.AfterTest(t) 252 clus := NewClusterV3(t, &ClusterConfig{Size: 1}) 253 defer clus.Terminate(t) 254 255 ctx0, cancel0 := context.WithCancel(context.Background()) 256 defer cancel0() 257 258 // create leases 259 ids := []int64{} 260 for i := 0; i < 5; i++ { 261 lresp, err := toGRPC(clus.RandClient()).Lease.LeaseGrant( 262 ctx0, 263 &pb.LeaseGrantRequest{TTL: 30}) 264 if err != nil { 265 t.Fatal(err) 266 } 267 if lresp.Error != "" { 268 t.Fatal(lresp.Error) 269 } 270 ids = append(ids, lresp.ID) 271 } 272 273 lresp, err := toGRPC(clus.RandClient()).Lease.LeaseLeases( 274 context.Background(), 275 &pb.LeaseLeasesRequest{}) 276 if err != nil { 277 t.Fatal(err) 278 } 279 for i := range lresp.Leases { 280 if lresp.Leases[i].ID != ids[i] { 281 t.Fatalf("#%d: lease ID expected %d, got %d", i, ids[i], lresp.Leases[i].ID) 282 } 283 } 284} 285 286// TestV3LeaseRenewStress keeps creating lease and renewing it immediately to ensure the renewal goes through. 287// it was oberserved that the immediate lease renewal after granting a lease from follower resulted lease not found. 288// related issue https://github.com/coreos/etcd/issues/6978 289func TestV3LeaseRenewStress(t *testing.T) { 290 testLeaseStress(t, stressLeaseRenew) 291} 292 293// TestV3LeaseTimeToLiveStress keeps creating lease and retrieving it immediately to ensure the lease can be retrieved. 294// it was oberserved that the immediate lease retrieval after granting a lease from follower resulted lease not found. 295// related issue https://github.com/coreos/etcd/issues/6978 296func TestV3LeaseTimeToLiveStress(t *testing.T) { 297 testLeaseStress(t, stressLeaseTimeToLive) 298} 299 300func testLeaseStress(t *testing.T, stresser func(context.Context, pb.LeaseClient) error) { 301 defer testutil.AfterTest(t) 302 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 303 defer clus.Terminate(t) 304 305 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) 306 defer cancel() 307 errc := make(chan error) 308 309 for i := 0; i < 30; i++ { 310 for j := 0; j < 3; j++ { 311 go func(i int) { errc <- stresser(ctx, toGRPC(clus.Client(i)).Lease) }(j) 312 } 313 } 314 315 for i := 0; i < 90; i++ { 316 if err := <-errc; err != nil { 317 t.Fatal(err) 318 } 319 } 320} 321 322func stressLeaseRenew(tctx context.Context, lc pb.LeaseClient) (reterr error) { 323 defer func() { 324 if tctx.Err() != nil { 325 reterr = nil 326 } 327 }() 328 lac, err := lc.LeaseKeepAlive(tctx) 329 if err != nil { 330 return err 331 } 332 for tctx.Err() == nil { 333 resp, gerr := lc.LeaseGrant(tctx, &pb.LeaseGrantRequest{TTL: 60}) 334 if gerr != nil { 335 continue 336 } 337 err = lac.Send(&pb.LeaseKeepAliveRequest{ID: resp.ID}) 338 if err != nil { 339 continue 340 } 341 rresp, rxerr := lac.Recv() 342 if rxerr != nil { 343 continue 344 } 345 if rresp.TTL == 0 { 346 return fmt.Errorf("TTL shouldn't be 0 so soon") 347 } 348 } 349 return nil 350} 351 352func stressLeaseTimeToLive(tctx context.Context, lc pb.LeaseClient) (reterr error) { 353 defer func() { 354 if tctx.Err() != nil { 355 reterr = nil 356 } 357 }() 358 for tctx.Err() == nil { 359 resp, gerr := lc.LeaseGrant(tctx, &pb.LeaseGrantRequest{TTL: 60}) 360 if gerr != nil { 361 continue 362 } 363 _, kerr := lc.LeaseTimeToLive(tctx, &pb.LeaseTimeToLiveRequest{ID: resp.ID}) 364 if rpctypes.Error(kerr) == rpctypes.ErrLeaseNotFound { 365 return kerr 366 } 367 } 368 return nil 369} 370 371func TestV3PutOnNonExistLease(t *testing.T) { 372 defer testutil.AfterTest(t) 373 clus := NewClusterV3(t, &ClusterConfig{Size: 1}) 374 defer clus.Terminate(t) 375 376 ctx, cancel := context.WithCancel(context.Background()) 377 defer cancel() 378 379 badLeaseID := int64(0x12345678) 380 putr := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: badLeaseID} 381 _, err := toGRPC(clus.RandClient()).KV.Put(ctx, putr) 382 if !eqErrGRPC(err, rpctypes.ErrGRPCLeaseNotFound) { 383 t.Errorf("err = %v, want %v", err, rpctypes.ErrGRPCLeaseNotFound) 384 } 385} 386 387// TestV3GetNonExistLease ensures client retrieving nonexistent lease on a follower doesn't result node panic 388// related issue https://github.com/coreos/etcd/issues/6537 389func TestV3GetNonExistLease(t *testing.T) { 390 defer testutil.AfterTest(t) 391 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 392 defer clus.Terminate(t) 393 394 ctx, cancel := context.WithCancel(context.Background()) 395 defer cancel() 396 lc := toGRPC(clus.RandClient()).Lease 397 lresp, err := lc.LeaseGrant(ctx, &pb.LeaseGrantRequest{TTL: 10}) 398 if err != nil { 399 t.Errorf("failed to create lease %v", err) 400 } 401 _, err = lc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID}) 402 if err != nil { 403 t.Fatal(err) 404 } 405 406 leaseTTLr := &pb.LeaseTimeToLiveRequest{ 407 ID: lresp.ID, 408 Keys: true, 409 } 410 411 for _, client := range clus.clients { 412 // quorum-read to ensure revoke completes before TimeToLive 413 if _, err := toGRPC(client).KV.Range(ctx, &pb.RangeRequest{Key: []byte("_")}); err != nil { 414 t.Fatal(err) 415 } 416 resp, err := toGRPC(client).Lease.LeaseTimeToLive(ctx, leaseTTLr) 417 if err != nil { 418 t.Fatalf("expected non nil error, but go %v", err) 419 } 420 if resp.TTL != -1 { 421 t.Fatalf("expected TTL to be -1, but got %v", resp.TTL) 422 } 423 } 424} 425 426// TestV3LeaseSwitch tests a key can be switched from one lease to another. 427func TestV3LeaseSwitch(t *testing.T) { 428 defer testutil.AfterTest(t) 429 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 430 defer clus.Terminate(t) 431 432 key := "foo" 433 434 // create lease 435 ctx, cancel := context.WithCancel(context.Background()) 436 defer cancel() 437 lresp1, err1 := toGRPC(clus.RandClient()).Lease.LeaseGrant(ctx, &pb.LeaseGrantRequest{TTL: 30}) 438 if err1 != nil { 439 t.Fatal(err1) 440 } 441 lresp2, err2 := toGRPC(clus.RandClient()).Lease.LeaseGrant(ctx, &pb.LeaseGrantRequest{TTL: 30}) 442 if err2 != nil { 443 t.Fatal(err2) 444 } 445 446 // attach key on lease1 then switch it to lease2 447 put1 := &pb.PutRequest{Key: []byte(key), Lease: lresp1.ID} 448 _, err := toGRPC(clus.RandClient()).KV.Put(ctx, put1) 449 if err != nil { 450 t.Fatal(err) 451 } 452 put2 := &pb.PutRequest{Key: []byte(key), Lease: lresp2.ID} 453 _, err = toGRPC(clus.RandClient()).KV.Put(ctx, put2) 454 if err != nil { 455 t.Fatal(err) 456 } 457 458 // revoke lease1 should not remove key 459 _, err = toGRPC(clus.RandClient()).Lease.LeaseRevoke(ctx, &pb.LeaseRevokeRequest{ID: lresp1.ID}) 460 if err != nil { 461 t.Fatal(err) 462 } 463 rreq := &pb.RangeRequest{Key: []byte("foo")} 464 rresp, err := toGRPC(clus.RandClient()).KV.Range(context.TODO(), rreq) 465 if err != nil { 466 t.Fatal(err) 467 } 468 if len(rresp.Kvs) != 1 { 469 t.Fatalf("unexpect removal of key") 470 } 471 472 // revoke lease2 should remove key 473 _, err = toGRPC(clus.RandClient()).Lease.LeaseRevoke(ctx, &pb.LeaseRevokeRequest{ID: lresp2.ID}) 474 if err != nil { 475 t.Fatal(err) 476 } 477 rresp, err = toGRPC(clus.RandClient()).KV.Range(context.TODO(), rreq) 478 if err != nil { 479 t.Fatal(err) 480 } 481 if len(rresp.Kvs) != 0 { 482 t.Fatalf("lease removed but key remains") 483 } 484} 485 486// TestV3LeaseFailover ensures the old leader drops lease keepalive requests within 487// election timeout after it loses its quorum. And the new leader extends the TTL of 488// the lease to at least TTL + election timeout. 489func TestV3LeaseFailover(t *testing.T) { 490 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 491 defer clus.Terminate(t) 492 493 toIsolate := clus.waitLeader(t, clus.Members) 494 495 lc := toGRPC(clus.Client(toIsolate)).Lease 496 497 // create lease 498 lresp, err := lc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: 5}) 499 if err != nil { 500 t.Fatal(err) 501 } 502 if lresp.Error != "" { 503 t.Fatal(lresp.Error) 504 } 505 506 // isolate the current leader with its followers. 507 clus.Members[toIsolate].Pause() 508 509 lreq := &pb.LeaseKeepAliveRequest{ID: lresp.ID} 510 511 md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader) 512 mctx := metadata.NewOutgoingContext(context.Background(), md) 513 ctx, cancel := context.WithCancel(mctx) 514 lac, err := lc.LeaseKeepAlive(ctx) 515 if err != nil { 516 t.Fatal(err) 517 } 518 defer func() { 519 lac.CloseSend() 520 cancel() 521 }() 522 523 // send keep alive to old leader until the old leader starts 524 // to drop lease request. 525 var expectedExp time.Time 526 for { 527 if err = lac.Send(lreq); err != nil { 528 break 529 } 530 lkresp, rxerr := lac.Recv() 531 if rxerr != nil { 532 break 533 } 534 expectedExp = time.Now().Add(time.Duration(lkresp.TTL) * time.Second) 535 time.Sleep(time.Duration(lkresp.TTL/2) * time.Second) 536 } 537 538 clus.Members[toIsolate].Resume() 539 clus.waitLeader(t, clus.Members) 540 541 // lease should not expire at the last received expire deadline. 542 time.Sleep(time.Until(expectedExp) - 500*time.Millisecond) 543 544 if !leaseExist(t, clus, lresp.ID) { 545 t.Error("unexpected lease not exists") 546 } 547} 548 549// TestV3LeaseRequireLeader ensures that a Recv will get a leader 550// loss error if there is no leader. 551func TestV3LeaseRequireLeader(t *testing.T) { 552 defer testutil.AfterTest(t) 553 554 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 555 defer clus.Terminate(t) 556 557 lc := toGRPC(clus.Client(0)).Lease 558 clus.Members[1].Stop(t) 559 clus.Members[2].Stop(t) 560 561 md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader) 562 mctx := metadata.NewOutgoingContext(context.Background(), md) 563 ctx, cancel := context.WithCancel(mctx) 564 defer cancel() 565 lac, err := lc.LeaseKeepAlive(ctx) 566 if err != nil { 567 t.Fatal(err) 568 } 569 570 donec := make(chan struct{}) 571 go func() { 572 defer close(donec) 573 resp, err := lac.Recv() 574 if err == nil { 575 t.Fatalf("got response %+v, expected error", resp) 576 } 577 if rpctypes.ErrorDesc(err) != rpctypes.ErrNoLeader.Error() { 578 t.Errorf("err = %v, want %v", err, rpctypes.ErrNoLeader) 579 } 580 }() 581 select { 582 case <-time.After(5 * time.Second): 583 t.Fatal("did not receive leader loss error (in 5-sec)") 584 case <-donec: 585 } 586} 587 588const fiveMinTTL int64 = 300 589 590// TestV3LeaseRecoverAndRevoke ensures that revoking a lease after restart deletes the attached key. 591func TestV3LeaseRecoverAndRevoke(t *testing.T) { 592 clus := NewClusterV3(t, &ClusterConfig{Size: 1}) 593 defer clus.Terminate(t) 594 595 kvc := toGRPC(clus.Client(0)).KV 596 lsc := toGRPC(clus.Client(0)).Lease 597 598 lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL}) 599 if err != nil { 600 t.Fatal(err) 601 } 602 if lresp.Error != "" { 603 t.Fatal(lresp.Error) 604 } 605 _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID}) 606 if err != nil { 607 t.Fatal(err) 608 } 609 610 // restart server and ensure lease still exists 611 clus.Members[0].Stop(t) 612 clus.Members[0].Restart(t) 613 clus.waitLeader(t, clus.Members) 614 615 // overwrite old client with newly dialed connection 616 // otherwise, error with "grpc: RPC failed fast due to transport failure" 617 nc, err := NewClientV3(clus.Members[0]) 618 if err != nil { 619 t.Fatal(err) 620 } 621 kvc = toGRPC(nc).KV 622 lsc = toGRPC(nc).Lease 623 defer nc.Close() 624 625 // revoke should delete the key 626 _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID}) 627 if err != nil { 628 t.Fatal(err) 629 } 630 rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")}) 631 if err != nil { 632 t.Fatal(err) 633 } 634 if len(rresp.Kvs) != 0 { 635 t.Fatalf("lease removed but key remains") 636 } 637} 638 639// TestV3LeaseRevokeAndRecover ensures that revoked key stays deleted after restart. 640func TestV3LeaseRevokeAndRecover(t *testing.T) { 641 clus := NewClusterV3(t, &ClusterConfig{Size: 1}) 642 defer clus.Terminate(t) 643 644 kvc := toGRPC(clus.Client(0)).KV 645 lsc := toGRPC(clus.Client(0)).Lease 646 647 lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL}) 648 if err != nil { 649 t.Fatal(err) 650 } 651 if lresp.Error != "" { 652 t.Fatal(lresp.Error) 653 } 654 _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID}) 655 if err != nil { 656 t.Fatal(err) 657 } 658 659 // revoke should delete the key 660 _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID}) 661 if err != nil { 662 t.Fatal(err) 663 } 664 665 // restart server and ensure revoked key doesn't exist 666 clus.Members[0].Stop(t) 667 clus.Members[0].Restart(t) 668 clus.waitLeader(t, clus.Members) 669 670 // overwrite old client with newly dialed connection 671 // otherwise, error with "grpc: RPC failed fast due to transport failure" 672 nc, err := NewClientV3(clus.Members[0]) 673 if err != nil { 674 t.Fatal(err) 675 } 676 kvc = toGRPC(nc).KV 677 defer nc.Close() 678 679 rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")}) 680 if err != nil { 681 t.Fatal(err) 682 } 683 if len(rresp.Kvs) != 0 { 684 t.Fatalf("lease removed but key remains") 685 } 686} 687 688// TestV3LeaseRecoverKeyWithDetachedLease ensures that revoking a detached lease after restart 689// does not delete the key. 690func TestV3LeaseRecoverKeyWithDetachedLease(t *testing.T) { 691 clus := NewClusterV3(t, &ClusterConfig{Size: 1}) 692 defer clus.Terminate(t) 693 694 kvc := toGRPC(clus.Client(0)).KV 695 lsc := toGRPC(clus.Client(0)).Lease 696 697 lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL}) 698 if err != nil { 699 t.Fatal(err) 700 } 701 if lresp.Error != "" { 702 t.Fatal(lresp.Error) 703 } 704 _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID}) 705 if err != nil { 706 t.Fatal(err) 707 } 708 709 // overwrite lease with none 710 _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}) 711 if err != nil { 712 t.Fatal(err) 713 } 714 715 // restart server and ensure lease still exists 716 clus.Members[0].Stop(t) 717 clus.Members[0].Restart(t) 718 clus.waitLeader(t, clus.Members) 719 720 // overwrite old client with newly dialed connection 721 // otherwise, error with "grpc: RPC failed fast due to transport failure" 722 nc, err := NewClientV3(clus.Members[0]) 723 if err != nil { 724 t.Fatal(err) 725 } 726 kvc = toGRPC(nc).KV 727 lsc = toGRPC(nc).Lease 728 defer nc.Close() 729 730 // revoke the detached lease 731 _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID}) 732 if err != nil { 733 t.Fatal(err) 734 } 735 rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")}) 736 if err != nil { 737 t.Fatal(err) 738 } 739 if len(rresp.Kvs) != 1 { 740 t.Fatalf("only detached lease removed, key should remain") 741 } 742} 743 744func TestV3LeaseRecoverKeyWithMutipleLease(t *testing.T) { 745 clus := NewClusterV3(t, &ClusterConfig{Size: 1}) 746 defer clus.Terminate(t) 747 748 kvc := toGRPC(clus.Client(0)).KV 749 lsc := toGRPC(clus.Client(0)).Lease 750 751 var leaseIDs []int64 752 for i := 0; i < 2; i++ { 753 lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL}) 754 if err != nil { 755 t.Fatal(err) 756 } 757 if lresp.Error != "" { 758 t.Fatal(lresp.Error) 759 } 760 leaseIDs = append(leaseIDs, lresp.ID) 761 762 _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID}) 763 if err != nil { 764 t.Fatal(err) 765 } 766 } 767 768 // restart server and ensure lease still exists 769 clus.Members[0].Stop(t) 770 clus.Members[0].Restart(t) 771 clus.waitLeader(t, clus.Members) 772 for i, leaseID := range leaseIDs { 773 if !leaseExist(t, clus, leaseID) { 774 t.Errorf("#%d: unexpected lease not exists", i) 775 } 776 } 777 778 // overwrite old client with newly dialed connection 779 // otherwise, error with "grpc: RPC failed fast due to transport failure" 780 nc, err := NewClientV3(clus.Members[0]) 781 if err != nil { 782 t.Fatal(err) 783 } 784 kvc = toGRPC(nc).KV 785 lsc = toGRPC(nc).Lease 786 defer nc.Close() 787 788 // revoke the old lease 789 _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: leaseIDs[0]}) 790 if err != nil { 791 t.Fatal(err) 792 } 793 // key should still exist 794 rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")}) 795 if err != nil { 796 t.Fatal(err) 797 } 798 if len(rresp.Kvs) != 1 { 799 t.Fatalf("only detached lease removed, key should remain") 800 } 801 802 // revoke the latest lease 803 _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: leaseIDs[1]}) 804 if err != nil { 805 t.Fatal(err) 806 } 807 rresp, err = kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")}) 808 if err != nil { 809 t.Fatal(err) 810 } 811 if len(rresp.Kvs) != 0 { 812 t.Fatalf("lease removed but key remains") 813 } 814} 815 816// acquireLeaseAndKey creates a new lease and creates an attached key. 817func acquireLeaseAndKey(clus *ClusterV3, key string) (int64, error) { 818 // create lease 819 lresp, err := toGRPC(clus.RandClient()).Lease.LeaseGrant( 820 context.TODO(), 821 &pb.LeaseGrantRequest{TTL: 1}) 822 if err != nil { 823 return 0, err 824 } 825 if lresp.Error != "" { 826 return 0, fmt.Errorf(lresp.Error) 827 } 828 // attach to key 829 put := &pb.PutRequest{Key: []byte(key), Lease: lresp.ID} 830 if _, err := toGRPC(clus.RandClient()).KV.Put(context.TODO(), put); err != nil { 831 return 0, err 832 } 833 return lresp.ID, nil 834} 835 836// testLeaseRemoveLeasedKey performs some action while holding a lease with an 837// attached key "foo", then confirms the key is gone. 838func testLeaseRemoveLeasedKey(t *testing.T, act func(*ClusterV3, int64) error) { 839 clus := NewClusterV3(t, &ClusterConfig{Size: 3}) 840 defer clus.Terminate(t) 841 842 leaseID, err := acquireLeaseAndKey(clus, "foo") 843 if err != nil { 844 t.Fatal(err) 845 } 846 847 if err = act(clus, leaseID); err != nil { 848 t.Fatal(err) 849 } 850 851 // confirm no key 852 rreq := &pb.RangeRequest{Key: []byte("foo")} 853 rresp, err := toGRPC(clus.RandClient()).KV.Range(context.TODO(), rreq) 854 if err != nil { 855 t.Fatal(err) 856 } 857 if len(rresp.Kvs) != 0 { 858 t.Fatalf("lease removed but key remains") 859 } 860} 861 862func leaseExist(t *testing.T, clus *ClusterV3, leaseID int64) bool { 863 l := toGRPC(clus.RandClient()).Lease 864 865 _, err := l.LeaseGrant(context.Background(), &pb.LeaseGrantRequest{ID: leaseID, TTL: 5}) 866 if err == nil { 867 _, err = l.LeaseRevoke(context.Background(), &pb.LeaseRevokeRequest{ID: leaseID}) 868 if err != nil { 869 t.Fatalf("failed to check lease %v", err) 870 } 871 return false 872 } 873 874 if eqErrGRPC(err, rpctypes.ErrGRPCLeaseExist) { 875 return true 876 } 877 t.Fatalf("unexpecter error %v", err) 878 879 return true 880} 881