server_test.go 33 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342
  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. if len(gaction) != 2+int(s.snapCount) {
  734. t.Fatalf("len(action) = %d, want %d", len(gaction), 2+int(s.snapCount))
  735. }
  736. if !reflect.DeepEqual(gaction[11], action{name: "SaveSnap"}) {
  737. t.Errorf("action = %s, want SaveSnap", gaction[11])
  738. }
  739. }
  740. // TestRecvSnapshot tests when it receives a snapshot from raft leader,
  741. // it should trigger storage.SaveSnap and also store.Recover.
  742. func TestRecvSnapshot(t *testing.T) {
  743. n := newReadyNode()
  744. st := &storeRecorder{}
  745. p := &storageRecorder{}
  746. s := &EtcdServer{
  747. store: st,
  748. send: func(_ []raftpb.Message) {},
  749. storage: p,
  750. node: n,
  751. }
  752. s.start()
  753. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  754. // make goroutines move forward to receive snapshot
  755. pkg.ForceGosched()
  756. s.Stop()
  757. wactions := []action{action{name: "Recovery"}}
  758. if g := st.Action(); !reflect.DeepEqual(g, wactions) {
  759. t.Errorf("store action = %v, want %v", g, wactions)
  760. }
  761. wactions = []action{action{name: "Save"}, action{name: "SaveSnap"}}
  762. if g := p.Action(); !reflect.DeepEqual(g, wactions) {
  763. t.Errorf("storage action = %v, want %v", g, wactions)
  764. }
  765. }
  766. // TestRecvSlowSnapshot tests that slow snapshot will not be applied
  767. // to store.
  768. func TestRecvSlowSnapshot(t *testing.T) {
  769. n := newReadyNode()
  770. st := &storeRecorder{}
  771. s := &EtcdServer{
  772. store: st,
  773. send: func(_ []raftpb.Message) {},
  774. storage: &storageRecorder{},
  775. node: n,
  776. }
  777. s.start()
  778. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  779. // make goroutines move forward to receive snapshot
  780. pkg.ForceGosched()
  781. action := st.Action()
  782. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Index: 1}}
  783. // make goroutines move forward to receive snapshot
  784. pkg.ForceGosched()
  785. s.Stop()
  786. if g := st.Action(); !reflect.DeepEqual(g, action) {
  787. t.Errorf("store action = %v, want %v", g, action)
  788. }
  789. }
  790. // TestAddMember tests AddMember can propose and perform node addition.
  791. func TestAddMember(t *testing.T) {
  792. n := newNodeConfChangeCommitterRecorder()
  793. n.readyc <- raft.Ready{
  794. SoftState: &raft.SoftState{
  795. RaftState: raft.StateLeader,
  796. Nodes: []uint64{2, 3},
  797. },
  798. }
  799. cs := &clusterStoreRecorder{}
  800. s := &EtcdServer{
  801. node: n,
  802. store: &storeRecorder{},
  803. send: func(_ []raftpb.Message) {},
  804. storage: &storageRecorder{},
  805. ClusterStore: cs,
  806. }
  807. s.start()
  808. m := Member{ID: 1, RaftAttributes: RaftAttributes{PeerURLs: []string{"foo"}}}
  809. err := s.AddMember(context.TODO(), m)
  810. gaction := n.Action()
  811. s.Stop()
  812. if err != nil {
  813. t.Fatalf("AddMember error: %v", err)
  814. }
  815. wactions := []action{action{name: "ProposeConfChange:ConfChangeAddNode"}, action{name: "ApplyConfChange:ConfChangeAddNode"}}
  816. if !reflect.DeepEqual(gaction, wactions) {
  817. t.Errorf("action = %v, want %v", gaction, wactions)
  818. }
  819. wcsactions := []action{{name: "Add", params: []interface{}{m}}}
  820. if g := cs.Action(); !reflect.DeepEqual(g, wcsactions) {
  821. t.Errorf("csaction = %v, want %v", g, wcsactions)
  822. }
  823. }
  824. // TestRemoveMember tests RemoveMember can propose and perform node removal.
  825. func TestRemoveMember(t *testing.T) {
  826. n := newNodeConfChangeCommitterRecorder()
  827. n.readyc <- raft.Ready{
  828. SoftState: &raft.SoftState{
  829. RaftState: raft.StateLeader,
  830. Nodes: []uint64{1, 2, 3},
  831. },
  832. }
  833. cs := &clusterStoreRecorder{}
  834. s := &EtcdServer{
  835. node: n,
  836. store: &storeRecorder{},
  837. send: func(_ []raftpb.Message) {},
  838. storage: &storageRecorder{},
  839. ClusterStore: cs,
  840. }
  841. s.start()
  842. id := uint64(1)
  843. err := s.RemoveMember(context.TODO(), id)
  844. gaction := n.Action()
  845. s.Stop()
  846. if err != nil {
  847. t.Fatalf("RemoveMember error: %v", err)
  848. }
  849. wactions := []action{action{name: "ProposeConfChange:ConfChangeRemoveNode"}, action{name: "ApplyConfChange:ConfChangeRemoveNode"}}
  850. if !reflect.DeepEqual(gaction, wactions) {
  851. t.Errorf("action = %v, want %v", gaction, wactions)
  852. }
  853. wcsactions := []action{{name: "Remove", params: []interface{}{id}}}
  854. if g := cs.Action(); !reflect.DeepEqual(g, wcsactions) {
  855. t.Errorf("csaction = %v, want %v", g, wcsactions)
  856. }
  857. }
  858. // TestServerStopItself tests that if node sends out Ready with ShouldStop,
  859. // server will stop.
  860. func TestServerStopItself(t *testing.T) {
  861. n := newReadyNode()
  862. s := &EtcdServer{
  863. node: n,
  864. store: &storeRecorder{},
  865. send: func(_ []raftpb.Message) {},
  866. storage: &storageRecorder{},
  867. }
  868. s.start()
  869. n.readyc <- raft.Ready{SoftState: &raft.SoftState{ShouldStop: true}}
  870. select {
  871. case <-s.done:
  872. case <-time.After(time.Millisecond):
  873. t.Errorf("did not receive from closed done channel as expected")
  874. }
  875. }
  876. // TODO: test wait trigger correctness in multi-server case
  877. func TestPublish(t *testing.T) {
  878. n := &nodeProposeDataRecorder{}
  879. ch := make(chan interface{}, 1)
  880. // simulate that request has gone through consensus
  881. ch <- Response{}
  882. w := &waitWithResponse{ch: ch}
  883. srv := &EtcdServer{
  884. id: 1,
  885. attributes: Attributes{Name: "node1", ClientURLs: []string{"http://a", "http://b"}},
  886. node: n,
  887. w: w,
  888. }
  889. srv.publish(time.Hour)
  890. data := n.data()
  891. if len(data) != 1 {
  892. t.Fatalf("len(proposeData) = %d, want 1", len(data))
  893. }
  894. var r pb.Request
  895. if err := r.Unmarshal(data[0]); err != nil {
  896. t.Fatalf("unmarshal request error: %v", err)
  897. }
  898. if r.Method != "PUT" {
  899. t.Errorf("method = %s, want PUT", r.Method)
  900. }
  901. wm := Member{ID: 1, Attributes: Attributes{Name: "node1", ClientURLs: []string{"http://a", "http://b"}}}
  902. if r.Path != wm.storeKey()+attributesSuffix {
  903. t.Errorf("path = %s, want %s", r.Path, wm.storeKey()+attributesSuffix)
  904. }
  905. var gattr Attributes
  906. if err := json.Unmarshal([]byte(r.Val), &gattr); err != nil {
  907. t.Fatalf("unmarshal val error: %v", err)
  908. }
  909. if !reflect.DeepEqual(gattr, wm.Attributes) {
  910. t.Errorf("member = %v, want %v", gattr, wm.Attributes)
  911. }
  912. }
  913. // TestPublishStopped tests that publish will be stopped if server is stopped.
  914. func TestPublishStopped(t *testing.T) {
  915. srv := &EtcdServer{
  916. node: &nodeRecorder{},
  917. w: &waitRecorder{},
  918. done: make(chan struct{}),
  919. }
  920. srv.Stop()
  921. srv.publish(time.Hour)
  922. }
  923. // TestPublishRetry tests that publish will keep retry until success.
  924. func TestPublishRetry(t *testing.T) {
  925. n := &nodeRecorder{}
  926. srv := &EtcdServer{
  927. node: n,
  928. w: &waitRecorder{},
  929. done: make(chan struct{}),
  930. }
  931. time.AfterFunc(500*time.Microsecond, srv.Stop)
  932. srv.publish(10 * time.Nanosecond)
  933. action := n.Action()
  934. // multiple Proposes
  935. if len(action) < 2 {
  936. t.Errorf("len(action) = %d, want >= 2", action)
  937. }
  938. }
  939. func TestGetBool(t *testing.T) {
  940. tests := []struct {
  941. b *bool
  942. wb bool
  943. wset bool
  944. }{
  945. {nil, false, false},
  946. {boolp(true), true, true},
  947. {boolp(false), false, true},
  948. }
  949. for i, tt := range tests {
  950. b, set := getBool(tt.b)
  951. if b != tt.wb {
  952. t.Errorf("#%d: value = %v, want %v", i, b, tt.wb)
  953. }
  954. if set != tt.wset {
  955. t.Errorf("#%d: set = %v, want %v", i, set, tt.wset)
  956. }
  957. }
  958. }
  959. func TestGenID(t *testing.T) {
  960. // Sanity check that the GenID function has been seeded appropriately
  961. // (math/rand is seeded with 1 by default)
  962. r := rand.NewSource(int64(1))
  963. var n uint64
  964. for n == 0 {
  965. n = uint64(r.Int63())
  966. }
  967. if n == GenID() {
  968. t.Fatalf("GenID's rand seeded with 1!")
  969. }
  970. }
  971. type action struct {
  972. name string
  973. params []interface{}
  974. }
  975. type recorder struct {
  976. sync.Mutex
  977. actions []action
  978. }
  979. func (r *recorder) record(a action) {
  980. r.Lock()
  981. r.actions = append(r.actions, a)
  982. r.Unlock()
  983. }
  984. func (r *recorder) Action() []action {
  985. r.Lock()
  986. cpy := make([]action, len(r.actions))
  987. copy(cpy, r.actions)
  988. r.Unlock()
  989. return cpy
  990. }
  991. type storeRecorder struct {
  992. recorder
  993. }
  994. func (s *storeRecorder) Version() int { return 0 }
  995. func (s *storeRecorder) Index() uint64 { return 0 }
  996. func (s *storeRecorder) Get(path string, recursive, sorted bool) (*store.Event, error) {
  997. s.record(action{
  998. name: "Get",
  999. params: []interface{}{path, recursive, sorted},
  1000. })
  1001. return &store.Event{}, nil
  1002. }
  1003. func (s *storeRecorder) Set(path string, dir bool, val string, expr time.Time) (*store.Event, error) {
  1004. s.record(action{
  1005. name: "Set",
  1006. params: []interface{}{path, dir, val, expr},
  1007. })
  1008. return &store.Event{}, nil
  1009. }
  1010. func (s *storeRecorder) Update(path, val string, expr time.Time) (*store.Event, error) {
  1011. s.record(action{
  1012. name: "Update",
  1013. params: []interface{}{path, val, expr},
  1014. })
  1015. return &store.Event{}, nil
  1016. }
  1017. func (s *storeRecorder) Create(path string, dir bool, val string, uniq bool, exp time.Time) (*store.Event, error) {
  1018. s.record(action{
  1019. name: "Create",
  1020. params: []interface{}{path, dir, val, uniq, exp},
  1021. })
  1022. return &store.Event{}, nil
  1023. }
  1024. func (s *storeRecorder) CompareAndSwap(path, prevVal string, prevIdx uint64, val string, expr time.Time) (*store.Event, error) {
  1025. s.record(action{
  1026. name: "CompareAndSwap",
  1027. params: []interface{}{path, prevVal, prevIdx, val, expr},
  1028. })
  1029. return &store.Event{}, nil
  1030. }
  1031. func (s *storeRecorder) Delete(path string, dir, recursive bool) (*store.Event, error) {
  1032. s.record(action{
  1033. name: "Delete",
  1034. params: []interface{}{path, dir, recursive},
  1035. })
  1036. return &store.Event{}, nil
  1037. }
  1038. func (s *storeRecorder) CompareAndDelete(path, prevVal string, prevIdx uint64) (*store.Event, error) {
  1039. s.record(action{
  1040. name: "CompareAndDelete",
  1041. params: []interface{}{path, prevVal, prevIdx},
  1042. })
  1043. return &store.Event{}, nil
  1044. }
  1045. func (s *storeRecorder) Watch(_ string, _, _ bool, _ uint64) (store.Watcher, error) {
  1046. s.record(action{name: "Watch"})
  1047. return &stubWatcher{}, nil
  1048. }
  1049. func (s *storeRecorder) Save() ([]byte, error) {
  1050. s.record(action{name: "Save"})
  1051. return nil, nil
  1052. }
  1053. func (s *storeRecorder) Recovery(b []byte) error {
  1054. s.record(action{name: "Recovery"})
  1055. return nil
  1056. }
  1057. func (s *storeRecorder) TotalTransactions() uint64 { return 0 }
  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 nil
  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. }