server_test.go 37 KB

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