v3_lease_test.go 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829
  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. // quorum-read to ensure revoke completes before TimeToLive
  322. if _, err := toGRPC(client).KV.Range(ctx, &pb.RangeRequest{Key: []byte("_")}); err != nil {
  323. t.Fatal(err)
  324. }
  325. resp, err := toGRPC(client).Lease.LeaseTimeToLive(ctx, leaseTTLr)
  326. if err != nil {
  327. t.Fatalf("expected non nil error, but go %v", err)
  328. }
  329. if resp.TTL != -1 {
  330. t.Fatalf("expected TTL to be -1, but got %v", resp.TTL)
  331. }
  332. }
  333. }
  334. // TestV3LeaseSwitch tests a key can be switched from one lease to another.
  335. func TestV3LeaseSwitch(t *testing.T) {
  336. defer testutil.AfterTest(t)
  337. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  338. defer clus.Terminate(t)
  339. key := "foo"
  340. // create lease
  341. ctx, cancel := context.WithCancel(context.Background())
  342. defer cancel()
  343. lresp1, err1 := toGRPC(clus.RandClient()).Lease.LeaseGrant(ctx, &pb.LeaseGrantRequest{TTL: 30})
  344. if err1 != nil {
  345. t.Fatal(err1)
  346. }
  347. lresp2, err2 := toGRPC(clus.RandClient()).Lease.LeaseGrant(ctx, &pb.LeaseGrantRequest{TTL: 30})
  348. if err2 != nil {
  349. t.Fatal(err2)
  350. }
  351. // attach key on lease1 then switch it to lease2
  352. put1 := &pb.PutRequest{Key: []byte(key), Lease: lresp1.ID}
  353. _, err := toGRPC(clus.RandClient()).KV.Put(ctx, put1)
  354. if err != nil {
  355. t.Fatal(err)
  356. }
  357. put2 := &pb.PutRequest{Key: []byte(key), Lease: lresp2.ID}
  358. _, err = toGRPC(clus.RandClient()).KV.Put(ctx, put2)
  359. if err != nil {
  360. t.Fatal(err)
  361. }
  362. // revoke lease1 should not remove key
  363. _, err = toGRPC(clus.RandClient()).Lease.LeaseRevoke(ctx, &pb.LeaseRevokeRequest{ID: lresp1.ID})
  364. if err != nil {
  365. t.Fatal(err)
  366. }
  367. rreq := &pb.RangeRequest{Key: []byte("foo")}
  368. rresp, err := toGRPC(clus.RandClient()).KV.Range(context.TODO(), rreq)
  369. if err != nil {
  370. t.Fatal(err)
  371. }
  372. if len(rresp.Kvs) != 1 {
  373. t.Fatalf("unexpect removal of key")
  374. }
  375. // revoke lease2 should remove key
  376. _, err = toGRPC(clus.RandClient()).Lease.LeaseRevoke(ctx, &pb.LeaseRevokeRequest{ID: lresp2.ID})
  377. if err != nil {
  378. t.Fatal(err)
  379. }
  380. rresp, err = toGRPC(clus.RandClient()).KV.Range(context.TODO(), rreq)
  381. if err != nil {
  382. t.Fatal(err)
  383. }
  384. if len(rresp.Kvs) != 0 {
  385. t.Fatalf("lease removed but key remains")
  386. }
  387. }
  388. // TestV3LeaseFailover ensures the old leader drops lease keepalive requests within
  389. // election timeout after it loses its quorum. And the new leader extends the TTL of
  390. // the lease to at least TTL + election timeout.
  391. func TestV3LeaseFailover(t *testing.T) {
  392. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  393. defer clus.Terminate(t)
  394. toIsolate := clus.waitLeader(t, clus.Members)
  395. lc := toGRPC(clus.Client(toIsolate)).Lease
  396. // create lease
  397. lresp, err := lc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: 5})
  398. if err != nil {
  399. t.Fatal(err)
  400. }
  401. if lresp.Error != "" {
  402. t.Fatal(lresp.Error)
  403. }
  404. // isolate the current leader with its followers.
  405. clus.Members[toIsolate].Pause()
  406. lreq := &pb.LeaseKeepAliveRequest{ID: lresp.ID}
  407. md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
  408. mctx := metadata.NewContext(context.Background(), md)
  409. ctx, cancel := context.WithCancel(mctx)
  410. defer cancel()
  411. lac, err := lc.LeaseKeepAlive(ctx)
  412. if err != nil {
  413. t.Fatal(err)
  414. }
  415. defer lac.CloseSend()
  416. // send keep alive to old leader until the old leader starts
  417. // to drop lease request.
  418. var expectedExp time.Time
  419. for {
  420. if err = lac.Send(lreq); err != nil {
  421. break
  422. }
  423. lkresp, rxerr := lac.Recv()
  424. if rxerr != nil {
  425. break
  426. }
  427. expectedExp = time.Now().Add(time.Duration(lkresp.TTL) * time.Second)
  428. time.Sleep(time.Duration(lkresp.TTL/2) * time.Second)
  429. }
  430. clus.Members[toIsolate].Resume()
  431. clus.waitLeader(t, clus.Members)
  432. // lease should not expire at the last received expire deadline.
  433. time.Sleep(time.Until(expectedExp) - 500*time.Millisecond)
  434. if !leaseExist(t, clus, lresp.ID) {
  435. t.Error("unexpected lease not exists")
  436. }
  437. }
  438. // TestV3LeaseRequireLeader ensures that a Recv will get a leader
  439. // loss error if there is no leader.
  440. func TestV3LeaseRequireLeader(t *testing.T) {
  441. defer testutil.AfterTest(t)
  442. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  443. defer clus.Terminate(t)
  444. lc := toGRPC(clus.Client(0)).Lease
  445. clus.Members[1].Stop(t)
  446. clus.Members[2].Stop(t)
  447. md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
  448. mctx := metadata.NewContext(context.Background(), md)
  449. ctx, cancel := context.WithCancel(mctx)
  450. defer cancel()
  451. lac, err := lc.LeaseKeepAlive(ctx)
  452. if err != nil {
  453. t.Fatal(err)
  454. }
  455. donec := make(chan struct{})
  456. go func() {
  457. defer close(donec)
  458. resp, err := lac.Recv()
  459. if err == nil {
  460. t.Fatalf("got response %+v, expected error", resp)
  461. }
  462. if grpc.ErrorDesc(err) != rpctypes.ErrNoLeader.Error() {
  463. t.Errorf("err = %v, want %v", err, rpctypes.ErrNoLeader)
  464. }
  465. }()
  466. select {
  467. case <-time.After(5 * time.Second):
  468. t.Fatal("did not receive leader loss error (in 5-sec)")
  469. case <-donec:
  470. }
  471. }
  472. const fiveMinTTL int64 = 300
  473. // TestV3LeaseRecoverAndRevoke ensures that revoking a lease after restart deletes the attached key.
  474. func TestV3LeaseRecoverAndRevoke(t *testing.T) {
  475. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  476. defer clus.Terminate(t)
  477. kvc := toGRPC(clus.Client(0)).KV
  478. lsc := toGRPC(clus.Client(0)).Lease
  479. lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL})
  480. if err != nil {
  481. t.Fatal(err)
  482. }
  483. if lresp.Error != "" {
  484. t.Fatal(lresp.Error)
  485. }
  486. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID})
  487. if err != nil {
  488. t.Fatal(err)
  489. }
  490. // restart server and ensure lease still exists
  491. clus.Members[0].Stop(t)
  492. clus.Members[0].Restart(t)
  493. clus.waitLeader(t, clus.Members)
  494. // overwrite old client with newly dialed connection
  495. // otherwise, error with "grpc: RPC failed fast due to transport failure"
  496. nc, err := NewClientV3(clus.Members[0])
  497. if err != nil {
  498. t.Fatal(err)
  499. }
  500. kvc = toGRPC(nc).KV
  501. lsc = toGRPC(nc).Lease
  502. defer nc.Close()
  503. // revoke should delete the key
  504. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  505. if err != nil {
  506. t.Fatal(err)
  507. }
  508. rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  509. if err != nil {
  510. t.Fatal(err)
  511. }
  512. if len(rresp.Kvs) != 0 {
  513. t.Fatalf("lease removed but key remains")
  514. }
  515. }
  516. // TestV3LeaseRevokeAndRecover ensures that revoked key stays deleted after restart.
  517. func TestV3LeaseRevokeAndRecover(t *testing.T) {
  518. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  519. defer clus.Terminate(t)
  520. kvc := toGRPC(clus.Client(0)).KV
  521. lsc := toGRPC(clus.Client(0)).Lease
  522. lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL})
  523. if err != nil {
  524. t.Fatal(err)
  525. }
  526. if lresp.Error != "" {
  527. t.Fatal(lresp.Error)
  528. }
  529. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID})
  530. if err != nil {
  531. t.Fatal(err)
  532. }
  533. // revoke should delete the key
  534. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  535. if err != nil {
  536. t.Fatal(err)
  537. }
  538. // restart server and ensure revoked key doesn't exist
  539. clus.Members[0].Stop(t)
  540. clus.Members[0].Restart(t)
  541. clus.waitLeader(t, clus.Members)
  542. // overwrite old client with newly dialed connection
  543. // otherwise, error with "grpc: RPC failed fast due to transport failure"
  544. nc, err := NewClientV3(clus.Members[0])
  545. if err != nil {
  546. t.Fatal(err)
  547. }
  548. kvc = toGRPC(nc).KV
  549. defer nc.Close()
  550. rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  551. if err != nil {
  552. t.Fatal(err)
  553. }
  554. if len(rresp.Kvs) != 0 {
  555. t.Fatalf("lease removed but key remains")
  556. }
  557. }
  558. // TestV3LeaseRecoverKeyWithDetachedLease ensures that revoking a detached lease after restart
  559. // does not delete the key.
  560. func TestV3LeaseRecoverKeyWithDetachedLease(t *testing.T) {
  561. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  562. defer clus.Terminate(t)
  563. kvc := toGRPC(clus.Client(0)).KV
  564. lsc := toGRPC(clus.Client(0)).Lease
  565. lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL})
  566. if err != nil {
  567. t.Fatal(err)
  568. }
  569. if lresp.Error != "" {
  570. t.Fatal(lresp.Error)
  571. }
  572. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID})
  573. if err != nil {
  574. t.Fatal(err)
  575. }
  576. // overwrite lease with none
  577. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")})
  578. if err != nil {
  579. t.Fatal(err)
  580. }
  581. // restart server and ensure lease still exists
  582. clus.Members[0].Stop(t)
  583. clus.Members[0].Restart(t)
  584. clus.waitLeader(t, clus.Members)
  585. // overwrite old client with newly dialed connection
  586. // otherwise, error with "grpc: RPC failed fast due to transport failure"
  587. nc, err := NewClientV3(clus.Members[0])
  588. if err != nil {
  589. t.Fatal(err)
  590. }
  591. kvc = toGRPC(nc).KV
  592. lsc = toGRPC(nc).Lease
  593. defer nc.Close()
  594. // revoke the detached lease
  595. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  596. if err != nil {
  597. t.Fatal(err)
  598. }
  599. rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  600. if err != nil {
  601. t.Fatal(err)
  602. }
  603. if len(rresp.Kvs) != 1 {
  604. t.Fatalf("only detached lease removed, key should remain")
  605. }
  606. }
  607. func TestV3LeaseRecoverKeyWithMutipleLease(t *testing.T) {
  608. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  609. defer clus.Terminate(t)
  610. kvc := toGRPC(clus.Client(0)).KV
  611. lsc := toGRPC(clus.Client(0)).Lease
  612. var leaseIDs []int64
  613. for i := 0; i < 2; i++ {
  614. lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL})
  615. if err != nil {
  616. t.Fatal(err)
  617. }
  618. if lresp.Error != "" {
  619. t.Fatal(lresp.Error)
  620. }
  621. leaseIDs = append(leaseIDs, lresp.ID)
  622. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID})
  623. if err != nil {
  624. t.Fatal(err)
  625. }
  626. }
  627. // restart server and ensure lease still exists
  628. clus.Members[0].Stop(t)
  629. clus.Members[0].Restart(t)
  630. clus.waitLeader(t, clus.Members)
  631. for i, leaseID := range leaseIDs {
  632. if !leaseExist(t, clus, leaseID) {
  633. t.Errorf("#%d: unexpected lease not exists", i)
  634. }
  635. }
  636. // overwrite old client with newly dialed connection
  637. // otherwise, error with "grpc: RPC failed fast due to transport failure"
  638. nc, err := NewClientV3(clus.Members[0])
  639. if err != nil {
  640. t.Fatal(err)
  641. }
  642. kvc = toGRPC(nc).KV
  643. lsc = toGRPC(nc).Lease
  644. defer nc.Close()
  645. // revoke the old lease
  646. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: leaseIDs[0]})
  647. if err != nil {
  648. t.Fatal(err)
  649. }
  650. // key should still exist
  651. rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  652. if err != nil {
  653. t.Fatal(err)
  654. }
  655. if len(rresp.Kvs) != 1 {
  656. t.Fatalf("only detached lease removed, key should remain")
  657. }
  658. // revoke the latest lease
  659. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: leaseIDs[1]})
  660. if err != nil {
  661. t.Fatal(err)
  662. }
  663. rresp, err = kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  664. if err != nil {
  665. t.Fatal(err)
  666. }
  667. if len(rresp.Kvs) != 0 {
  668. t.Fatalf("lease removed but key remains")
  669. }
  670. }
  671. // acquireLeaseAndKey creates a new lease and creates an attached key.
  672. func acquireLeaseAndKey(clus *ClusterV3, key string) (int64, error) {
  673. // create lease
  674. lresp, err := toGRPC(clus.RandClient()).Lease.LeaseGrant(
  675. context.TODO(),
  676. &pb.LeaseGrantRequest{TTL: 1})
  677. if err != nil {
  678. return 0, err
  679. }
  680. if lresp.Error != "" {
  681. return 0, fmt.Errorf(lresp.Error)
  682. }
  683. // attach to key
  684. put := &pb.PutRequest{Key: []byte(key), Lease: lresp.ID}
  685. if _, err := toGRPC(clus.RandClient()).KV.Put(context.TODO(), put); err != nil {
  686. return 0, err
  687. }
  688. return lresp.ID, nil
  689. }
  690. // testLeaseRemoveLeasedKey performs some action while holding a lease with an
  691. // attached key "foo", then confirms the key is gone.
  692. func testLeaseRemoveLeasedKey(t *testing.T, act func(*ClusterV3, int64) error) {
  693. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  694. defer clus.Terminate(t)
  695. leaseID, err := acquireLeaseAndKey(clus, "foo")
  696. if err != nil {
  697. t.Fatal(err)
  698. }
  699. if err = act(clus, leaseID); err != nil {
  700. t.Fatal(err)
  701. }
  702. // confirm no key
  703. rreq := &pb.RangeRequest{Key: []byte("foo")}
  704. rresp, err := toGRPC(clus.RandClient()).KV.Range(context.TODO(), rreq)
  705. if err != nil {
  706. t.Fatal(err)
  707. }
  708. if len(rresp.Kvs) != 0 {
  709. t.Fatalf("lease removed but key remains")
  710. }
  711. }
  712. func leaseExist(t *testing.T, clus *ClusterV3, leaseID int64) bool {
  713. l := toGRPC(clus.RandClient()).Lease
  714. _, err := l.LeaseGrant(context.Background(), &pb.LeaseGrantRequest{ID: leaseID, TTL: 5})
  715. if err == nil {
  716. _, err = l.LeaseRevoke(context.Background(), &pb.LeaseRevokeRequest{ID: leaseID})
  717. if err != nil {
  718. t.Fatalf("failed to check lease %v", err)
  719. }
  720. return false
  721. }
  722. if eqErrGRPC(err, rpctypes.ErrGRPCLeaseExist) {
  723. return true
  724. }
  725. t.Fatalf("unexpecter error %v", err)
  726. return true
  727. }