server_test.go 33 KB

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