v3_lease_test.go 21 KB

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