v3_lease_test.go 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785
  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 ensures client retriving nonexistent lease on a follower doesn't result node panic
  299. // related issue https://github.com/coreos/etcd/issues/6537
  300. func TestV3GetNonExistLease(t *testing.T) {
  301. defer testutil.AfterTest(t)
  302. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  303. defer clus.Terminate(t)
  304. ctx, cancel := context.WithCancel(context.Background())
  305. defer cancel()
  306. lc := toGRPC(clus.RandClient()).Lease
  307. lresp, err := lc.LeaseGrant(ctx, &pb.LeaseGrantRequest{TTL: 10})
  308. if err != nil {
  309. t.Errorf("failed to create lease %v", err)
  310. }
  311. _, err = lc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  312. if err != nil {
  313. t.Fatal(err)
  314. }
  315. leaseTTLr := &pb.LeaseTimeToLiveRequest{
  316. ID: lresp.ID,
  317. Keys: true,
  318. }
  319. for _, client := range clus.clients {
  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 \n", 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(expectedExp.Sub(time.Now()) - 500*time.Millisecond)
  429. if !leaseExist(t, clus, lresp.ID) {
  430. t.Error("unexpected lease not exists")
  431. }
  432. }
  433. const fiveMinTTL int64 = 300
  434. // TestV3LeaseRecoverAndRevoke ensures that revoking a lease after restart deletes the attached key.
  435. func TestV3LeaseRecoverAndRevoke(t *testing.T) {
  436. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  437. defer clus.Terminate(t)
  438. kvc := toGRPC(clus.Client(0)).KV
  439. lsc := toGRPC(clus.Client(0)).Lease
  440. lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL})
  441. if err != nil {
  442. t.Fatal(err)
  443. }
  444. if lresp.Error != "" {
  445. t.Fatal(lresp.Error)
  446. }
  447. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID})
  448. if err != nil {
  449. t.Fatal(err)
  450. }
  451. // restart server and ensure lease still exists
  452. clus.Members[0].Stop(t)
  453. clus.Members[0].Restart(t)
  454. clus.waitLeader(t, clus.Members)
  455. // overwrite old client with newly dialed connection
  456. // otherwise, error with "grpc: RPC failed fast due to transport failure"
  457. nc, err := NewClientV3(clus.Members[0])
  458. if err != nil {
  459. t.Fatal(err)
  460. }
  461. kvc = toGRPC(nc).KV
  462. lsc = toGRPC(nc).Lease
  463. defer nc.Close()
  464. // revoke should delete the key
  465. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  466. if err != nil {
  467. t.Fatal(err)
  468. }
  469. rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  470. if err != nil {
  471. t.Fatal(err)
  472. }
  473. if len(rresp.Kvs) != 0 {
  474. t.Fatalf("lease removed but key remains")
  475. }
  476. }
  477. // TestV3LeaseRevokeAndRecover ensures that revoked key stays deleted after restart.
  478. func TestV3LeaseRevokeAndRecover(t *testing.T) {
  479. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  480. defer clus.Terminate(t)
  481. kvc := toGRPC(clus.Client(0)).KV
  482. lsc := toGRPC(clus.Client(0)).Lease
  483. lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL})
  484. if err != nil {
  485. t.Fatal(err)
  486. }
  487. if lresp.Error != "" {
  488. t.Fatal(lresp.Error)
  489. }
  490. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID})
  491. if err != nil {
  492. t.Fatal(err)
  493. }
  494. // revoke should delete the key
  495. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  496. if err != nil {
  497. t.Fatal(err)
  498. }
  499. // restart server and ensure revoked key doesn't exist
  500. clus.Members[0].Stop(t)
  501. clus.Members[0].Restart(t)
  502. clus.waitLeader(t, clus.Members)
  503. // overwrite old client with newly dialed connection
  504. // otherwise, error with "grpc: RPC failed fast due to transport failure"
  505. nc, err := NewClientV3(clus.Members[0])
  506. if err != nil {
  507. t.Fatal(err)
  508. }
  509. kvc = toGRPC(nc).KV
  510. defer nc.Close()
  511. rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  512. if err != nil {
  513. t.Fatal(err)
  514. }
  515. if len(rresp.Kvs) != 0 {
  516. t.Fatalf("lease removed but key remains")
  517. }
  518. }
  519. // TestV3LeaseRecoverKeyWithDetachedLease ensures that revoking a detached lease after restart
  520. // does not delete the key.
  521. func TestV3LeaseRecoverKeyWithDetachedLease(t *testing.T) {
  522. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  523. defer clus.Terminate(t)
  524. kvc := toGRPC(clus.Client(0)).KV
  525. lsc := toGRPC(clus.Client(0)).Lease
  526. lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL})
  527. if err != nil {
  528. t.Fatal(err)
  529. }
  530. if lresp.Error != "" {
  531. t.Fatal(lresp.Error)
  532. }
  533. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID})
  534. if err != nil {
  535. t.Fatal(err)
  536. }
  537. // overwrite lease with none
  538. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")})
  539. if err != nil {
  540. t.Fatal(err)
  541. }
  542. // restart server and ensure lease still exists
  543. clus.Members[0].Stop(t)
  544. clus.Members[0].Restart(t)
  545. clus.waitLeader(t, clus.Members)
  546. // overwrite old client with newly dialed connection
  547. // otherwise, error with "grpc: RPC failed fast due to transport failure"
  548. nc, err := NewClientV3(clus.Members[0])
  549. if err != nil {
  550. t.Fatal(err)
  551. }
  552. kvc = toGRPC(nc).KV
  553. lsc = toGRPC(nc).Lease
  554. defer nc.Close()
  555. // revoke the detached lease
  556. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  557. if err != nil {
  558. t.Fatal(err)
  559. }
  560. rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  561. if err != nil {
  562. t.Fatal(err)
  563. }
  564. if len(rresp.Kvs) != 1 {
  565. t.Fatalf("only detached lease removed, key should remain")
  566. }
  567. }
  568. func TestV3LeaseRecoverKeyWithMutipleLease(t *testing.T) {
  569. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  570. defer clus.Terminate(t)
  571. kvc := toGRPC(clus.Client(0)).KV
  572. lsc := toGRPC(clus.Client(0)).Lease
  573. var leaseIDs []int64
  574. for i := 0; i < 2; i++ {
  575. lresp, err := lsc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: fiveMinTTL})
  576. if err != nil {
  577. t.Fatal(err)
  578. }
  579. if lresp.Error != "" {
  580. t.Fatal(lresp.Error)
  581. }
  582. leaseIDs = append(leaseIDs, lresp.ID)
  583. _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar"), Lease: lresp.ID})
  584. if err != nil {
  585. t.Fatal(err)
  586. }
  587. }
  588. // restart server and ensure lease still exists
  589. clus.Members[0].Stop(t)
  590. clus.Members[0].Restart(t)
  591. clus.waitLeader(t, clus.Members)
  592. for i, leaseID := range leaseIDs {
  593. if !leaseExist(t, clus, leaseID) {
  594. t.Errorf("#%d: unexpected lease not exists", i)
  595. }
  596. }
  597. // overwrite old client with newly dialed connection
  598. // otherwise, error with "grpc: RPC failed fast due to transport failure"
  599. nc, err := NewClientV3(clus.Members[0])
  600. if err != nil {
  601. t.Fatal(err)
  602. }
  603. kvc = toGRPC(nc).KV
  604. lsc = toGRPC(nc).Lease
  605. defer nc.Close()
  606. // revoke the old lease
  607. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: leaseIDs[0]})
  608. if err != nil {
  609. t.Fatal(err)
  610. }
  611. // key should still exist
  612. rresp, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  613. if err != nil {
  614. t.Fatal(err)
  615. }
  616. if len(rresp.Kvs) != 1 {
  617. t.Fatalf("only detached lease removed, key should remain")
  618. }
  619. // revoke the latest lease
  620. _, err = lsc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: leaseIDs[1]})
  621. if err != nil {
  622. t.Fatal(err)
  623. }
  624. rresp, err = kvc.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  625. if err != nil {
  626. t.Fatal(err)
  627. }
  628. if len(rresp.Kvs) != 0 {
  629. t.Fatalf("lease removed but key remains")
  630. }
  631. }
  632. // acquireLeaseAndKey creates a new lease and creates an attached key.
  633. func acquireLeaseAndKey(clus *ClusterV3, key string) (int64, error) {
  634. // create lease
  635. lresp, err := toGRPC(clus.RandClient()).Lease.LeaseGrant(
  636. context.TODO(),
  637. &pb.LeaseGrantRequest{TTL: 1})
  638. if err != nil {
  639. return 0, err
  640. }
  641. if lresp.Error != "" {
  642. return 0, fmt.Errorf(lresp.Error)
  643. }
  644. // attach to key
  645. put := &pb.PutRequest{Key: []byte(key), Lease: lresp.ID}
  646. if _, err := toGRPC(clus.RandClient()).KV.Put(context.TODO(), put); err != nil {
  647. return 0, err
  648. }
  649. return lresp.ID, nil
  650. }
  651. // testLeaseRemoveLeasedKey performs some action while holding a lease with an
  652. // attached key "foo", then confirms the key is gone.
  653. func testLeaseRemoveLeasedKey(t *testing.T, act func(*ClusterV3, int64) error) {
  654. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  655. defer clus.Terminate(t)
  656. leaseID, err := acquireLeaseAndKey(clus, "foo")
  657. if err != nil {
  658. t.Fatal(err)
  659. }
  660. if err = act(clus, leaseID); err != nil {
  661. t.Fatal(err)
  662. }
  663. // confirm no key
  664. rreq := &pb.RangeRequest{Key: []byte("foo")}
  665. rresp, err := toGRPC(clus.RandClient()).KV.Range(context.TODO(), rreq)
  666. if err != nil {
  667. t.Fatal(err)
  668. }
  669. if len(rresp.Kvs) != 0 {
  670. t.Fatalf("lease removed but key remains")
  671. }
  672. }
  673. func leaseExist(t *testing.T, clus *ClusterV3, leaseID int64) bool {
  674. l := toGRPC(clus.RandClient()).Lease
  675. _, err := l.LeaseGrant(context.Background(), &pb.LeaseGrantRequest{ID: leaseID, TTL: 5})
  676. if err == nil {
  677. _, err = l.LeaseRevoke(context.Background(), &pb.LeaseRevokeRequest{ID: leaseID})
  678. if err != nil {
  679. t.Fatalf("failed to check lease %v", err)
  680. }
  681. return false
  682. }
  683. if eqErrGRPC(err, rpctypes.ErrGRPCLeaseExist) {
  684. return true
  685. }
  686. t.Fatalf("unexpecter error %v", err)
  687. return true
  688. }