server_test.go 37 KB

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