v3_grpc_test.go 38 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439
  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. "bytes"
  17. "fmt"
  18. "math/rand"
  19. "os"
  20. "reflect"
  21. "testing"
  22. "time"
  23. "github.com/coreos/etcd/etcdserver/api/v3rpc"
  24. "github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
  25. pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
  26. "github.com/coreos/etcd/pkg/testutil"
  27. "golang.org/x/net/context"
  28. "google.golang.org/grpc"
  29. "google.golang.org/grpc/metadata"
  30. )
  31. // TestV3PutOverwrite puts a key with the v3 api to a random cluster member,
  32. // overwrites it, then checks that the change was applied.
  33. func TestV3PutOverwrite(t *testing.T) {
  34. defer testutil.AfterTest(t)
  35. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  36. defer clus.Terminate(t)
  37. kvc := toGRPC(clus.RandClient()).KV
  38. key := []byte("foo")
  39. reqput := &pb.PutRequest{Key: key, Value: []byte("bar")}
  40. respput, err := kvc.Put(context.TODO(), reqput)
  41. if err != nil {
  42. t.Fatalf("couldn't put key (%v)", err)
  43. }
  44. // overwrite
  45. reqput.Value = []byte("baz")
  46. respput2, err := kvc.Put(context.TODO(), reqput)
  47. if err != nil {
  48. t.Fatalf("couldn't put key (%v)", err)
  49. }
  50. if respput2.Header.Revision <= respput.Header.Revision {
  51. t.Fatalf("expected newer revision on overwrite, got %v <= %v",
  52. respput2.Header.Revision, respput.Header.Revision)
  53. }
  54. reqrange := &pb.RangeRequest{Key: key}
  55. resprange, err := kvc.Range(context.TODO(), reqrange)
  56. if err != nil {
  57. t.Fatalf("couldn't get key (%v)", err)
  58. }
  59. if len(resprange.Kvs) != 1 {
  60. t.Fatalf("expected 1 key, got %v", len(resprange.Kvs))
  61. }
  62. kv := resprange.Kvs[0]
  63. if kv.ModRevision <= kv.CreateRevision {
  64. t.Errorf("expected modRev > createRev, got %d <= %d",
  65. kv.ModRevision, kv.CreateRevision)
  66. }
  67. if !reflect.DeepEqual(reqput.Value, kv.Value) {
  68. t.Errorf("expected value %v, got %v", reqput.Value, kv.Value)
  69. }
  70. }
  71. // TestPutRestart checks if a put after an unrelated member restart succeeds
  72. func TestV3PutRestart(t *testing.T) {
  73. defer testutil.AfterTest(t)
  74. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  75. defer clus.Terminate(t)
  76. kvIdx := rand.Intn(3)
  77. kvc := toGRPC(clus.Client(kvIdx)).KV
  78. stopIdx := kvIdx
  79. for stopIdx == kvIdx {
  80. stopIdx = rand.Intn(3)
  81. }
  82. clus.clients[stopIdx].Close()
  83. clus.Members[stopIdx].Stop(t)
  84. clus.Members[stopIdx].Restart(t)
  85. c, cerr := NewClientV3(clus.Members[stopIdx])
  86. if cerr != nil {
  87. t.Fatalf("cannot create client: %v", cerr)
  88. }
  89. clus.clients[stopIdx] = c
  90. ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
  91. defer cancel()
  92. reqput := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
  93. _, err := kvc.Put(ctx, reqput)
  94. if err != nil && err == ctx.Err() {
  95. t.Fatalf("expected grpc error, got local ctx error (%v)", err)
  96. }
  97. }
  98. // TestV3CompactCurrentRev ensures keys are present when compacting on current revision.
  99. func TestV3CompactCurrentRev(t *testing.T) {
  100. defer testutil.AfterTest(t)
  101. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  102. defer clus.Terminate(t)
  103. kvc := toGRPC(clus.RandClient()).KV
  104. preq := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
  105. for i := 0; i < 3; i++ {
  106. if _, err := kvc.Put(context.Background(), preq); err != nil {
  107. t.Fatalf("couldn't put key (%v)", err)
  108. }
  109. }
  110. // compact on current revision
  111. _, err := kvc.Compact(context.Background(), &pb.CompactionRequest{Revision: 4})
  112. if err != nil {
  113. t.Fatalf("couldn't compact kv space (%v)", err)
  114. }
  115. // key still exists?
  116. _, err = kvc.Range(context.Background(), &pb.RangeRequest{Key: []byte("foo")})
  117. if err != nil {
  118. t.Fatalf("couldn't get key after compaction (%v)", err)
  119. }
  120. }
  121. func TestV3TxnTooManyOps(t *testing.T) {
  122. defer testutil.AfterTest(t)
  123. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  124. defer clus.Terminate(t)
  125. kvc := toGRPC(clus.RandClient()).KV
  126. // unique keys
  127. i := new(int)
  128. keyf := func() []byte {
  129. *i++
  130. return []byte(fmt.Sprintf("key-%d", i))
  131. }
  132. addCompareOps := func(txn *pb.TxnRequest) {
  133. txn.Compare = append(txn.Compare,
  134. &pb.Compare{
  135. Result: pb.Compare_GREATER,
  136. Target: pb.Compare_CREATE,
  137. Key: keyf(),
  138. })
  139. }
  140. addSuccessOps := func(txn *pb.TxnRequest) {
  141. txn.Success = append(txn.Success,
  142. &pb.RequestOp{
  143. Request: &pb.RequestOp_RequestPut{
  144. RequestPut: &pb.PutRequest{
  145. Key: keyf(),
  146. Value: []byte("bar"),
  147. },
  148. },
  149. })
  150. }
  151. addFailureOps := func(txn *pb.TxnRequest) {
  152. txn.Failure = append(txn.Failure,
  153. &pb.RequestOp{
  154. Request: &pb.RequestOp_RequestPut{
  155. RequestPut: &pb.PutRequest{
  156. Key: keyf(),
  157. Value: []byte("bar"),
  158. },
  159. },
  160. })
  161. }
  162. tests := []func(txn *pb.TxnRequest){
  163. addCompareOps,
  164. addSuccessOps,
  165. addFailureOps,
  166. }
  167. for i, tt := range tests {
  168. txn := &pb.TxnRequest{}
  169. for j := 0; j < v3rpc.MaxOpsPerTxn+1; j++ {
  170. tt(txn)
  171. }
  172. _, err := kvc.Txn(context.Background(), txn)
  173. if !eqErrGRPC(err, rpctypes.ErrGRPCTooManyOps) {
  174. t.Errorf("#%d: err = %v, want %v", i, err, rpctypes.ErrGRPCTooManyOps)
  175. }
  176. }
  177. }
  178. func TestV3TxnDuplicateKeys(t *testing.T) {
  179. defer testutil.AfterTest(t)
  180. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  181. defer clus.Terminate(t)
  182. putreq := &pb.RequestOp{Request: &pb.RequestOp_RequestPut{RequestPut: &pb.PutRequest{Key: []byte("abc"), Value: []byte("def")}}}
  183. delKeyReq := &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{
  184. RequestDeleteRange: &pb.DeleteRangeRequest{
  185. Key: []byte("abc"),
  186. },
  187. },
  188. }
  189. delInRangeReq := &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{
  190. RequestDeleteRange: &pb.DeleteRangeRequest{
  191. Key: []byte("a"), RangeEnd: []byte("b"),
  192. },
  193. },
  194. }
  195. delOutOfRangeReq := &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{
  196. RequestDeleteRange: &pb.DeleteRangeRequest{
  197. Key: []byte("abb"), RangeEnd: []byte("abc"),
  198. },
  199. },
  200. }
  201. kvc := toGRPC(clus.RandClient()).KV
  202. tests := []struct {
  203. txnSuccess []*pb.RequestOp
  204. werr error
  205. }{
  206. {
  207. txnSuccess: []*pb.RequestOp{putreq, putreq},
  208. werr: rpctypes.ErrGRPCDuplicateKey,
  209. },
  210. {
  211. txnSuccess: []*pb.RequestOp{putreq, delKeyReq},
  212. werr: rpctypes.ErrGRPCDuplicateKey,
  213. },
  214. {
  215. txnSuccess: []*pb.RequestOp{putreq, delInRangeReq},
  216. werr: rpctypes.ErrGRPCDuplicateKey,
  217. },
  218. {
  219. txnSuccess: []*pb.RequestOp{delKeyReq, delInRangeReq, delKeyReq, delInRangeReq},
  220. werr: nil,
  221. },
  222. {
  223. txnSuccess: []*pb.RequestOp{putreq, delOutOfRangeReq},
  224. werr: nil,
  225. },
  226. }
  227. for i, tt := range tests {
  228. txn := &pb.TxnRequest{Success: tt.txnSuccess}
  229. _, err := kvc.Txn(context.Background(), txn)
  230. if !eqErrGRPC(err, tt.werr) {
  231. t.Errorf("#%d: err = %v, want %v", i, err, tt.werr)
  232. }
  233. }
  234. }
  235. // Testv3TxnRevision tests that the transaction header revision is set as expected.
  236. func TestV3TxnRevision(t *testing.T) {
  237. defer testutil.AfterTest(t)
  238. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  239. defer clus.Terminate(t)
  240. kvc := toGRPC(clus.RandClient()).KV
  241. pr := &pb.PutRequest{Key: []byte("abc"), Value: []byte("def")}
  242. presp, err := kvc.Put(context.TODO(), pr)
  243. if err != nil {
  244. t.Fatal(err)
  245. }
  246. txnget := &pb.RequestOp{Request: &pb.RequestOp_RequestRange{RequestRange: &pb.RangeRequest{Key: []byte("abc")}}}
  247. txn := &pb.TxnRequest{Success: []*pb.RequestOp{txnget}}
  248. tresp, err := kvc.Txn(context.TODO(), txn)
  249. if err != nil {
  250. t.Fatal(err)
  251. }
  252. // did not update revision
  253. if presp.Header.Revision != tresp.Header.Revision {
  254. t.Fatalf("got rev %d, wanted rev %d", tresp.Header.Revision, presp.Header.Revision)
  255. }
  256. txndr := &pb.RequestOp{Request: &pb.RequestOp_RequestDeleteRange{RequestDeleteRange: &pb.DeleteRangeRequest{Key: []byte("def")}}}
  257. txn = &pb.TxnRequest{Success: []*pb.RequestOp{txndr}}
  258. tresp, err = kvc.Txn(context.TODO(), txn)
  259. if err != nil {
  260. t.Fatal(err)
  261. }
  262. // did not update revision
  263. if presp.Header.Revision != tresp.Header.Revision {
  264. t.Fatalf("got rev %d, wanted rev %d", tresp.Header.Revision, presp.Header.Revision)
  265. }
  266. txnput := &pb.RequestOp{Request: &pb.RequestOp_RequestPut{RequestPut: &pb.PutRequest{Key: []byte("abc"), Value: []byte("123")}}}
  267. txn = &pb.TxnRequest{Success: []*pb.RequestOp{txnput}}
  268. tresp, err = kvc.Txn(context.TODO(), txn)
  269. if err != nil {
  270. t.Fatal(err)
  271. }
  272. // updated revision
  273. if tresp.Header.Revision != presp.Header.Revision+1 {
  274. t.Fatalf("got rev %d, wanted rev %d", tresp.Header.Revision, presp.Header.Revision+1)
  275. }
  276. }
  277. // TestV3PutIgnoreValue ensures that writes with ignore_value overwrites with previous key-value pair.
  278. func TestV3PutIgnoreValue(t *testing.T) {
  279. defer testutil.AfterTest(t)
  280. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  281. defer clus.Terminate(t)
  282. kvc := toGRPC(clus.RandClient()).KV
  283. key, val := []byte("foo"), []byte("bar")
  284. putReq := pb.PutRequest{Key: key, Value: val}
  285. // create lease
  286. lc := toGRPC(clus.RandClient()).Lease
  287. lresp, err := lc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: 30})
  288. if err != nil {
  289. t.Fatal(err)
  290. }
  291. if lresp.Error != "" {
  292. t.Fatal(lresp.Error)
  293. }
  294. tests := []struct {
  295. putFunc func() error
  296. putErr error
  297. wleaseID int64
  298. }{
  299. { // put failure for non-existent key
  300. func() error {
  301. preq := putReq
  302. preq.IgnoreValue = true
  303. _, err := kvc.Put(context.TODO(), &preq)
  304. return err
  305. },
  306. rpctypes.ErrGRPCKeyNotFound,
  307. 0,
  308. },
  309. { // txn failure for non-existent key
  310. func() error {
  311. preq := putReq
  312. preq.Value = nil
  313. preq.IgnoreValue = true
  314. txn := &pb.TxnRequest{}
  315. txn.Success = append(txn.Success, &pb.RequestOp{
  316. Request: &pb.RequestOp_RequestPut{RequestPut: &preq}})
  317. _, err := kvc.Txn(context.TODO(), txn)
  318. return err
  319. },
  320. rpctypes.ErrGRPCKeyNotFound,
  321. 0,
  322. },
  323. { // put success
  324. func() error {
  325. _, err := kvc.Put(context.TODO(), &putReq)
  326. return err
  327. },
  328. nil,
  329. 0,
  330. },
  331. { // txn success, attach lease
  332. func() error {
  333. preq := putReq
  334. preq.Value = nil
  335. preq.Lease = lresp.ID
  336. preq.IgnoreValue = true
  337. txn := &pb.TxnRequest{}
  338. txn.Success = append(txn.Success, &pb.RequestOp{
  339. Request: &pb.RequestOp_RequestPut{RequestPut: &preq}})
  340. _, err := kvc.Txn(context.TODO(), txn)
  341. return err
  342. },
  343. nil,
  344. lresp.ID,
  345. },
  346. { // non-empty value with ignore_value should error
  347. func() error {
  348. preq := putReq
  349. preq.IgnoreValue = true
  350. _, err := kvc.Put(context.TODO(), &preq)
  351. return err
  352. },
  353. rpctypes.ErrGRPCValueProvided,
  354. 0,
  355. },
  356. { // overwrite with previous value, ensure no prev-kv is returned and lease is detached
  357. func() error {
  358. preq := putReq
  359. preq.Value = nil
  360. preq.IgnoreValue = true
  361. presp, err := kvc.Put(context.TODO(), &preq)
  362. if err != nil {
  363. return err
  364. }
  365. if presp.PrevKv != nil && len(presp.PrevKv.Key) != 0 {
  366. return fmt.Errorf("unexexpected previous key-value %v", presp.PrevKv)
  367. }
  368. return nil
  369. },
  370. nil,
  371. 0,
  372. },
  373. { // revoke lease, ensure detached key doesn't get deleted
  374. func() error {
  375. _, err := lc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  376. return err
  377. },
  378. nil,
  379. 0,
  380. },
  381. }
  382. for i, tt := range tests {
  383. if err := tt.putFunc(); !eqErrGRPC(err, tt.putErr) {
  384. t.Fatalf("#%d: err expected %v, got %v", i, tt.putErr, err)
  385. }
  386. if tt.putErr != nil {
  387. continue
  388. }
  389. rr, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: key})
  390. if err != nil {
  391. t.Fatalf("#%d: %v", i, err)
  392. }
  393. if len(rr.Kvs) != 1 {
  394. t.Fatalf("#%d: len(rr.KVs) expected 1, got %d", i, len(rr.Kvs))
  395. }
  396. if !bytes.Equal(rr.Kvs[0].Value, val) {
  397. t.Fatalf("#%d: value expected %q, got %q", i, val, rr.Kvs[0].Value)
  398. }
  399. if rr.Kvs[0].Lease != tt.wleaseID {
  400. t.Fatalf("#%d: lease ID expected %d, got %d", i, tt.wleaseID, rr.Kvs[0].Lease)
  401. }
  402. }
  403. }
  404. // TestV3PutIgnoreLease ensures that writes with ignore_lease uses previous lease for the key overwrites.
  405. func TestV3PutIgnoreLease(t *testing.T) {
  406. defer testutil.AfterTest(t)
  407. clus := NewClusterV3(t, &ClusterConfig{Size: 1})
  408. defer clus.Terminate(t)
  409. kvc := toGRPC(clus.RandClient()).KV
  410. // create lease
  411. lc := toGRPC(clus.RandClient()).Lease
  412. lresp, err := lc.LeaseGrant(context.TODO(), &pb.LeaseGrantRequest{TTL: 30})
  413. if err != nil {
  414. t.Fatal(err)
  415. }
  416. if lresp.Error != "" {
  417. t.Fatal(lresp.Error)
  418. }
  419. key, val, val1 := []byte("zoo"), []byte("bar"), []byte("bar1")
  420. putReq := pb.PutRequest{Key: key, Value: val}
  421. tests := []struct {
  422. putFunc func() error
  423. putErr error
  424. wleaseID int64
  425. wvalue []byte
  426. }{
  427. { // put failure for non-existent key
  428. func() error {
  429. preq := putReq
  430. preq.IgnoreLease = true
  431. _, err := kvc.Put(context.TODO(), &preq)
  432. return err
  433. },
  434. rpctypes.ErrGRPCKeyNotFound,
  435. 0,
  436. nil,
  437. },
  438. { // txn failure for non-existent key
  439. func() error {
  440. preq := putReq
  441. preq.IgnoreLease = true
  442. txn := &pb.TxnRequest{}
  443. txn.Success = append(txn.Success, &pb.RequestOp{
  444. Request: &pb.RequestOp_RequestPut{RequestPut: &preq}})
  445. _, err := kvc.Txn(context.TODO(), txn)
  446. return err
  447. },
  448. rpctypes.ErrGRPCKeyNotFound,
  449. 0,
  450. nil,
  451. },
  452. { // put success
  453. func() error {
  454. preq := putReq
  455. preq.Lease = lresp.ID
  456. _, err := kvc.Put(context.TODO(), &preq)
  457. return err
  458. },
  459. nil,
  460. lresp.ID,
  461. val,
  462. },
  463. { // txn success, modify value using 'ignore_lease' and ensure lease is not detached
  464. func() error {
  465. preq := putReq
  466. preq.Value = val1
  467. preq.IgnoreLease = true
  468. txn := &pb.TxnRequest{}
  469. txn.Success = append(txn.Success, &pb.RequestOp{
  470. Request: &pb.RequestOp_RequestPut{RequestPut: &preq}})
  471. _, err := kvc.Txn(context.TODO(), txn)
  472. return err
  473. },
  474. nil,
  475. lresp.ID,
  476. val1,
  477. },
  478. { // non-empty lease with ignore_lease should error
  479. func() error {
  480. preq := putReq
  481. preq.Lease = lresp.ID
  482. preq.IgnoreLease = true
  483. _, err := kvc.Put(context.TODO(), &preq)
  484. return err
  485. },
  486. rpctypes.ErrGRPCLeaseProvided,
  487. 0,
  488. nil,
  489. },
  490. { // overwrite with previous value, ensure no prev-kv is returned and lease is detached
  491. func() error {
  492. presp, err := kvc.Put(context.TODO(), &putReq)
  493. if err != nil {
  494. return err
  495. }
  496. if presp.PrevKv != nil && len(presp.PrevKv.Key) != 0 {
  497. return fmt.Errorf("unexexpected previous key-value %v", presp.PrevKv)
  498. }
  499. return nil
  500. },
  501. nil,
  502. 0,
  503. val,
  504. },
  505. { // revoke lease, ensure detached key doesn't get deleted
  506. func() error {
  507. _, err := lc.LeaseRevoke(context.TODO(), &pb.LeaseRevokeRequest{ID: lresp.ID})
  508. return err
  509. },
  510. nil,
  511. 0,
  512. val,
  513. },
  514. }
  515. for i, tt := range tests {
  516. if err := tt.putFunc(); !eqErrGRPC(err, tt.putErr) {
  517. t.Fatalf("#%d: err expected %v, got %v", i, tt.putErr, err)
  518. }
  519. if tt.putErr != nil {
  520. continue
  521. }
  522. rr, err := kvc.Range(context.TODO(), &pb.RangeRequest{Key: key})
  523. if err != nil {
  524. t.Fatalf("#%d: %v", i, err)
  525. }
  526. if len(rr.Kvs) != 1 {
  527. t.Fatalf("#%d: len(rr.KVs) expected 1, got %d", i, len(rr.Kvs))
  528. }
  529. if !bytes.Equal(rr.Kvs[0].Value, tt.wvalue) {
  530. t.Fatalf("#%d: value expected %q, got %q", i, val, rr.Kvs[0].Value)
  531. }
  532. if rr.Kvs[0].Lease != tt.wleaseID {
  533. t.Fatalf("#%d: lease ID expected %d, got %d", i, tt.wleaseID, rr.Kvs[0].Lease)
  534. }
  535. }
  536. }
  537. // TestV3PutMissingLease ensures that a Put on a key with a bogus lease fails.
  538. func TestV3PutMissingLease(t *testing.T) {
  539. defer testutil.AfterTest(t)
  540. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  541. defer clus.Terminate(t)
  542. kvc := toGRPC(clus.RandClient()).KV
  543. key := []byte("foo")
  544. preq := &pb.PutRequest{Key: key, Lease: 123456}
  545. tests := []func(){
  546. // put case
  547. func() {
  548. if presp, err := kvc.Put(context.TODO(), preq); err == nil {
  549. t.Errorf("succeeded put key. req: %v. resp: %v", preq, presp)
  550. }
  551. },
  552. // txn success case
  553. func() {
  554. txn := &pb.TxnRequest{}
  555. txn.Success = append(txn.Success, &pb.RequestOp{
  556. Request: &pb.RequestOp_RequestPut{
  557. RequestPut: preq}})
  558. if tresp, err := kvc.Txn(context.TODO(), txn); err == nil {
  559. t.Errorf("succeeded txn success. req: %v. resp: %v", txn, tresp)
  560. }
  561. },
  562. // txn failure case
  563. func() {
  564. txn := &pb.TxnRequest{}
  565. txn.Failure = append(txn.Failure, &pb.RequestOp{
  566. Request: &pb.RequestOp_RequestPut{
  567. RequestPut: preq}})
  568. cmp := &pb.Compare{
  569. Result: pb.Compare_GREATER,
  570. Target: pb.Compare_CREATE,
  571. Key: []byte("bar"),
  572. }
  573. txn.Compare = append(txn.Compare, cmp)
  574. if tresp, err := kvc.Txn(context.TODO(), txn); err == nil {
  575. t.Errorf("succeeded txn failure. req: %v. resp: %v", txn, tresp)
  576. }
  577. },
  578. // ignore bad lease in failure on success txn
  579. func() {
  580. txn := &pb.TxnRequest{}
  581. rreq := &pb.RangeRequest{Key: []byte("bar")}
  582. txn.Success = append(txn.Success, &pb.RequestOp{
  583. Request: &pb.RequestOp_RequestRange{
  584. RequestRange: rreq}})
  585. txn.Failure = append(txn.Failure, &pb.RequestOp{
  586. Request: &pb.RequestOp_RequestPut{
  587. RequestPut: preq}})
  588. if tresp, err := kvc.Txn(context.TODO(), txn); err != nil {
  589. t.Errorf("failed good txn. req: %v. resp: %v", txn, tresp)
  590. }
  591. },
  592. }
  593. for i, f := range tests {
  594. f()
  595. // key shouldn't have been stored
  596. rreq := &pb.RangeRequest{Key: key}
  597. rresp, err := kvc.Range(context.TODO(), rreq)
  598. if err != nil {
  599. t.Errorf("#%d. could not rangereq (%v)", i, err)
  600. } else if len(rresp.Kvs) != 0 {
  601. t.Errorf("#%d. expected no keys, got %v", i, rresp)
  602. }
  603. }
  604. }
  605. // TestV3DeleteRange tests various edge cases in the DeleteRange API.
  606. func TestV3DeleteRange(t *testing.T) {
  607. defer testutil.AfterTest(t)
  608. tests := []struct {
  609. keySet []string
  610. begin string
  611. end string
  612. prevKV bool
  613. wantSet [][]byte
  614. deleted int64
  615. }{
  616. // delete middle
  617. {
  618. []string{"foo", "foo/abc", "fop"},
  619. "foo/", "fop", false,
  620. [][]byte{[]byte("foo"), []byte("fop")}, 1,
  621. },
  622. // no delete
  623. {
  624. []string{"foo", "foo/abc", "fop"},
  625. "foo/", "foo/", false,
  626. [][]byte{[]byte("foo"), []byte("foo/abc"), []byte("fop")}, 0,
  627. },
  628. // delete first
  629. {
  630. []string{"foo", "foo/abc", "fop"},
  631. "fo", "fop", false,
  632. [][]byte{[]byte("fop")}, 2,
  633. },
  634. // delete tail
  635. {
  636. []string{"foo", "foo/abc", "fop"},
  637. "foo/", "fos", false,
  638. [][]byte{[]byte("foo")}, 2,
  639. },
  640. // delete exact
  641. {
  642. []string{"foo", "foo/abc", "fop"},
  643. "foo/abc", "", false,
  644. [][]byte{[]byte("foo"), []byte("fop")}, 1,
  645. },
  646. // delete none, [x,x)
  647. {
  648. []string{"foo"},
  649. "foo", "foo", false,
  650. [][]byte{[]byte("foo")}, 0,
  651. },
  652. // delete middle with preserveKVs set
  653. {
  654. []string{"foo", "foo/abc", "fop"},
  655. "foo/", "fop", true,
  656. [][]byte{[]byte("foo"), []byte("fop")}, 1,
  657. },
  658. }
  659. for i, tt := range tests {
  660. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  661. kvc := toGRPC(clus.RandClient()).KV
  662. ks := tt.keySet
  663. for j := range ks {
  664. reqput := &pb.PutRequest{Key: []byte(ks[j]), Value: []byte{}}
  665. _, err := kvc.Put(context.TODO(), reqput)
  666. if err != nil {
  667. t.Fatalf("couldn't put key (%v)", err)
  668. }
  669. }
  670. dreq := &pb.DeleteRangeRequest{
  671. Key: []byte(tt.begin),
  672. RangeEnd: []byte(tt.end),
  673. PrevKv: tt.prevKV,
  674. }
  675. dresp, err := kvc.DeleteRange(context.TODO(), dreq)
  676. if err != nil {
  677. t.Fatalf("couldn't delete range on test %d (%v)", i, err)
  678. }
  679. if tt.deleted != dresp.Deleted {
  680. t.Errorf("expected %d on test %v, got %d", tt.deleted, i, dresp.Deleted)
  681. }
  682. if tt.prevKV {
  683. if len(dresp.PrevKvs) != int(dresp.Deleted) {
  684. t.Errorf("preserve %d keys, want %d", len(dresp.PrevKvs), dresp.Deleted)
  685. }
  686. }
  687. rreq := &pb.RangeRequest{Key: []byte{0x0}, RangeEnd: []byte{0xff}}
  688. rresp, err := kvc.Range(context.TODO(), rreq)
  689. if err != nil {
  690. t.Errorf("couldn't get range on test %v (%v)", i, err)
  691. }
  692. if dresp.Header.Revision != rresp.Header.Revision {
  693. t.Errorf("expected revision %v, got %v",
  694. dresp.Header.Revision, rresp.Header.Revision)
  695. }
  696. keys := [][]byte{}
  697. for j := range rresp.Kvs {
  698. keys = append(keys, rresp.Kvs[j].Key)
  699. }
  700. if !reflect.DeepEqual(tt.wantSet, keys) {
  701. t.Errorf("expected %v on test %v, got %v", tt.wantSet, i, keys)
  702. }
  703. // can't defer because tcp ports will be in use
  704. clus.Terminate(t)
  705. }
  706. }
  707. // TestV3TxnInvalidRange tests that invalid ranges are rejected in txns.
  708. func TestV3TxnInvalidRange(t *testing.T) {
  709. defer testutil.AfterTest(t)
  710. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  711. defer clus.Terminate(t)
  712. kvc := toGRPC(clus.RandClient()).KV
  713. preq := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
  714. for i := 0; i < 3; i++ {
  715. _, err := kvc.Put(context.Background(), preq)
  716. if err != nil {
  717. t.Fatalf("couldn't put key (%v)", err)
  718. }
  719. }
  720. _, err := kvc.Compact(context.Background(), &pb.CompactionRequest{Revision: 2})
  721. if err != nil {
  722. t.Fatalf("couldn't compact kv space (%v)", err)
  723. }
  724. // future rev
  725. txn := &pb.TxnRequest{}
  726. txn.Success = append(txn.Success, &pb.RequestOp{
  727. Request: &pb.RequestOp_RequestPut{
  728. RequestPut: preq}})
  729. rreq := &pb.RangeRequest{Key: []byte("foo"), Revision: 100}
  730. txn.Success = append(txn.Success, &pb.RequestOp{
  731. Request: &pb.RequestOp_RequestRange{
  732. RequestRange: rreq}})
  733. if _, err := kvc.Txn(context.TODO(), txn); !eqErrGRPC(err, rpctypes.ErrGRPCFutureRev) {
  734. t.Errorf("err = %v, want %v", err, rpctypes.ErrGRPCFutureRev)
  735. }
  736. // compacted rev
  737. tv, _ := txn.Success[1].Request.(*pb.RequestOp_RequestRange)
  738. tv.RequestRange.Revision = 1
  739. if _, err := kvc.Txn(context.TODO(), txn); !eqErrGRPC(err, rpctypes.ErrGRPCCompacted) {
  740. t.Errorf("err = %v, want %v", err, rpctypes.ErrGRPCCompacted)
  741. }
  742. }
  743. func TestV3TooLargeRequest(t *testing.T) {
  744. defer testutil.AfterTest(t)
  745. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  746. defer clus.Terminate(t)
  747. kvc := toGRPC(clus.RandClient()).KV
  748. // 2MB request value
  749. largeV := make([]byte, 2*1024*1024)
  750. preq := &pb.PutRequest{Key: []byte("foo"), Value: largeV}
  751. _, err := kvc.Put(context.Background(), preq)
  752. if !eqErrGRPC(err, rpctypes.ErrGRPCRequestTooLarge) {
  753. t.Errorf("err = %v, want %v", err, rpctypes.ErrGRPCRequestTooLarge)
  754. }
  755. }
  756. // TestV3Hash tests hash.
  757. func TestV3Hash(t *testing.T) {
  758. defer testutil.AfterTest(t)
  759. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  760. defer clus.Terminate(t)
  761. cli := clus.RandClient()
  762. kvc := toGRPC(cli).KV
  763. m := toGRPC(cli).Maintenance
  764. preq := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
  765. for i := 0; i < 3; i++ {
  766. _, err := kvc.Put(context.Background(), preq)
  767. if err != nil {
  768. t.Fatalf("couldn't put key (%v)", err)
  769. }
  770. }
  771. resp, err := m.Hash(context.Background(), &pb.HashRequest{})
  772. if err != nil || resp.Hash == 0 {
  773. t.Fatalf("couldn't hash (%v, hash %d)", err, resp.Hash)
  774. }
  775. }
  776. // TestV3StorageQuotaAPI tests the V3 server respects quotas at the API layer
  777. func TestV3StorageQuotaAPI(t *testing.T) {
  778. defer testutil.AfterTest(t)
  779. quotasize := int64(16 * os.Getpagesize())
  780. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  781. // Set a quota on one node
  782. clus.Members[0].QuotaBackendBytes = quotasize
  783. clus.Members[0].Stop(t)
  784. clus.Members[0].Restart(t)
  785. defer clus.Terminate(t)
  786. kvc := toGRPC(clus.Client(0)).KV
  787. waitForRestart(t, kvc)
  788. key := []byte("abc")
  789. // test small put that fits in quota
  790. smallbuf := make([]byte, 512)
  791. if _, err := kvc.Put(context.TODO(), &pb.PutRequest{Key: key, Value: smallbuf}); err != nil {
  792. t.Fatal(err)
  793. }
  794. // test big put
  795. bigbuf := make([]byte, quotasize)
  796. _, err := kvc.Put(context.TODO(), &pb.PutRequest{Key: key, Value: bigbuf})
  797. if !eqErrGRPC(err, rpctypes.ErrGRPCNoSpace) {
  798. t.Fatalf("big put got %v, expected %v", err, rpctypes.ErrGRPCNoSpace)
  799. }
  800. // test big txn
  801. puttxn := &pb.RequestOp{
  802. Request: &pb.RequestOp_RequestPut{
  803. RequestPut: &pb.PutRequest{
  804. Key: key,
  805. Value: bigbuf,
  806. },
  807. },
  808. }
  809. txnreq := &pb.TxnRequest{}
  810. txnreq.Success = append(txnreq.Success, puttxn)
  811. _, txnerr := kvc.Txn(context.TODO(), txnreq)
  812. if !eqErrGRPC(txnerr, rpctypes.ErrGRPCNoSpace) {
  813. t.Fatalf("big txn got %v, expected %v", err, rpctypes.ErrGRPCNoSpace)
  814. }
  815. }
  816. // TestV3StorageQuotaApply tests the V3 server respects quotas during apply
  817. func TestV3StorageQuotaApply(t *testing.T) {
  818. testutil.AfterTest(t)
  819. quotasize := int64(16 * os.Getpagesize())
  820. clus := NewClusterV3(t, &ClusterConfig{Size: 2})
  821. defer clus.Terminate(t)
  822. kvc0 := toGRPC(clus.Client(0)).KV
  823. kvc1 := toGRPC(clus.Client(1)).KV
  824. // Set a quota on one node
  825. clus.Members[0].QuotaBackendBytes = quotasize
  826. clus.Members[0].Stop(t)
  827. clus.Members[0].Restart(t)
  828. clus.waitLeader(t, clus.Members)
  829. waitForRestart(t, kvc0)
  830. key := []byte("abc")
  831. // test small put still works
  832. smallbuf := make([]byte, 1024)
  833. _, serr := kvc0.Put(context.TODO(), &pb.PutRequest{Key: key, Value: smallbuf})
  834. if serr != nil {
  835. t.Fatal(serr)
  836. }
  837. // test big put
  838. bigbuf := make([]byte, quotasize)
  839. _, err := kvc1.Put(context.TODO(), &pb.PutRequest{Key: key, Value: bigbuf})
  840. if err != nil {
  841. t.Fatal(err)
  842. }
  843. // quorum get should work regardless of whether alarm is raised
  844. _, err = kvc0.Range(context.TODO(), &pb.RangeRequest{Key: []byte("foo")})
  845. if err != nil {
  846. t.Fatal(err)
  847. }
  848. // wait until alarm is raised for sure-- poll the alarms
  849. stopc := time.After(5 * time.Second)
  850. for {
  851. req := &pb.AlarmRequest{Action: pb.AlarmRequest_GET}
  852. resp, aerr := clus.Members[0].s.Alarm(context.TODO(), req)
  853. if aerr != nil {
  854. t.Fatal(aerr)
  855. }
  856. if len(resp.Alarms) != 0 {
  857. break
  858. }
  859. select {
  860. case <-stopc:
  861. t.Fatalf("timed out waiting for alarm")
  862. case <-time.After(10 * time.Millisecond):
  863. }
  864. }
  865. // small quota machine should reject put
  866. if _, err := kvc0.Put(context.TODO(), &pb.PutRequest{Key: key, Value: smallbuf}); err == nil {
  867. t.Fatalf("past-quota instance should reject put")
  868. }
  869. // large quota machine should reject put
  870. if _, err := kvc1.Put(context.TODO(), &pb.PutRequest{Key: key, Value: smallbuf}); err == nil {
  871. t.Fatalf("past-quota instance should reject put")
  872. }
  873. // reset large quota node to ensure alarm persisted
  874. clus.Members[1].Stop(t)
  875. clus.Members[1].Restart(t)
  876. clus.waitLeader(t, clus.Members)
  877. if _, err := kvc1.Put(context.TODO(), &pb.PutRequest{Key: key, Value: smallbuf}); err == nil {
  878. t.Fatalf("alarmed instance should reject put after reset")
  879. }
  880. }
  881. // TestV3AlarmDeactivate ensures that space alarms can be deactivated so puts go through.
  882. func TestV3AlarmDeactivate(t *testing.T) {
  883. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  884. defer clus.Terminate(t)
  885. kvc := toGRPC(clus.RandClient()).KV
  886. mt := toGRPC(clus.RandClient()).Maintenance
  887. alarmReq := &pb.AlarmRequest{
  888. MemberID: 123,
  889. Action: pb.AlarmRequest_ACTIVATE,
  890. Alarm: pb.AlarmType_NOSPACE,
  891. }
  892. if _, err := mt.Alarm(context.TODO(), alarmReq); err != nil {
  893. t.Fatal(err)
  894. }
  895. key := []byte("abc")
  896. smallbuf := make([]byte, 512)
  897. _, err := kvc.Put(context.TODO(), &pb.PutRequest{Key: key, Value: smallbuf})
  898. if err == nil && !eqErrGRPC(err, rpctypes.ErrGRPCNoSpace) {
  899. t.Fatalf("put got %v, expected %v", err, rpctypes.ErrGRPCNoSpace)
  900. }
  901. alarmReq.Action = pb.AlarmRequest_DEACTIVATE
  902. if _, err = mt.Alarm(context.TODO(), alarmReq); err != nil {
  903. t.Fatal(err)
  904. }
  905. if _, err = kvc.Put(context.TODO(), &pb.PutRequest{Key: key, Value: smallbuf}); err != nil {
  906. t.Fatal(err)
  907. }
  908. }
  909. func TestV3RangeRequest(t *testing.T) {
  910. defer testutil.AfterTest(t)
  911. tests := []struct {
  912. putKeys []string
  913. reqs []pb.RangeRequest
  914. wresps [][]string
  915. wmores []bool
  916. }{
  917. // single key
  918. {
  919. []string{"foo", "bar"},
  920. []pb.RangeRequest{
  921. // exists
  922. {Key: []byte("foo")},
  923. // doesn't exist
  924. {Key: []byte("baz")},
  925. },
  926. [][]string{
  927. {"foo"},
  928. {},
  929. },
  930. []bool{false, false},
  931. },
  932. // multi-key
  933. {
  934. []string{"a", "b", "c", "d", "e"},
  935. []pb.RangeRequest{
  936. // all in range
  937. {Key: []byte("a"), RangeEnd: []byte("z")},
  938. // [b, d)
  939. {Key: []byte("b"), RangeEnd: []byte("d")},
  940. // out of range
  941. {Key: []byte("f"), RangeEnd: []byte("z")},
  942. // [c,c) = empty
  943. {Key: []byte("c"), RangeEnd: []byte("c")},
  944. // [d, b) = empty
  945. {Key: []byte("d"), RangeEnd: []byte("b")},
  946. // ["\0", "\0") => all in range
  947. {Key: []byte{0}, RangeEnd: []byte{0}},
  948. },
  949. [][]string{
  950. {"a", "b", "c", "d", "e"},
  951. {"b", "c"},
  952. {},
  953. {},
  954. {},
  955. {"a", "b", "c", "d", "e"},
  956. },
  957. []bool{false, false, false, false, false, false},
  958. },
  959. // revision
  960. {
  961. []string{"a", "b", "c", "d", "e"},
  962. []pb.RangeRequest{
  963. {Key: []byte("a"), RangeEnd: []byte("z"), Revision: 0},
  964. {Key: []byte("a"), RangeEnd: []byte("z"), Revision: 1},
  965. {Key: []byte("a"), RangeEnd: []byte("z"), Revision: 2},
  966. {Key: []byte("a"), RangeEnd: []byte("z"), Revision: 3},
  967. },
  968. [][]string{
  969. {"a", "b", "c", "d", "e"},
  970. {},
  971. {"a"},
  972. {"a", "b"},
  973. },
  974. []bool{false, false, false, false},
  975. },
  976. // limit
  977. {
  978. []string{"foo", "bar"},
  979. []pb.RangeRequest{
  980. // more
  981. {Key: []byte("a"), RangeEnd: []byte("z"), Limit: 1},
  982. // no more
  983. {Key: []byte("a"), RangeEnd: []byte("z"), Limit: 2},
  984. },
  985. [][]string{
  986. {"bar"},
  987. {"bar", "foo"},
  988. },
  989. []bool{true, false},
  990. },
  991. // sort
  992. {
  993. []string{"b", "a", "c", "d", "c"},
  994. []pb.RangeRequest{
  995. {
  996. Key: []byte("a"), RangeEnd: []byte("z"),
  997. Limit: 1,
  998. SortOrder: pb.RangeRequest_ASCEND,
  999. SortTarget: pb.RangeRequest_KEY,
  1000. },
  1001. {
  1002. Key: []byte("a"), RangeEnd: []byte("z"),
  1003. Limit: 1,
  1004. SortOrder: pb.RangeRequest_DESCEND,
  1005. SortTarget: pb.RangeRequest_KEY,
  1006. },
  1007. {
  1008. Key: []byte("a"), RangeEnd: []byte("z"),
  1009. Limit: 1,
  1010. SortOrder: pb.RangeRequest_ASCEND,
  1011. SortTarget: pb.RangeRequest_CREATE,
  1012. },
  1013. {
  1014. Key: []byte("a"), RangeEnd: []byte("z"),
  1015. Limit: 1,
  1016. SortOrder: pb.RangeRequest_DESCEND,
  1017. SortTarget: pb.RangeRequest_MOD,
  1018. },
  1019. {
  1020. Key: []byte("z"), RangeEnd: []byte("z"),
  1021. Limit: 1,
  1022. SortOrder: pb.RangeRequest_DESCEND,
  1023. SortTarget: pb.RangeRequest_CREATE,
  1024. },
  1025. { // sort ASCEND by default
  1026. Key: []byte("a"), RangeEnd: []byte("z"),
  1027. Limit: 10,
  1028. SortOrder: pb.RangeRequest_NONE,
  1029. SortTarget: pb.RangeRequest_CREATE,
  1030. },
  1031. },
  1032. [][]string{
  1033. {"a"},
  1034. {"d"},
  1035. {"b"},
  1036. {"c"},
  1037. {},
  1038. {"b", "a", "c", "d"},
  1039. },
  1040. []bool{true, true, true, true, false, false},
  1041. },
  1042. // min/max mod rev
  1043. {
  1044. []string{"rev2", "rev3", "rev4", "rev5", "rev6"},
  1045. []pb.RangeRequest{
  1046. {
  1047. Key: []byte{0}, RangeEnd: []byte{0},
  1048. MinModRevision: 3,
  1049. },
  1050. {
  1051. Key: []byte{0}, RangeEnd: []byte{0},
  1052. MaxModRevision: 3,
  1053. },
  1054. {
  1055. Key: []byte{0}, RangeEnd: []byte{0},
  1056. MinModRevision: 3,
  1057. MaxModRevision: 5,
  1058. },
  1059. {
  1060. Key: []byte{0}, RangeEnd: []byte{0},
  1061. MaxModRevision: 10,
  1062. },
  1063. },
  1064. [][]string{
  1065. {"rev3", "rev4", "rev5", "rev6"},
  1066. {"rev2", "rev3"},
  1067. {"rev3", "rev4", "rev5"},
  1068. {"rev2", "rev3", "rev4", "rev5", "rev6"},
  1069. },
  1070. []bool{false, false, false, false},
  1071. },
  1072. // min/max create rev
  1073. {
  1074. []string{"rev2", "rev3", "rev2", "rev2", "rev6", "rev3"},
  1075. []pb.RangeRequest{
  1076. {
  1077. Key: []byte{0}, RangeEnd: []byte{0},
  1078. MinCreateRevision: 3,
  1079. },
  1080. {
  1081. Key: []byte{0}, RangeEnd: []byte{0},
  1082. MaxCreateRevision: 3,
  1083. },
  1084. {
  1085. Key: []byte{0}, RangeEnd: []byte{0},
  1086. MinCreateRevision: 3,
  1087. MaxCreateRevision: 5,
  1088. },
  1089. {
  1090. Key: []byte{0}, RangeEnd: []byte{0},
  1091. MaxCreateRevision: 10,
  1092. },
  1093. },
  1094. [][]string{
  1095. {"rev3", "rev6"},
  1096. {"rev2", "rev3"},
  1097. {"rev3"},
  1098. {"rev2", "rev3", "rev6"},
  1099. },
  1100. []bool{false, false, false, false},
  1101. },
  1102. }
  1103. for i, tt := range tests {
  1104. clus := NewClusterV3(t, &ClusterConfig{Size: 3})
  1105. for _, k := range tt.putKeys {
  1106. kvc := toGRPC(clus.RandClient()).KV
  1107. req := &pb.PutRequest{Key: []byte(k), Value: []byte("bar")}
  1108. if _, err := kvc.Put(context.TODO(), req); err != nil {
  1109. t.Fatalf("#%d: couldn't put key (%v)", i, err)
  1110. }
  1111. }
  1112. for j, req := range tt.reqs {
  1113. kvc := toGRPC(clus.RandClient()).KV
  1114. resp, err := kvc.Range(context.TODO(), &req)
  1115. if err != nil {
  1116. t.Errorf("#%d.%d: Range error: %v", i, j, err)
  1117. continue
  1118. }
  1119. if len(resp.Kvs) != len(tt.wresps[j]) {
  1120. t.Errorf("#%d.%d: bad len(resp.Kvs). got = %d, want = %d, ", i, j, len(resp.Kvs), len(tt.wresps[j]))
  1121. continue
  1122. }
  1123. for k, wKey := range tt.wresps[j] {
  1124. respKey := string(resp.Kvs[k].Key)
  1125. if respKey != wKey {
  1126. t.Errorf("#%d.%d: key[%d]. got = %v, want = %v, ", i, j, k, respKey, wKey)
  1127. }
  1128. }
  1129. if resp.More != tt.wmores[j] {
  1130. t.Errorf("#%d.%d: bad more. got = %v, want = %v, ", i, j, resp.More, tt.wmores[j])
  1131. }
  1132. wrev := int64(len(tt.putKeys) + 1)
  1133. if resp.Header.Revision != wrev {
  1134. t.Errorf("#%d.%d: bad header revision. got = %d. want = %d", i, j, resp.Header.Revision, wrev)
  1135. }
  1136. }
  1137. clus.Terminate(t)
  1138. }
  1139. }
  1140. func newClusterV3NoClients(t *testing.T, cfg *ClusterConfig) *ClusterV3 {
  1141. cfg.UseGRPC = true
  1142. clus := &ClusterV3{cluster: NewClusterByConfig(t, cfg)}
  1143. clus.Launch(t)
  1144. return clus
  1145. }
  1146. // TestTLSGRPCRejectInsecureClient checks that connection is rejected if server is TLS but not client.
  1147. func TestTLSGRPCRejectInsecureClient(t *testing.T) {
  1148. defer testutil.AfterTest(t)
  1149. cfg := ClusterConfig{Size: 3, ClientTLS: &testTLSInfo}
  1150. clus := newClusterV3NoClients(t, &cfg)
  1151. defer clus.Terminate(t)
  1152. // nil out TLS field so client will use an insecure connection
  1153. clus.Members[0].ClientTLSInfo = nil
  1154. client, err := NewClientV3(clus.Members[0])
  1155. if err != nil && err != grpc.ErrClientConnTimeout {
  1156. t.Fatalf("unexpected error (%v)", err)
  1157. } else if client == nil {
  1158. // Ideally, no client would be returned. However, grpc will
  1159. // return a connection without trying to handshake first so
  1160. // the connection appears OK.
  1161. return
  1162. }
  1163. defer client.Close()
  1164. donec := make(chan error, 1)
  1165. go func() {
  1166. ctx, cancel := context.WithTimeout(context.TODO(), 5*time.Second)
  1167. reqput := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
  1168. _, perr := toGRPC(client).KV.Put(ctx, reqput)
  1169. cancel()
  1170. donec <- perr
  1171. }()
  1172. if perr := <-donec; perr == nil {
  1173. t.Fatalf("expected client error on put")
  1174. }
  1175. }
  1176. // TestTLSGRPCRejectSecureClient checks that connection is rejected if client is TLS but not server.
  1177. func TestTLSGRPCRejectSecureClient(t *testing.T) {
  1178. defer testutil.AfterTest(t)
  1179. cfg := ClusterConfig{Size: 3}
  1180. clus := newClusterV3NoClients(t, &cfg)
  1181. defer clus.Terminate(t)
  1182. clus.Members[0].ClientTLSInfo = &testTLSInfo
  1183. client, err := NewClientV3(clus.Members[0])
  1184. if client != nil || err == nil {
  1185. t.Fatalf("expected no client")
  1186. } else if err != grpc.ErrClientConnTimeout {
  1187. t.Fatalf("unexpected error (%v)", err)
  1188. }
  1189. }
  1190. // TestTLSGRPCAcceptSecureAll checks that connection is accepted if both client and server are TLS
  1191. func TestTLSGRPCAcceptSecureAll(t *testing.T) {
  1192. defer testutil.AfterTest(t)
  1193. cfg := ClusterConfig{Size: 3, ClientTLS: &testTLSInfo}
  1194. clus := newClusterV3NoClients(t, &cfg)
  1195. defer clus.Terminate(t)
  1196. client, err := NewClientV3(clus.Members[0])
  1197. if err != nil {
  1198. t.Fatalf("expected tls client (%v)", err)
  1199. }
  1200. defer client.Close()
  1201. reqput := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
  1202. if _, err := toGRPC(client).KV.Put(context.TODO(), reqput); err != nil {
  1203. t.Fatalf("unexpected error on put over tls (%v)", err)
  1204. }
  1205. }
  1206. func TestGRPCRequireLeader(t *testing.T) {
  1207. defer testutil.AfterTest(t)
  1208. cfg := ClusterConfig{Size: 3}
  1209. clus := newClusterV3NoClients(t, &cfg)
  1210. defer clus.Terminate(t)
  1211. clus.Members[1].Stop(t)
  1212. clus.Members[2].Stop(t)
  1213. client, err := NewClientV3(clus.Members[0])
  1214. if err != nil {
  1215. t.Fatalf("cannot create client: %v", err)
  1216. }
  1217. defer client.Close()
  1218. // wait for election timeout, then member[0] will not have a leader.
  1219. time.Sleep(time.Duration(3*electionTicks) * tickDuration)
  1220. md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
  1221. ctx := metadata.NewContext(context.Background(), md)
  1222. reqput := &pb.PutRequest{Key: []byte("foo"), Value: []byte("bar")}
  1223. if _, err := toGRPC(client).KV.Put(ctx, reqput); grpc.ErrorDesc(err) != rpctypes.ErrNoLeader.Error() {
  1224. t.Errorf("err = %v, want %v", err, rpctypes.ErrNoLeader)
  1225. }
  1226. }
  1227. func TestGRPCStreamRequireLeader(t *testing.T) {
  1228. defer testutil.AfterTest(t)
  1229. cfg := ClusterConfig{Size: 3}
  1230. clus := newClusterV3NoClients(t, &cfg)
  1231. defer clus.Terminate(t)
  1232. client, err := NewClientV3(clus.Members[0])
  1233. if err != nil {
  1234. t.Fatalf("failed to create client (%v)", err)
  1235. }
  1236. defer client.Close()
  1237. wAPI := toGRPC(client).Watch
  1238. md := metadata.Pairs(rpctypes.MetadataRequireLeaderKey, rpctypes.MetadataHasLeader)
  1239. ctx := metadata.NewContext(context.Background(), md)
  1240. wStream, err := wAPI.Watch(ctx)
  1241. if err != nil {
  1242. t.Fatalf("wAPI.Watch error: %v", err)
  1243. }
  1244. clus.Members[1].Stop(t)
  1245. clus.Members[2].Stop(t)
  1246. // existing stream should be rejected
  1247. _, err = wStream.Recv()
  1248. if grpc.ErrorDesc(err) != rpctypes.ErrNoLeader.Error() {
  1249. t.Errorf("err = %v, want %v", err, rpctypes.ErrNoLeader)
  1250. }
  1251. // new stream should also be rejected
  1252. wStream, err = wAPI.Watch(ctx)
  1253. if err != nil {
  1254. t.Fatalf("wAPI.Watch error: %v", err)
  1255. }
  1256. _, err = wStream.Recv()
  1257. if grpc.ErrorDesc(err) != rpctypes.ErrNoLeader.Error() {
  1258. t.Errorf("err = %v, want %v", err, rpctypes.ErrNoLeader)
  1259. }
  1260. clus.Members[1].Restart(t)
  1261. clus.Members[2].Restart(t)
  1262. clus.waitLeader(t, clus.Members)
  1263. time.Sleep(time.Duration(2*electionTicks) * tickDuration)
  1264. // new stream should also be OK now after we restarted the other members
  1265. wStream, err = wAPI.Watch(ctx)
  1266. if err != nil {
  1267. t.Fatalf("wAPI.Watch error: %v", err)
  1268. }
  1269. wreq := &pb.WatchRequest{
  1270. RequestUnion: &pb.WatchRequest_CreateRequest{
  1271. CreateRequest: &pb.WatchCreateRequest{Key: []byte("foo")},
  1272. },
  1273. }
  1274. err = wStream.Send(wreq)
  1275. if err != nil {
  1276. t.Errorf("err = %v, want nil", err)
  1277. }
  1278. }
  1279. func eqErrGRPC(err1 error, err2 error) bool {
  1280. return !(err1 == nil && err2 != nil) || err1.Error() == err2.Error()
  1281. }
  1282. // waitForRestart tries a range request until the client's server responds.
  1283. // This is mainly a stop-gap function until grpcproxy's KVClient adapter
  1284. // (and by extension, clientv3) supports grpc.CallOption pass-through so
  1285. // FailFast=false works with Put.
  1286. func waitForRestart(t *testing.T, kvc pb.KVClient) {
  1287. req := &pb.RangeRequest{Key: []byte("_"), Serializable: true}
  1288. if _, err := kvc.Range(context.TODO(), req, grpc.FailFast(false)); err != nil {
  1289. t.Fatal(err)
  1290. }
  1291. }