server_test.go 35 KB

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