server_test.go 28 KB

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