server_test.go 42 KB

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