server_test.go 42 KB

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