server_test.go 36 KB

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