v3_lease_test.go 22 KB

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