server_test.go 27 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082
  1. package etcdserver
  2. import (
  3. "fmt"
  4. "math/rand"
  5. "reflect"
  6. "sync"
  7. "testing"
  8. "time"
  9. pb "github.com/coreos/etcd/etcdserver/etcdserverpb"
  10. "github.com/coreos/etcd/raft"
  11. "github.com/coreos/etcd/raft/raftpb"
  12. "github.com/coreos/etcd/store"
  13. "github.com/coreos/etcd/testutil"
  14. "github.com/coreos/etcd/third_party/code.google.com/p/go.net/context"
  15. )
  16. func TestGetExpirationTime(t *testing.T) {
  17. tests := []struct {
  18. r pb.Request
  19. want time.Time
  20. }{
  21. {
  22. pb.Request{Expiration: 0},
  23. time.Time{},
  24. },
  25. {
  26. pb.Request{Expiration: 60000},
  27. time.Unix(0, 60000),
  28. },
  29. {
  30. pb.Request{Expiration: -60000},
  31. time.Unix(0, -60000),
  32. },
  33. }
  34. for i, tt := range tests {
  35. got := getExpirationTime(&tt.r)
  36. if !reflect.DeepEqual(tt.want, got) {
  37. t.Errorf("#%d: incorrect expiration time: want=%v got=%v", i, tt.want, got)
  38. }
  39. }
  40. }
  41. // TestDoLocalAction tests requests which do not need to go through raft to be applied,
  42. // and are served through local data.
  43. func TestDoLocalAction(t *testing.T) {
  44. tests := []struct {
  45. req pb.Request
  46. wresp Response
  47. werr error
  48. wactions []action
  49. }{
  50. {
  51. pb.Request{Method: "GET", Id: 1, Wait: true},
  52. Response{Watcher: &stubWatcher{}}, nil, []action{action{name: "Watch"}},
  53. },
  54. {
  55. pb.Request{Method: "GET", Id: 1},
  56. Response{Event: &store.Event{}}, nil,
  57. []action{
  58. action{
  59. name: "Get",
  60. params: []interface{}{"", false, false},
  61. },
  62. },
  63. },
  64. {
  65. pb.Request{Method: "BADMETHOD", Id: 1},
  66. Response{}, ErrUnknownMethod, []action{},
  67. },
  68. }
  69. for i, tt := range tests {
  70. st := &storeRecorder{}
  71. srv := &EtcdServer{Store: st}
  72. resp, err := srv.Do(context.TODO(), tt.req)
  73. if err != tt.werr {
  74. t.Fatalf("#%d: err = %+v, want %+v", i, err, tt.werr)
  75. }
  76. if !reflect.DeepEqual(resp, tt.wresp) {
  77. t.Errorf("#%d: resp = %+v, want %+v", i, resp, tt.wresp)
  78. }
  79. gaction := st.Action()
  80. if !reflect.DeepEqual(gaction, tt.wactions) {
  81. t.Errorf("#%d: action = %+v, want %+v", i, gaction, tt.wactions)
  82. }
  83. }
  84. }
  85. // TestDoBadLocalAction tests server requests which do not need to go through consensus,
  86. // and return errors when they fetch from local data.
  87. func TestDoBadLocalAction(t *testing.T) {
  88. storeErr := fmt.Errorf("bah")
  89. tests := []struct {
  90. req pb.Request
  91. wactions []action
  92. }{
  93. {
  94. pb.Request{Method: "GET", Id: 1, Wait: true},
  95. []action{action{name: "Watch"}},
  96. },
  97. {
  98. pb.Request{Method: "GET", Id: 1},
  99. []action{action{name: "Get"}},
  100. },
  101. }
  102. for i, tt := range tests {
  103. st := &errStoreRecorder{err: storeErr}
  104. srv := &EtcdServer{Store: st}
  105. resp, err := srv.Do(context.Background(), tt.req)
  106. if err != storeErr {
  107. t.Fatalf("#%d: err = %+v, want %+v", i, err, storeErr)
  108. }
  109. if !reflect.DeepEqual(resp, Response{}) {
  110. t.Errorf("#%d: resp = %+v, want %+v", i, resp, Response{})
  111. }
  112. gaction := st.Action()
  113. if !reflect.DeepEqual(gaction, tt.wactions) {
  114. t.Errorf("#%d: action = %+v, want %+v", i, gaction, tt.wactions)
  115. }
  116. }
  117. }
  118. func TestApply(t *testing.T) {
  119. tests := []struct {
  120. req pb.Request
  121. wresp Response
  122. wactions []action
  123. }{
  124. // POST ==> Create
  125. {
  126. pb.Request{Method: "POST", Id: 1},
  127. Response{Event: &store.Event{}},
  128. []action{
  129. action{
  130. name: "Create",
  131. params: []interface{}{"", false, "", true, time.Time{}},
  132. },
  133. },
  134. },
  135. // PUT ==> Set
  136. {
  137. pb.Request{Method: "PUT", Id: 1},
  138. Response{Event: &store.Event{}}, []action{action{name: "Set"}},
  139. },
  140. // PUT with PrevExist=true ==> Update
  141. {
  142. pb.Request{Method: "PUT", Id: 1, PrevExist: boolp(true)},
  143. Response{Event: &store.Event{}},
  144. []action{
  145. action{
  146. name: "Update",
  147. params: []interface{}{"", "", time.Time{}},
  148. },
  149. },
  150. },
  151. // PUT with PrevExist=false ==> Create
  152. {
  153. pb.Request{Method: "PUT", Id: 1, PrevExist: boolp(false)},
  154. Response{Event: &store.Event{}},
  155. []action{
  156. action{
  157. name: "Create",
  158. params: []interface{}{"", false, "", false, time.Time{}},
  159. },
  160. },
  161. },
  162. // PUT with PrevExist=true *and* PrevIndex set ==> Update
  163. // TODO(jonboulle): is this expected?!
  164. {
  165. pb.Request{Method: "PUT", Id: 1, PrevExist: boolp(true), PrevIndex: 1},
  166. Response{Event: &store.Event{}},
  167. []action{
  168. action{
  169. name: "Update",
  170. params: []interface{}{"", "", time.Time{}},
  171. },
  172. },
  173. },
  174. // PUT with PrevExist=false *and* PrevIndex set ==> Create
  175. // TODO(jonboulle): is this expected?!
  176. {
  177. pb.Request{Method: "PUT", Id: 1, PrevExist: boolp(false), PrevIndex: 1},
  178. Response{Event: &store.Event{}},
  179. []action{
  180. action{
  181. name: "Create",
  182. params: []interface{}{"", false, "", false, time.Time{}},
  183. },
  184. },
  185. },
  186. // PUT with PrevIndex set ==> CompareAndSwap
  187. {
  188. pb.Request{Method: "PUT", Id: 1, PrevIndex: 1},
  189. Response{Event: &store.Event{}},
  190. []action{
  191. action{
  192. name: "CompareAndSwap",
  193. params: []interface{}{"", "", uint64(1), "", time.Time{}},
  194. },
  195. },
  196. },
  197. // PUT with PrevValue set ==> CompareAndSwap
  198. {
  199. pb.Request{Method: "PUT", Id: 1, PrevValue: "bar"},
  200. Response{Event: &store.Event{}},
  201. []action{
  202. action{
  203. name: "CompareAndSwap",
  204. params: []interface{}{"", "bar", uint64(0), "", time.Time{}},
  205. },
  206. },
  207. },
  208. // PUT with PrevIndex and PrevValue set ==> CompareAndSwap
  209. {
  210. pb.Request{Method: "PUT", Id: 1, PrevIndex: 1, PrevValue: "bar"},
  211. Response{Event: &store.Event{}},
  212. []action{
  213. action{
  214. name: "CompareAndSwap",
  215. params: []interface{}{"", "bar", uint64(1), "", time.Time{}},
  216. },
  217. },
  218. },
  219. // DELETE ==> Delete
  220. {
  221. pb.Request{Method: "DELETE", Id: 1},
  222. Response{Event: &store.Event{}},
  223. []action{
  224. action{
  225. name: "Delete",
  226. params: []interface{}{"", false, false},
  227. },
  228. },
  229. },
  230. // DELETE with PrevIndex set ==> CompareAndDelete
  231. {
  232. pb.Request{Method: "DELETE", Id: 1, PrevIndex: 1},
  233. Response{Event: &store.Event{}},
  234. []action{
  235. action{
  236. name: "CompareAndDelete",
  237. params: []interface{}{"", "", uint64(1)},
  238. },
  239. },
  240. },
  241. // DELETE with PrevValue set ==> CompareAndDelete
  242. {
  243. pb.Request{Method: "DELETE", Id: 1, PrevValue: "bar"},
  244. Response{Event: &store.Event{}},
  245. []action{
  246. action{
  247. name: "CompareAndDelete",
  248. params: []interface{}{"", "bar", uint64(0)},
  249. },
  250. },
  251. },
  252. // DELETE with PrevIndex *and* PrevValue set ==> CompareAndDelete
  253. {
  254. pb.Request{Method: "DELETE", Id: 1, PrevIndex: 5, PrevValue: "bar"},
  255. Response{Event: &store.Event{}},
  256. []action{
  257. action{
  258. name: "CompareAndDelete",
  259. params: []interface{}{"", "bar", uint64(5)},
  260. },
  261. },
  262. },
  263. // QGET ==> Get
  264. {
  265. pb.Request{Method: "QGET", Id: 1},
  266. Response{Event: &store.Event{}},
  267. []action{
  268. action{
  269. name: "Get",
  270. params: []interface{}{"", false, false},
  271. },
  272. },
  273. },
  274. // SYNC ==> DeleteExpiredKeys
  275. {
  276. pb.Request{Method: "SYNC", Id: 1},
  277. Response{},
  278. []action{
  279. action{
  280. name: "DeleteExpiredKeys",
  281. params: []interface{}{time.Unix(0, 0)},
  282. },
  283. },
  284. },
  285. {
  286. pb.Request{Method: "SYNC", Id: 1, Time: 12345},
  287. Response{},
  288. []action{
  289. action{
  290. name: "DeleteExpiredKeys",
  291. params: []interface{}{time.Unix(0, 12345)},
  292. },
  293. },
  294. },
  295. // Unknown method - error
  296. {
  297. pb.Request{Method: "BADMETHOD", Id: 1},
  298. Response{err: ErrUnknownMethod},
  299. []action{},
  300. },
  301. }
  302. for i, tt := range tests {
  303. st := &storeRecorder{}
  304. srv := &EtcdServer{Store: st}
  305. resp := srv.apply(tt.req)
  306. if !reflect.DeepEqual(resp, tt.wresp) {
  307. t.Errorf("#%d: resp = %+v, want %+v", i, resp, tt.wresp)
  308. }
  309. gaction := st.Action()
  310. if !reflect.DeepEqual(gaction, tt.wactions) {
  311. t.Errorf("#%d: action = %#v, want %#v", i, gaction, tt.wactions)
  312. }
  313. }
  314. }
  315. func TestClusterOf1(t *testing.T) { testServer(t, 1) }
  316. func TestClusterOf3(t *testing.T) { testServer(t, 3) }
  317. func testServer(t *testing.T, ns int64) {
  318. ctx, cancel := context.WithCancel(context.Background())
  319. defer cancel()
  320. ss := make([]*EtcdServer, ns)
  321. send := func(msgs []raftpb.Message) {
  322. for _, m := range msgs {
  323. t.Logf("m = %+v\n", m)
  324. ss[m.To-1].Node.Step(ctx, m)
  325. }
  326. }
  327. peers := make([]int64, ns)
  328. for i := int64(0); i < ns; i++ {
  329. peers[i] = i + 1
  330. }
  331. for i := int64(0); i < ns; i++ {
  332. id := i + 1
  333. n := raft.StartNode(id, peers, 10, 1)
  334. tk := time.NewTicker(10 * time.Millisecond)
  335. defer tk.Stop()
  336. srv := &EtcdServer{
  337. Node: n,
  338. Store: store.New(),
  339. Send: send,
  340. Storage: &storageRecorder{},
  341. Ticker: tk.C,
  342. }
  343. srv.Start()
  344. // TODO(xiangli): randomize election timeout
  345. // then remove this sleep.
  346. time.Sleep(1 * time.Millisecond)
  347. ss[i] = srv
  348. }
  349. for i := 1; i <= 10; i++ {
  350. r := pb.Request{
  351. Method: "PUT",
  352. Id: int64(i),
  353. Path: "/foo",
  354. Val: "bar",
  355. }
  356. j := rand.Intn(len(ss))
  357. t.Logf("ss = %d", j)
  358. resp, err := ss[j].Do(ctx, r)
  359. if err != nil {
  360. t.Fatal(err)
  361. }
  362. g, w := resp.Event.Node, &store.NodeExtern{
  363. Key: "/foo",
  364. ModifiedIndex: uint64(i),
  365. CreatedIndex: uint64(i),
  366. Value: stringp("bar"),
  367. }
  368. if !reflect.DeepEqual(g, w) {
  369. t.Error("value:", *g.Value)
  370. t.Errorf("g = %+v, w %+v", g, w)
  371. }
  372. }
  373. time.Sleep(10 * time.Millisecond)
  374. var last interface{}
  375. for i, sv := range ss {
  376. sv.Stop()
  377. g, _ := sv.Store.Get("/", true, true)
  378. if last != nil && !reflect.DeepEqual(last, g) {
  379. t.Errorf("server %d: Root = %#v, want %#v", i, g, last)
  380. }
  381. last = g
  382. }
  383. }
  384. func TestDoProposal(t *testing.T) {
  385. tests := []pb.Request{
  386. pb.Request{Method: "POST", Id: 1},
  387. pb.Request{Method: "PUT", Id: 1},
  388. pb.Request{Method: "DELETE", Id: 1},
  389. pb.Request{Method: "GET", Id: 1, Quorum: true},
  390. }
  391. for i, tt := range tests {
  392. ctx, _ := context.WithCancel(context.Background())
  393. n := raft.StartNode(0xBAD0, []int64{0xBAD0}, 10, 1)
  394. st := &storeRecorder{}
  395. tk := make(chan time.Time)
  396. // this makes <-tk always successful, which accelerates internal clock
  397. close(tk)
  398. srv := &EtcdServer{
  399. Node: n,
  400. Store: st,
  401. Send: func(_ []raftpb.Message) {},
  402. Storage: &storageRecorder{},
  403. Ticker: tk,
  404. }
  405. srv.Start()
  406. resp, err := srv.Do(ctx, tt)
  407. srv.Stop()
  408. action := st.Action()
  409. if len(action) != 1 {
  410. t.Errorf("#%d: len(action) = %d, want 1", i, len(action))
  411. }
  412. if err != nil {
  413. t.Fatalf("#%d: err = %v, want nil", i, err)
  414. }
  415. wresp := Response{Event: &store.Event{}}
  416. if !reflect.DeepEqual(resp, wresp) {
  417. t.Errorf("#%d: resp = %v, want %v", i, resp, wresp)
  418. }
  419. }
  420. }
  421. func TestDoProposalCancelled(t *testing.T) {
  422. ctx, cancel := context.WithCancel(context.Background())
  423. // node cannot make any progress because there are two nodes
  424. n := raft.StartNode(0xBAD0, []int64{0xBAD0, 0xBAD1}, 10, 1)
  425. st := &storeRecorder{}
  426. wait := &waitRecorder{}
  427. srv := &EtcdServer{
  428. // TODO: use fake node for better testability
  429. Node: n,
  430. Store: st,
  431. w: wait,
  432. }
  433. done := make(chan struct{})
  434. var err error
  435. go func() {
  436. _, err = srv.Do(ctx, pb.Request{Method: "PUT", Id: 1})
  437. close(done)
  438. }()
  439. cancel()
  440. <-done
  441. gaction := st.Action()
  442. if len(gaction) != 0 {
  443. t.Errorf("len(action) = %v, want 0", len(gaction))
  444. }
  445. if err != context.Canceled {
  446. t.Fatalf("err = %v, want %v", err, context.Canceled)
  447. }
  448. w := []action{action{name: "Register1"}, action{name: "Trigger1"}}
  449. if !reflect.DeepEqual(wait.action, w) {
  450. t.Errorf("wait.action = %+v, want %+v", wait.action, w)
  451. }
  452. }
  453. func TestDoProposalStopped(t *testing.T) {
  454. ctx, cancel := context.WithCancel(context.Background())
  455. defer cancel()
  456. // node cannot make any progress because there are two nodes
  457. n := raft.StartNode(0xBAD0, []int64{0xBAD0, 0xBAD1}, 10, 1)
  458. st := &storeRecorder{}
  459. tk := make(chan time.Time)
  460. // this makes <-tk always successful, which accelarates internal clock
  461. close(tk)
  462. srv := &EtcdServer{
  463. // TODO: use fake node for better testability
  464. Node: n,
  465. Store: st,
  466. Send: func(_ []raftpb.Message) {},
  467. Storage: &storageRecorder{},
  468. Ticker: tk,
  469. }
  470. srv.Start()
  471. done := make(chan struct{})
  472. var err error
  473. go func() {
  474. _, err = srv.Do(ctx, pb.Request{Method: "PUT", Id: 1})
  475. close(done)
  476. }()
  477. srv.Stop()
  478. <-done
  479. action := st.Action()
  480. if len(action) != 0 {
  481. t.Errorf("len(action) = %v, want 0", len(action))
  482. }
  483. if err != ErrStopped {
  484. t.Errorf("err = %v, want %v", err, ErrStopped)
  485. }
  486. }
  487. // TestSync tests sync 1. is nonblocking 2. sends out SYNC request.
  488. func TestSync(t *testing.T) {
  489. n := &nodeProposeDataRecorder{}
  490. srv := &EtcdServer{
  491. Node: n,
  492. }
  493. start := time.Now()
  494. srv.sync(defaultSyncTimeout)
  495. // check that sync is non-blocking
  496. if d := time.Since(start); d > time.Millisecond {
  497. t.Errorf("CallSyncTime = %v, want < %v", d, time.Millisecond)
  498. }
  499. testutil.ForceGosched()
  500. data := n.data()
  501. if len(data) != 1 {
  502. t.Fatalf("len(proposeData) = %d, want 1", len(data))
  503. }
  504. var r pb.Request
  505. if err := r.Unmarshal(data[0]); err != nil {
  506. t.Fatalf("unmarshal request error: %v", err)
  507. }
  508. if r.Method != "SYNC" {
  509. t.Errorf("method = %s, want SYNC", r.Method)
  510. }
  511. }
  512. // TestSyncTimeout tests the case that sync 1. is non-blocking 2. cancel request
  513. // after timeout
  514. func TestSyncTimeout(t *testing.T) {
  515. n := &nodeProposalBlockerRecorder{}
  516. srv := &EtcdServer{
  517. Node: n,
  518. }
  519. start := time.Now()
  520. srv.sync(0)
  521. // check that sync is non-blocking
  522. if d := time.Since(start); d > time.Millisecond {
  523. t.Errorf("CallSyncTime = %v, want < %v", d, time.Millisecond)
  524. }
  525. // give time for goroutine in sync to cancel
  526. // TODO: use fake clock
  527. testutil.ForceGosched()
  528. w := []action{action{name: "Propose blocked"}}
  529. if g := n.Action(); !reflect.DeepEqual(g, w) {
  530. t.Errorf("action = %v, want %v", g, w)
  531. }
  532. }
  533. // TODO: TestNoSyncWhenNoLeader
  534. // blockingNodeProposer implements the node interface to allow users to
  535. // block until Propose has been called and then verify the Proposed data
  536. type blockingNodeProposer struct {
  537. ch chan []byte
  538. readyNode
  539. }
  540. func (n *blockingNodeProposer) Propose(_ context.Context, data []byte) error {
  541. n.ch <- data
  542. return nil
  543. }
  544. // TestSyncTrigger tests that the server proposes a SYNC request when its sync timer ticks
  545. func TestSyncTrigger(t *testing.T) {
  546. n := &blockingNodeProposer{
  547. ch: make(chan []byte),
  548. readyNode: *newReadyNode(),
  549. }
  550. st := make(chan time.Time, 1)
  551. srv := &EtcdServer{
  552. Node: n,
  553. Store: &storeRecorder{},
  554. Send: func(_ []raftpb.Message) {},
  555. Storage: &storageRecorder{},
  556. SyncTicker: st,
  557. }
  558. srv.Start()
  559. // trigger the server to become a leader and accept sync requests
  560. n.readyc <- raft.Ready{
  561. SoftState: &raft.SoftState{
  562. RaftState: raft.StateLeader,
  563. },
  564. }
  565. // trigger a sync request
  566. st <- time.Time{}
  567. var data []byte
  568. select {
  569. case <-time.After(time.Second):
  570. t.Fatalf("did not receive proposed request as expected!")
  571. case data = <-n.ch:
  572. }
  573. srv.Stop()
  574. var req pb.Request
  575. if err := req.Unmarshal(data); err != nil {
  576. t.Fatalf("error unmarshalling data: %v", err)
  577. }
  578. if req.Method != "SYNC" {
  579. t.Fatalf("unexpected proposed request: %#v", req.Method)
  580. }
  581. }
  582. // snapshot should snapshot the store and cut the persistent
  583. // TODO: node.Compact is called... we need to make the node an interface
  584. func TestSnapshot(t *testing.T) {
  585. n := raft.StartNode(0xBAD0, []int64{0xBAD0}, 10, 1)
  586. defer n.Stop()
  587. st := &storeRecorder{}
  588. p := &storageRecorder{}
  589. s := &EtcdServer{
  590. Store: st,
  591. Storage: p,
  592. Node: n,
  593. }
  594. s.snapshot()
  595. gaction := st.Action()
  596. if len(gaction) != 1 {
  597. t.Fatalf("len(action) = %d, want 1", len(gaction))
  598. }
  599. if !reflect.DeepEqual(gaction[0], action{name: "Save"}) {
  600. t.Errorf("action = %s, want Save", gaction[0])
  601. }
  602. gaction = p.Action()
  603. if len(gaction) != 1 {
  604. t.Fatalf("len(action) = %d, want 1", len(gaction))
  605. }
  606. if !reflect.DeepEqual(gaction[0], action{name: "Cut"}) {
  607. t.Errorf("action = %s, want Cut", gaction[0])
  608. }
  609. }
  610. // Applied > SnapCount should trigger a SaveSnap event
  611. func TestTriggerSnap(t *testing.T) {
  612. ctx := context.Background()
  613. n := raft.StartNode(0xBAD0, []int64{0xBAD0}, 10, 1)
  614. n.Campaign(ctx)
  615. st := &storeRecorder{}
  616. p := &storageRecorder{}
  617. s := &EtcdServer{
  618. Store: st,
  619. Send: func(_ []raftpb.Message) {},
  620. Storage: p,
  621. Node: n,
  622. SnapCount: 10,
  623. }
  624. s.Start()
  625. for i := 0; int64(i) < s.SnapCount; i++ {
  626. s.Do(ctx, pb.Request{Method: "PUT", Id: 1})
  627. }
  628. time.Sleep(time.Millisecond)
  629. s.Stop()
  630. gaction := p.Action()
  631. // each operation is recorded as a Save
  632. // Nop + SnapCount * Puts + Cut + SaveSnap = Save + SnapCount * Save + Cut + SaveSnap
  633. if len(gaction) != 3+int(s.SnapCount) {
  634. t.Fatalf("len(action) = %d, want %d", len(gaction), 3+int(s.SnapCount))
  635. }
  636. if !reflect.DeepEqual(gaction[12], action{name: "SaveSnap"}) {
  637. t.Errorf("action = %s, want SaveSnap", gaction[12])
  638. }
  639. }
  640. // TestRecvSnapshot tests when it receives a snapshot from raft leader,
  641. // it should trigger storage.SaveSnap and also store.Recover.
  642. func TestRecvSnapshot(t *testing.T) {
  643. n := newReadyNode()
  644. st := &storeRecorder{}
  645. p := &storageRecorder{}
  646. s := &EtcdServer{
  647. Store: st,
  648. Send: func(_ []raftpb.Message) {},
  649. Storage: p,
  650. Node: n,
  651. }
  652. s.Start()
  653. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  654. // make goroutines move forward to receive snapshot
  655. testutil.ForceGosched()
  656. s.Stop()
  657. wactions := []action{action{name: "Recovery"}}
  658. if g := st.Action(); !reflect.DeepEqual(g, wactions) {
  659. t.Errorf("store action = %v, want %v", g, wactions)
  660. }
  661. wactions = []action{action{name: "Save"}, action{name: "SaveSnap"}}
  662. if g := p.Action(); !reflect.DeepEqual(g, wactions) {
  663. t.Errorf("storage action = %v, want %v", g, wactions)
  664. }
  665. }
  666. // TestRecvSlowSnapshot tests that slow snapshot will not be applied
  667. // to store.
  668. func TestRecvSlowSnapshot(t *testing.T) {
  669. n := newReadyNode()
  670. st := &storeRecorder{}
  671. s := &EtcdServer{
  672. Store: st,
  673. Send: func(_ []raftpb.Message) {},
  674. Storage: &storageRecorder{},
  675. Node: n,
  676. }
  677. s.Start()
  678. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  679. // make goroutines move forward to receive snapshot
  680. testutil.ForceGosched()
  681. action := st.Action()
  682. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  683. // make goroutines move forward to receive snapshot
  684. testutil.ForceGosched()
  685. s.Stop()
  686. if g := st.Action(); !reflect.DeepEqual(g, action) {
  687. t.Errorf("store action = %v, want %v", g, action)
  688. }
  689. }
  690. // TestAddNode tests AddNode can propose and perform node addition.
  691. func TestAddNode(t *testing.T) {
  692. n := newNodeConfChangeCommitterRecorder()
  693. s := &EtcdServer{
  694. Node: n,
  695. Store: &storeRecorder{},
  696. Send: func(_ []raftpb.Message) {},
  697. Storage: &storageRecorder{},
  698. }
  699. s.Start()
  700. s.AddNode(context.TODO(), 1, []byte("foo"))
  701. gaction := n.Action()
  702. s.Stop()
  703. wactions := []action{action{name: "ProposeConfChange:ConfChangeAddNode"}, action{name: "ApplyConfChange:ConfChangeAddNode"}}
  704. if !reflect.DeepEqual(gaction, wactions) {
  705. t.Errorf("action = %v, want %v", gaction, wactions)
  706. }
  707. }
  708. // TestRemoveNode tests RemoveNode can propose and perform node removal.
  709. func TestRemoveNode(t *testing.T) {
  710. n := newNodeConfChangeCommitterRecorder()
  711. s := &EtcdServer{
  712. Node: n,
  713. Store: &storeRecorder{},
  714. Send: func(_ []raftpb.Message) {},
  715. Storage: &storageRecorder{},
  716. }
  717. s.Start()
  718. s.RemoveNode(context.TODO(), 1)
  719. gaction := n.Action()
  720. s.Stop()
  721. wactions := []action{action{name: "ProposeConfChange:ConfChangeRemoveNode"}, action{name: "ApplyConfChange:ConfChangeRemoveNode"}}
  722. if !reflect.DeepEqual(gaction, wactions) {
  723. t.Errorf("action = %v, want %v", gaction, wactions)
  724. }
  725. }
  726. // TODO: test wait trigger correctness in multi-server case
  727. func TestGetBool(t *testing.T) {
  728. tests := []struct {
  729. b *bool
  730. wb bool
  731. wset bool
  732. }{
  733. {nil, false, false},
  734. {boolp(true), true, true},
  735. {boolp(false), false, true},
  736. }
  737. for i, tt := range tests {
  738. b, set := getBool(tt.b)
  739. if b != tt.wb {
  740. t.Errorf("#%d: value = %v, want %v", i, b, tt.wb)
  741. }
  742. if set != tt.wset {
  743. t.Errorf("#%d: set = %v, want %v", i, set, tt.wset)
  744. }
  745. }
  746. }
  747. func TestGenID(t *testing.T) {
  748. // Sanity check that the GenID function has been seeded appropriately
  749. // (math/rand is seeded with 1 by default)
  750. r := rand.NewSource(int64(1))
  751. var n int64
  752. for n == 0 {
  753. n = r.Int63()
  754. }
  755. if n == GenID() {
  756. t.Fatalf("GenID's rand seeded with 1!")
  757. }
  758. }
  759. type action struct {
  760. name string
  761. params []interface{}
  762. }
  763. type recorder struct {
  764. sync.Mutex
  765. actions []action
  766. }
  767. func (r *recorder) record(a action) {
  768. r.Lock()
  769. r.actions = append(r.actions, a)
  770. r.Unlock()
  771. }
  772. func (r *recorder) Action() []action {
  773. r.Lock()
  774. cpy := make([]action, len(r.actions))
  775. copy(cpy, r.actions)
  776. r.Unlock()
  777. return cpy
  778. }
  779. type storeRecorder struct {
  780. recorder
  781. }
  782. func (s *storeRecorder) Version() int { return 0 }
  783. func (s *storeRecorder) Index() uint64 { return 0 }
  784. func (s *storeRecorder) Get(path string, recursive, sorted bool) (*store.Event, error) {
  785. s.record(action{
  786. name: "Get",
  787. params: []interface{}{path, recursive, sorted},
  788. })
  789. return &store.Event{}, nil
  790. }
  791. func (s *storeRecorder) Set(_ string, _ bool, _ string, _ time.Time) (*store.Event, error) {
  792. s.record(action{name: "Set"})
  793. return &store.Event{}, nil
  794. }
  795. func (s *storeRecorder) Update(path, val string, expr time.Time) (*store.Event, error) {
  796. s.record(action{
  797. name: "Update",
  798. params: []interface{}{path, val, expr},
  799. })
  800. return &store.Event{}, nil
  801. }
  802. func (s *storeRecorder) Create(path string, dir bool, val string, uniq bool, exp time.Time) (*store.Event, error) {
  803. s.record(action{
  804. name: "Create",
  805. params: []interface{}{path, dir, val, uniq, exp},
  806. })
  807. return &store.Event{}, nil
  808. }
  809. func (s *storeRecorder) CompareAndSwap(path, prevVal string, prevIdx uint64, val string, expr time.Time) (*store.Event, error) {
  810. s.record(action{
  811. name: "CompareAndSwap",
  812. params: []interface{}{path, prevVal, prevIdx, val, expr},
  813. })
  814. return &store.Event{}, nil
  815. }
  816. func (s *storeRecorder) Delete(path string, dir, recursive bool) (*store.Event, error) {
  817. s.record(action{
  818. name: "Delete",
  819. params: []interface{}{path, dir, recursive},
  820. })
  821. return &store.Event{}, nil
  822. }
  823. func (s *storeRecorder) CompareAndDelete(path, prevVal string, prevIdx uint64) (*store.Event, error) {
  824. s.record(action{
  825. name: "CompareAndDelete",
  826. params: []interface{}{path, prevVal, prevIdx},
  827. })
  828. return &store.Event{}, nil
  829. }
  830. func (s *storeRecorder) Watch(_ string, _, _ bool, _ uint64) (store.Watcher, error) {
  831. s.record(action{name: "Watch"})
  832. return &stubWatcher{}, nil
  833. }
  834. func (s *storeRecorder) Save() ([]byte, error) {
  835. s.record(action{name: "Save"})
  836. return nil, nil
  837. }
  838. func (s *storeRecorder) Recovery(b []byte) error {
  839. s.record(action{name: "Recovery"})
  840. return nil
  841. }
  842. func (s *storeRecorder) TotalTransactions() uint64 { return 0 }
  843. func (s *storeRecorder) JsonStats() []byte { return nil }
  844. func (s *storeRecorder) DeleteExpiredKeys(cutoff time.Time) {
  845. s.record(action{
  846. name: "DeleteExpiredKeys",
  847. params: []interface{}{cutoff},
  848. })
  849. }
  850. type stubWatcher struct{}
  851. func (w *stubWatcher) EventChan() chan *store.Event { return nil }
  852. func (w *stubWatcher) Remove() {}
  853. // errStoreRecorder returns an store error on Get, Watch request
  854. type errStoreRecorder struct {
  855. storeRecorder
  856. err error
  857. }
  858. func (s *errStoreRecorder) Get(_ string, _, _ bool) (*store.Event, error) {
  859. s.record(action{name: "Get"})
  860. return nil, s.err
  861. }
  862. func (s *errStoreRecorder) Watch(_ string, _, _ bool, _ uint64) (store.Watcher, error) {
  863. s.record(action{name: "Watch"})
  864. return nil, s.err
  865. }
  866. type waitRecorder struct {
  867. action []action
  868. }
  869. func (w *waitRecorder) Register(id int64) <-chan interface{} {
  870. w.action = append(w.action, action{name: fmt.Sprint("Register", id)})
  871. return nil
  872. }
  873. func (w *waitRecorder) Trigger(id int64, x interface{}) {
  874. w.action = append(w.action, action{name: fmt.Sprint("Trigger", id)})
  875. }
  876. func boolp(b bool) *bool { return &b }
  877. func stringp(s string) *string { return &s }
  878. type storageRecorder struct {
  879. recorder
  880. }
  881. func (p *storageRecorder) Save(st raftpb.HardState, ents []raftpb.Entry) {
  882. p.record(action{name: "Save"})
  883. }
  884. func (p *storageRecorder) Cut() error {
  885. p.record(action{name: "Cut"})
  886. return nil
  887. }
  888. func (p *storageRecorder) SaveSnap(st raftpb.Snapshot) {
  889. if raft.IsEmptySnap(st) {
  890. return
  891. }
  892. p.record(action{name: "SaveSnap"})
  893. }
  894. type readyNode struct {
  895. readyc chan raft.Ready
  896. }
  897. func newReadyNode() *readyNode {
  898. readyc := make(chan raft.Ready, 1)
  899. return &readyNode{readyc: readyc}
  900. }
  901. func (n *readyNode) Tick() {}
  902. func (n *readyNode) Campaign(ctx context.Context) error { return nil }
  903. func (n *readyNode) Propose(ctx context.Context, data []byte) error { return nil }
  904. func (n *readyNode) ProposeConfChange(ctx context.Context, conf raftpb.ConfChange) error {
  905. return nil
  906. }
  907. func (n *readyNode) Step(ctx context.Context, msg raftpb.Message) error { return nil }
  908. func (n *readyNode) Ready() <-chan raft.Ready { return n.readyc }
  909. func (n *readyNode) ApplyConfChange(conf raftpb.ConfChange) {}
  910. func (n *readyNode) Stop() {}
  911. func (n *readyNode) Compact(d []byte) {}
  912. type nodeRecorder struct {
  913. recorder
  914. }
  915. func (n *nodeRecorder) Tick() {
  916. n.record(action{name: "Tick"})
  917. }
  918. func (n *nodeRecorder) Campaign(ctx context.Context) error {
  919. n.record(action{name: "Campaign"})
  920. return nil
  921. }
  922. func (n *nodeRecorder) Propose(ctx context.Context, data []byte) error {
  923. n.record(action{name: "Propose"})
  924. return nil
  925. }
  926. func (n *nodeRecorder) ProposeConfChange(ctx context.Context, conf raftpb.ConfChange) error {
  927. n.record(action{name: "ProposeConfChange"})
  928. return nil
  929. }
  930. func (n *nodeRecorder) Step(ctx context.Context, msg raftpb.Message) error {
  931. n.record(action{name: "Step"})
  932. return nil
  933. }
  934. func (n *nodeRecorder) Ready() <-chan raft.Ready { return nil }
  935. func (n *nodeRecorder) ApplyConfChange(conf raftpb.ConfChange) {
  936. n.record(action{name: "ApplyConfChange"})
  937. }
  938. func (n *nodeRecorder) Stop() {
  939. n.record(action{name: "Stop"})
  940. }
  941. func (n *nodeRecorder) Compact(d []byte) {
  942. n.record(action{name: "Compact"})
  943. }
  944. type nodeProposeDataRecorder struct {
  945. nodeRecorder
  946. sync.Mutex
  947. d [][]byte
  948. }
  949. func (n *nodeProposeDataRecorder) data() [][]byte {
  950. n.Lock()
  951. d := n.d
  952. n.Unlock()
  953. return d
  954. }
  955. func (n *nodeProposeDataRecorder) Propose(ctx context.Context, data []byte) error {
  956. n.nodeRecorder.Propose(ctx, data)
  957. n.Lock()
  958. n.d = append(n.d, data)
  959. n.Unlock()
  960. return nil
  961. }
  962. type nodeProposalBlockerRecorder struct {
  963. nodeRecorder
  964. }
  965. func (n *nodeProposalBlockerRecorder) Propose(ctx context.Context, data []byte) error {
  966. <-ctx.Done()
  967. n.record(action{name: "Propose blocked"})
  968. return nil
  969. }
  970. type nodeConfChangeCommitterRecorder struct {
  971. nodeRecorder
  972. readyc chan raft.Ready
  973. }
  974. func newNodeConfChangeCommitterRecorder() *nodeConfChangeCommitterRecorder {
  975. readyc := make(chan raft.Ready, 1)
  976. readyc <- raft.Ready{SoftState: &raft.SoftState{RaftState: raft.StateLeader}}
  977. return &nodeConfChangeCommitterRecorder{readyc: readyc}
  978. }
  979. func (n *nodeConfChangeCommitterRecorder) ProposeConfChange(ctx context.Context, conf raftpb.ConfChange) error {
  980. data, err := conf.Marshal()
  981. if err != nil {
  982. return err
  983. }
  984. n.readyc <- raft.Ready{CommittedEntries: []raftpb.Entry{{Type: raftpb.EntryConfChange, Data: data}}}
  985. n.record(action{name: "ProposeConfChange:" + conf.Type.String()})
  986. return nil
  987. }
  988. func (n *nodeConfChangeCommitterRecorder) Ready() <-chan raft.Ready {
  989. return n.readyc
  990. }
  991. func (n *nodeConfChangeCommitterRecorder) ApplyConfChange(conf raftpb.ConfChange) {
  992. n.record(action{name: "ApplyConfChange:" + conf.Type.String()})
  993. }