server_test.go 38 KB

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