server_test.go 43 KB

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