v3_lease_test.go 22 KB

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