server_test.go 40 KB

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