server_test.go 37 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422
  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. }
  439. _, err := srv.applyConfChange(tt.cc, nil)
  440. if err != tt.werr {
  441. t.Errorf("#%d: applyConfChange error = %v, want %v", i, err, tt.werr)
  442. }
  443. cc := raftpb.ConfChange{Type: tt.cc.Type, NodeID: raft.None}
  444. w := []testutil.Action{
  445. {
  446. Name: "ApplyConfChange",
  447. Params: []interface{}{cc},
  448. },
  449. }
  450. if g := n.Action(); !reflect.DeepEqual(g, w) {
  451. t.Errorf("#%d: action = %+v, want %+v", i, g, w)
  452. }
  453. }
  454. }
  455. func TestApplyConfChangeShouldStop(t *testing.T) {
  456. cl := newCluster("")
  457. cl.SetStore(store.New())
  458. for i := 1; i <= 3; i++ {
  459. cl.AddMember(&Member{ID: types.ID(i)})
  460. }
  461. srv := &EtcdServer{
  462. id: 1,
  463. r: raftNode{
  464. Node: &nodeRecorder{},
  465. transport: &nopTransporter{},
  466. },
  467. cluster: cl,
  468. }
  469. cc := raftpb.ConfChange{
  470. Type: raftpb.ConfChangeRemoveNode,
  471. NodeID: 2,
  472. }
  473. // remove non-local member
  474. shouldStop, err := srv.applyConfChange(cc, &raftpb.ConfState{})
  475. if err != nil {
  476. t.Fatalf("unexpected error %v", err)
  477. }
  478. if shouldStop != false {
  479. t.Errorf("shouldStop = %t, want %t", shouldStop, false)
  480. }
  481. // remove local member
  482. cc.NodeID = 1
  483. shouldStop, err = srv.applyConfChange(cc, &raftpb.ConfState{})
  484. if err != nil {
  485. t.Fatalf("unexpected error %v", err)
  486. }
  487. if shouldStop != true {
  488. t.Errorf("shouldStop = %t, want %t", shouldStop, true)
  489. }
  490. }
  491. func TestDoProposal(t *testing.T) {
  492. tests := []pb.Request{
  493. pb.Request{Method: "POST", ID: 1},
  494. pb.Request{Method: "PUT", ID: 1},
  495. pb.Request{Method: "DELETE", ID: 1},
  496. pb.Request{Method: "GET", ID: 1, Quorum: true},
  497. }
  498. for i, tt := range tests {
  499. st := &storeRecorder{}
  500. srv := &EtcdServer{
  501. cfg: &ServerConfig{TickMs: 1},
  502. r: raftNode{
  503. Node: newNodeCommitter(),
  504. storage: &storageRecorder{},
  505. raftStorage: raft.NewMemoryStorage(),
  506. transport: &nopTransporter{},
  507. },
  508. store: st,
  509. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  510. }
  511. srv.start()
  512. resp, err := srv.Do(context.Background(), tt)
  513. srv.Stop()
  514. action := st.Action()
  515. if len(action) != 1 {
  516. t.Errorf("#%d: len(action) = %d, want 1", i, len(action))
  517. }
  518. if err != nil {
  519. t.Fatalf("#%d: err = %v, want nil", i, err)
  520. }
  521. wresp := Response{Event: &store.Event{}}
  522. if !reflect.DeepEqual(resp, wresp) {
  523. t.Errorf("#%d: resp = %v, want %v", i, resp, wresp)
  524. }
  525. }
  526. }
  527. func TestDoProposalCancelled(t *testing.T) {
  528. wait := &waitRecorder{}
  529. srv := &EtcdServer{
  530. cfg: &ServerConfig{TickMs: 1},
  531. r: raftNode{Node: &nodeRecorder{}},
  532. w: wait,
  533. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  534. }
  535. ctx, cancel := context.WithCancel(context.Background())
  536. cancel()
  537. _, err := srv.Do(ctx, pb.Request{Method: "PUT"})
  538. if err != ErrCanceled {
  539. t.Fatalf("err = %v, want %v", err, ErrCanceled)
  540. }
  541. w := []testutil.Action{{Name: "Register"}, {Name: "Trigger"}}
  542. if !reflect.DeepEqual(wait.action, w) {
  543. t.Errorf("wait.action = %+v, want %+v", wait.action, w)
  544. }
  545. }
  546. func TestDoProposalTimeout(t *testing.T) {
  547. srv := &EtcdServer{
  548. cfg: &ServerConfig{TickMs: 1},
  549. r: raftNode{Node: &nodeRecorder{}},
  550. w: &waitRecorder{},
  551. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  552. }
  553. ctx, _ := context.WithTimeout(context.Background(), 0)
  554. _, err := srv.Do(ctx, pb.Request{Method: "PUT"})
  555. if err != ErrTimeout {
  556. t.Fatalf("err = %v, want %v", err, ErrTimeout)
  557. }
  558. }
  559. func TestDoProposalStopped(t *testing.T) {
  560. srv := &EtcdServer{
  561. cfg: &ServerConfig{TickMs: 1},
  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.WaitSchedule()
  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.WaitSchedule()
  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. cfg: &ServerConfig{TickMs: 1},
  631. r: raftNode{
  632. Node: n,
  633. raftStorage: raft.NewMemoryStorage(),
  634. transport: &nopTransporter{},
  635. storage: &storageRecorder{},
  636. },
  637. store: &storeRecorder{},
  638. SyncTicker: st,
  639. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  640. }
  641. srv.start()
  642. defer srv.Stop()
  643. // trigger the server to become a leader and accept sync requests
  644. n.readyc <- raft.Ready{
  645. SoftState: &raft.SoftState{
  646. RaftState: raft.StateLeader,
  647. },
  648. }
  649. // trigger a sync request
  650. st <- time.Time{}
  651. testutil.WaitSchedule()
  652. action := n.Action()
  653. if len(action) != 1 {
  654. t.Fatalf("len(action) = %d, want 1", len(action))
  655. }
  656. if action[0].Name != "Propose" {
  657. t.Fatalf("action = %s, want Propose", action[0].Name)
  658. }
  659. data := action[0].Params[0].([]byte)
  660. var req pb.Request
  661. if err := req.Unmarshal(data); err != nil {
  662. t.Fatalf("error unmarshalling data: %v", err)
  663. }
  664. if req.Method != "SYNC" {
  665. t.Fatalf("unexpected proposed request: %#v", req.Method)
  666. }
  667. }
  668. // snapshot should snapshot the store and cut the persistent
  669. func TestSnapshot(t *testing.T) {
  670. s := raft.NewMemoryStorage()
  671. s.Append([]raftpb.Entry{{Index: 1}})
  672. st := &storeRecorder{}
  673. p := &storageRecorder{}
  674. srv := &EtcdServer{
  675. r: raftNode{
  676. Node: &nodeRecorder{},
  677. raftStorage: s,
  678. storage: p,
  679. },
  680. store: st,
  681. }
  682. srv.snapshot(1, raftpb.ConfState{Nodes: []uint64{1}})
  683. testutil.WaitSchedule()
  684. gaction := st.Action()
  685. if len(gaction) != 2 {
  686. t.Fatalf("len(action) = %d, want 1", len(gaction))
  687. }
  688. if !reflect.DeepEqual(gaction[0], testutil.Action{Name: "Clone"}) {
  689. t.Errorf("action = %s, want Clone", gaction[0])
  690. }
  691. if !reflect.DeepEqual(gaction[1], testutil.Action{Name: "SaveNoCopy"}) {
  692. t.Errorf("action = %s, want SaveNoCopy", gaction[1])
  693. }
  694. gaction = p.Action()
  695. if len(gaction) != 1 {
  696. t.Fatalf("len(action) = %d, want 1", len(gaction))
  697. }
  698. if !reflect.DeepEqual(gaction[0], testutil.Action{Name: "SaveSnap"}) {
  699. t.Errorf("action = %s, want SaveSnap", gaction[0])
  700. }
  701. }
  702. // Applied > SnapCount should trigger a SaveSnap event
  703. func TestTriggerSnap(t *testing.T) {
  704. snapc := 10
  705. st := &storeRecorder{}
  706. p := &storageRecorder{}
  707. srv := &EtcdServer{
  708. cfg: &ServerConfig{TickMs: 1},
  709. snapCount: uint64(snapc),
  710. r: raftNode{
  711. Node: newNodeCommitter(),
  712. raftStorage: raft.NewMemoryStorage(),
  713. storage: p,
  714. transport: &nopTransporter{},
  715. },
  716. store: st,
  717. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  718. }
  719. srv.start()
  720. for i := 0; i < snapc+1; i++ {
  721. srv.Do(context.Background(), pb.Request{Method: "PUT"})
  722. }
  723. srv.Stop()
  724. // wait for snapshot goroutine to finish
  725. testutil.WaitSchedule()
  726. gaction := p.Action()
  727. // each operation is recorded as a Save
  728. // (SnapCount+1) * Puts + SaveSnap = (SnapCount+1) * Save + SaveSnap
  729. wcnt := 2 + snapc
  730. if len(gaction) != wcnt {
  731. t.Fatalf("len(action) = %d, want %d", len(gaction), wcnt)
  732. }
  733. if !reflect.DeepEqual(gaction[wcnt-1], testutil.Action{Name: "SaveSnap"}) {
  734. t.Errorf("action = %s, want SaveSnap", gaction[wcnt-1])
  735. }
  736. }
  737. // TestRecvSnapshot tests when it receives a snapshot from raft leader,
  738. // it should trigger storage.SaveSnap and also store.Recover.
  739. func TestRecvSnapshot(t *testing.T) {
  740. n := newReadyNode()
  741. st := &storeRecorder{}
  742. p := &storageRecorder{}
  743. cl := newCluster("abc")
  744. cl.SetStore(store.New())
  745. s := &EtcdServer{
  746. r: raftNode{
  747. Node: n,
  748. transport: &nopTransporter{},
  749. storage: p,
  750. raftStorage: raft.NewMemoryStorage(),
  751. },
  752. store: st,
  753. cluster: cl,
  754. }
  755. s.start()
  756. n.readyc <- raft.Ready{Snapshot: raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 1}}}
  757. // make goroutines move forward to receive snapshot
  758. testutil.WaitSchedule()
  759. s.Stop()
  760. wactions := []testutil.Action{{Name: "Recovery"}}
  761. if g := st.Action(); !reflect.DeepEqual(g, wactions) {
  762. t.Errorf("store action = %v, want %v", g, wactions)
  763. }
  764. wactions = []testutil.Action{{Name: "SaveSnap"}, {Name: "Save"}}
  765. if g := p.Action(); !reflect.DeepEqual(g, wactions) {
  766. t.Errorf("storage action = %v, want %v", g, wactions)
  767. }
  768. }
  769. // TestApplySnapshotAndCommittedEntries tests that server applies snapshot
  770. // first and then committed entries.
  771. func TestApplySnapshotAndCommittedEntries(t *testing.T) {
  772. n := newReadyNode()
  773. st := &storeRecorder{}
  774. cl := newCluster("abc")
  775. cl.SetStore(store.New())
  776. storage := raft.NewMemoryStorage()
  777. s := &EtcdServer{
  778. r: raftNode{
  779. Node: n,
  780. storage: &storageRecorder{},
  781. raftStorage: storage,
  782. transport: &nopTransporter{},
  783. },
  784. store: st,
  785. cluster: cl,
  786. }
  787. s.start()
  788. req := &pb.Request{Method: "QGET"}
  789. n.readyc <- raft.Ready{
  790. Snapshot: raftpb.Snapshot{Metadata: raftpb.SnapshotMetadata{Index: 1}},
  791. CommittedEntries: []raftpb.Entry{
  792. {Index: 2, Data: pbutil.MustMarshal(req)},
  793. },
  794. }
  795. // make goroutines move forward to receive snapshot
  796. testutil.WaitSchedule()
  797. s.Stop()
  798. actions := st.Action()
  799. if len(actions) != 2 {
  800. t.Fatalf("len(action) = %d, want 2", len(actions))
  801. }
  802. if actions[0].Name != "Recovery" {
  803. t.Errorf("actions[0] = %s, want %s", actions[0].Name, "Recovery")
  804. }
  805. if actions[1].Name != "Get" {
  806. t.Errorf("actions[1] = %s, want %s", actions[1].Name, "Get")
  807. }
  808. }
  809. // TestAddMember tests AddMember can propose and perform node addition.
  810. func TestAddMember(t *testing.T) {
  811. n := newNodeConfChangeCommitterRecorder()
  812. n.readyc <- raft.Ready{
  813. SoftState: &raft.SoftState{RaftState: raft.StateLeader},
  814. }
  815. cl := newTestCluster(nil)
  816. st := store.New()
  817. cl.SetStore(st)
  818. s := &EtcdServer{
  819. r: raftNode{
  820. Node: n,
  821. raftStorage: raft.NewMemoryStorage(),
  822. storage: &storageRecorder{},
  823. transport: &nopTransporter{},
  824. },
  825. store: st,
  826. cluster: cl,
  827. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  828. }
  829. s.start()
  830. m := Member{ID: 1234, RaftAttributes: RaftAttributes{PeerURLs: []string{"foo"}}}
  831. err := s.AddMember(context.TODO(), m)
  832. gaction := n.Action()
  833. s.Stop()
  834. if err != nil {
  835. t.Fatalf("AddMember error: %v", err)
  836. }
  837. wactions := []testutil.Action{{Name: "ProposeConfChange:ConfChangeAddNode"}, {Name: "ApplyConfChange:ConfChangeAddNode"}}
  838. if !reflect.DeepEqual(gaction, wactions) {
  839. t.Errorf("action = %v, want %v", gaction, wactions)
  840. }
  841. if cl.Member(1234) == nil {
  842. t.Errorf("member with id 1234 is not added")
  843. }
  844. }
  845. // TestRemoveMember tests RemoveMember can propose and perform node removal.
  846. func TestRemoveMember(t *testing.T) {
  847. n := newNodeConfChangeCommitterRecorder()
  848. n.readyc <- raft.Ready{
  849. SoftState: &raft.SoftState{RaftState: raft.StateLeader},
  850. }
  851. cl := newTestCluster(nil)
  852. st := store.New()
  853. cl.SetStore(store.New())
  854. cl.AddMember(&Member{ID: 1234})
  855. s := &EtcdServer{
  856. r: raftNode{
  857. Node: n,
  858. raftStorage: raft.NewMemoryStorage(),
  859. storage: &storageRecorder{},
  860. transport: &nopTransporter{},
  861. },
  862. store: st,
  863. cluster: cl,
  864. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  865. }
  866. s.start()
  867. err := s.RemoveMember(context.TODO(), 1234)
  868. gaction := n.Action()
  869. s.Stop()
  870. if err != nil {
  871. t.Fatalf("RemoveMember error: %v", err)
  872. }
  873. wactions := []testutil.Action{{Name: "ProposeConfChange:ConfChangeRemoveNode"}, {Name: "ApplyConfChange:ConfChangeRemoveNode"}}
  874. if !reflect.DeepEqual(gaction, wactions) {
  875. t.Errorf("action = %v, want %v", gaction, wactions)
  876. }
  877. if cl.Member(1234) != nil {
  878. t.Errorf("member with id 1234 is not removed")
  879. }
  880. }
  881. // TestUpdateMember tests RemoveMember can propose and perform node update.
  882. func TestUpdateMember(t *testing.T) {
  883. n := newNodeConfChangeCommitterRecorder()
  884. n.readyc <- raft.Ready{
  885. SoftState: &raft.SoftState{RaftState: raft.StateLeader},
  886. }
  887. cl := newTestCluster(nil)
  888. st := store.New()
  889. cl.SetStore(st)
  890. cl.AddMember(&Member{ID: 1234})
  891. s := &EtcdServer{
  892. r: raftNode{
  893. Node: n,
  894. raftStorage: raft.NewMemoryStorage(),
  895. storage: &storageRecorder{},
  896. transport: &nopTransporter{},
  897. },
  898. store: st,
  899. cluster: cl,
  900. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  901. }
  902. s.start()
  903. wm := Member{ID: 1234, RaftAttributes: RaftAttributes{PeerURLs: []string{"http://127.0.0.1:1"}}}
  904. err := s.UpdateMember(context.TODO(), wm)
  905. gaction := n.Action()
  906. s.Stop()
  907. if err != nil {
  908. t.Fatalf("UpdateMember error: %v", err)
  909. }
  910. wactions := []testutil.Action{{Name: "ProposeConfChange:ConfChangeUpdateNode"}, {Name: "ApplyConfChange:ConfChangeUpdateNode"}}
  911. if !reflect.DeepEqual(gaction, wactions) {
  912. t.Errorf("action = %v, want %v", gaction, wactions)
  913. }
  914. if !reflect.DeepEqual(cl.Member(1234), &wm) {
  915. t.Errorf("member = %v, want %v", cl.Member(1234), &wm)
  916. }
  917. }
  918. // TODO: test server could stop itself when being removed
  919. func TestPublish(t *testing.T) {
  920. n := &nodeRecorder{}
  921. ch := make(chan interface{}, 1)
  922. // simulate that request has gone through consensus
  923. ch <- Response{}
  924. w := &waitWithResponse{ch: ch}
  925. srv := &EtcdServer{
  926. cfg: &ServerConfig{TickMs: 1},
  927. id: 1,
  928. r: raftNode{Node: n},
  929. attributes: Attributes{Name: "node1", ClientURLs: []string{"http://a", "http://b"}},
  930. cluster: &cluster{},
  931. w: w,
  932. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  933. }
  934. srv.publish(time.Hour)
  935. action := n.Action()
  936. if len(action) != 1 {
  937. t.Fatalf("len(action) = %d, want 1", len(action))
  938. }
  939. if action[0].Name != "Propose" {
  940. t.Fatalf("action = %s, want Propose", action[0].Name)
  941. }
  942. data := action[0].Params[0].([]byte)
  943. var r pb.Request
  944. if err := r.Unmarshal(data); err != nil {
  945. t.Fatalf("unmarshal request error: %v", err)
  946. }
  947. if r.Method != "PUT" {
  948. t.Errorf("method = %s, want PUT", r.Method)
  949. }
  950. wm := Member{ID: 1, Attributes: Attributes{Name: "node1", ClientURLs: []string{"http://a", "http://b"}}}
  951. if wpath := path.Join(memberStoreKey(wm.ID), attributesSuffix); r.Path != wpath {
  952. t.Errorf("path = %s, want %s", r.Path, wpath)
  953. }
  954. var gattr Attributes
  955. if err := json.Unmarshal([]byte(r.Val), &gattr); err != nil {
  956. t.Fatalf("unmarshal val error: %v", err)
  957. }
  958. if !reflect.DeepEqual(gattr, wm.Attributes) {
  959. t.Errorf("member = %v, want %v", gattr, wm.Attributes)
  960. }
  961. }
  962. // TestPublishStopped tests that publish will be stopped if server is stopped.
  963. func TestPublishStopped(t *testing.T) {
  964. srv := &EtcdServer{
  965. cfg: &ServerConfig{TickMs: 1},
  966. r: raftNode{
  967. Node: &nodeRecorder{},
  968. transport: &nopTransporter{},
  969. },
  970. cluster: &cluster{},
  971. w: &waitRecorder{},
  972. done: make(chan struct{}),
  973. stop: make(chan struct{}),
  974. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  975. }
  976. close(srv.done)
  977. srv.publish(time.Hour)
  978. }
  979. // TestPublishRetry tests that publish will keep retry until success.
  980. func TestPublishRetry(t *testing.T) {
  981. n := &nodeRecorder{}
  982. srv := &EtcdServer{
  983. cfg: &ServerConfig{TickMs: 1},
  984. r: raftNode{Node: n},
  985. w: &waitRecorder{},
  986. done: make(chan struct{}),
  987. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  988. }
  989. time.AfterFunc(500*time.Microsecond, func() { close(srv.done) })
  990. srv.publish(10 * time.Nanosecond)
  991. action := n.Action()
  992. // multiple Proposes
  993. if cnt := len(action); cnt < 2 {
  994. t.Errorf("len(action) = %d, want >= 2", cnt)
  995. }
  996. }
  997. func TestUpdateVersion(t *testing.T) {
  998. n := &nodeRecorder{}
  999. ch := make(chan interface{}, 1)
  1000. // simulate that request has gone through consensus
  1001. ch <- Response{}
  1002. w := &waitWithResponse{ch: ch}
  1003. srv := &EtcdServer{
  1004. id: 1,
  1005. cfg: &ServerConfig{TickMs: 1},
  1006. r: raftNode{Node: n},
  1007. attributes: Attributes{Name: "node1", ClientURLs: []string{"http://node1.com"}},
  1008. cluster: &cluster{},
  1009. w: w,
  1010. reqIDGen: idutil.NewGenerator(0, time.Time{}),
  1011. }
  1012. srv.updateClusterVersion("2.0.0")
  1013. action := n.Action()
  1014. if len(action) != 1 {
  1015. t.Fatalf("len(action) = %d, want 1", len(action))
  1016. }
  1017. if action[0].Name != "Propose" {
  1018. t.Fatalf("action = %s, want Propose", action[0].Name)
  1019. }
  1020. data := action[0].Params[0].([]byte)
  1021. var r pb.Request
  1022. if err := r.Unmarshal(data); err != nil {
  1023. t.Fatalf("unmarshal request error: %v", err)
  1024. }
  1025. if r.Method != "PUT" {
  1026. t.Errorf("method = %s, want PUT", r.Method)
  1027. }
  1028. if wpath := path.Join(StoreClusterPrefix, "version"); r.Path != wpath {
  1029. t.Errorf("path = %s, want %s", r.Path, wpath)
  1030. }
  1031. if r.Val != "2.0.0" {
  1032. t.Errorf("val = %s, want %s", r.Val, "2.0.0")
  1033. }
  1034. }
  1035. func TestStopNotify(t *testing.T) {
  1036. s := &EtcdServer{
  1037. stop: make(chan struct{}),
  1038. done: make(chan struct{}),
  1039. }
  1040. go func() {
  1041. <-s.stop
  1042. close(s.done)
  1043. }()
  1044. notifier := s.StopNotify()
  1045. select {
  1046. case <-notifier:
  1047. t.Fatalf("received unexpected stop notification")
  1048. default:
  1049. }
  1050. s.Stop()
  1051. select {
  1052. case <-notifier:
  1053. default:
  1054. t.Fatalf("cannot receive stop notification")
  1055. }
  1056. }
  1057. func TestGetOtherPeerURLs(t *testing.T) {
  1058. tests := []struct {
  1059. membs []*Member
  1060. self string
  1061. wurls []string
  1062. }{
  1063. {
  1064. []*Member{
  1065. newTestMember(1, []string{"http://10.0.0.1"}, "a", nil),
  1066. },
  1067. "a",
  1068. []string{},
  1069. },
  1070. {
  1071. []*Member{
  1072. newTestMember(1, []string{"http://10.0.0.1"}, "a", nil),
  1073. newTestMember(2, []string{"http://10.0.0.2"}, "b", nil),
  1074. newTestMember(3, []string{"http://10.0.0.3"}, "c", nil),
  1075. },
  1076. "a",
  1077. []string{"http://10.0.0.2", "http://10.0.0.3"},
  1078. },
  1079. {
  1080. []*Member{
  1081. newTestMember(1, []string{"http://10.0.0.1"}, "a", nil),
  1082. newTestMember(3, []string{"http://10.0.0.3"}, "c", nil),
  1083. newTestMember(2, []string{"http://10.0.0.2"}, "b", nil),
  1084. },
  1085. "a",
  1086. []string{"http://10.0.0.2", "http://10.0.0.3"},
  1087. },
  1088. }
  1089. for i, tt := range tests {
  1090. cl := newClusterFromMembers("", types.ID(0), tt.membs)
  1091. urls := getRemotePeerURLs(cl, tt.self)
  1092. if !reflect.DeepEqual(urls, tt.wurls) {
  1093. t.Errorf("#%d: urls = %+v, want %+v", i, urls, tt.wurls)
  1094. }
  1095. }
  1096. }
  1097. // storeRecorder records all the methods it receives.
  1098. // storeRecorder DOES NOT work as a actual store.
  1099. // It always returns invalid empty response and no error.
  1100. type storeRecorder struct{ testutil.Recorder }
  1101. func (s *storeRecorder) Version() int { return 0 }
  1102. func (s *storeRecorder) Index() uint64 { return 0 }
  1103. func (s *storeRecorder) Get(path string, recursive, sorted bool) (*store.Event, error) {
  1104. s.Record(testutil.Action{
  1105. Name: "Get",
  1106. Params: []interface{}{path, recursive, sorted},
  1107. })
  1108. return &store.Event{}, nil
  1109. }
  1110. func (s *storeRecorder) Set(path string, dir bool, val string, expr time.Time) (*store.Event, error) {
  1111. s.Record(testutil.Action{
  1112. Name: "Set",
  1113. Params: []interface{}{path, dir, val, expr},
  1114. })
  1115. return &store.Event{}, nil
  1116. }
  1117. func (s *storeRecorder) Update(path, val string, expr time.Time) (*store.Event, error) {
  1118. s.Record(testutil.Action{
  1119. Name: "Update",
  1120. Params: []interface{}{path, val, expr},
  1121. })
  1122. return &store.Event{}, nil
  1123. }
  1124. func (s *storeRecorder) Create(path string, dir bool, val string, uniq bool, exp time.Time) (*store.Event, error) {
  1125. s.Record(testutil.Action{
  1126. Name: "Create",
  1127. Params: []interface{}{path, dir, val, uniq, exp},
  1128. })
  1129. return &store.Event{}, nil
  1130. }
  1131. func (s *storeRecorder) CompareAndSwap(path, prevVal string, prevIdx uint64, val string, expr time.Time) (*store.Event, error) {
  1132. s.Record(testutil.Action{
  1133. Name: "CompareAndSwap",
  1134. Params: []interface{}{path, prevVal, prevIdx, val, expr},
  1135. })
  1136. return &store.Event{}, nil
  1137. }
  1138. func (s *storeRecorder) Delete(path string, dir, recursive bool) (*store.Event, error) {
  1139. s.Record(testutil.Action{
  1140. Name: "Delete",
  1141. Params: []interface{}{path, dir, recursive},
  1142. })
  1143. return &store.Event{}, nil
  1144. }
  1145. func (s *storeRecorder) CompareAndDelete(path, prevVal string, prevIdx uint64) (*store.Event, error) {
  1146. s.Record(testutil.Action{
  1147. Name: "CompareAndDelete",
  1148. Params: []interface{}{path, prevVal, prevIdx},
  1149. })
  1150. return &store.Event{}, nil
  1151. }
  1152. func (s *storeRecorder) Watch(_ string, _, _ bool, _ uint64) (store.Watcher, error) {
  1153. s.Record(testutil.Action{Name: "Watch"})
  1154. return &nopWatcher{}, nil
  1155. }
  1156. func (s *storeRecorder) Save() ([]byte, error) {
  1157. s.Record(testutil.Action{Name: "Save"})
  1158. return nil, nil
  1159. }
  1160. func (s *storeRecorder) Recovery(b []byte) error {
  1161. s.Record(testutil.Action{Name: "Recovery"})
  1162. return nil
  1163. }
  1164. func (s *storeRecorder) SaveNoCopy() ([]byte, error) {
  1165. s.Record(testutil.Action{Name: "SaveNoCopy"})
  1166. return nil, nil
  1167. }
  1168. func (s *storeRecorder) Clone() store.Store {
  1169. s.Record(testutil.Action{Name: "Clone"})
  1170. return s
  1171. }
  1172. func (s *storeRecorder) JsonStats() []byte { return nil }
  1173. func (s *storeRecorder) DeleteExpiredKeys(cutoff time.Time) {
  1174. s.Record(testutil.Action{
  1175. Name: "DeleteExpiredKeys",
  1176. Params: []interface{}{cutoff},
  1177. })
  1178. }
  1179. type nopWatcher struct{}
  1180. func (w *nopWatcher) EventChan() chan *store.Event { return nil }
  1181. func (w *nopWatcher) StartIndex() uint64 { return 0 }
  1182. func (w *nopWatcher) Remove() {}
  1183. // errStoreRecorder is a storeRecorder, but returns the given error on
  1184. // Get, Watch methods.
  1185. type errStoreRecorder struct {
  1186. storeRecorder
  1187. err error
  1188. }
  1189. func (s *errStoreRecorder) Get(path string, recursive, sorted bool) (*store.Event, error) {
  1190. s.storeRecorder.Get(path, recursive, sorted)
  1191. return nil, s.err
  1192. }
  1193. func (s *errStoreRecorder) Watch(path string, recursive, sorted bool, index uint64) (store.Watcher, error) {
  1194. s.storeRecorder.Watch(path, recursive, sorted, index)
  1195. return nil, s.err
  1196. }
  1197. type waitRecorder struct {
  1198. action []testutil.Action
  1199. }
  1200. func (w *waitRecorder) Register(id uint64) <-chan interface{} {
  1201. w.action = append(w.action, testutil.Action{Name: "Register"})
  1202. return nil
  1203. }
  1204. func (w *waitRecorder) Trigger(id uint64, x interface{}) {
  1205. w.action = append(w.action, testutil.Action{Name: "Trigger"})
  1206. }
  1207. type waitWithResponse struct {
  1208. ch <-chan interface{}
  1209. }
  1210. func (w *waitWithResponse) Register(id uint64) <-chan interface{} {
  1211. return w.ch
  1212. }
  1213. func (w *waitWithResponse) Trigger(id uint64, x interface{}) {}
  1214. type storageRecorder struct{ testutil.Recorder }
  1215. func (p *storageRecorder) Save(st raftpb.HardState, ents []raftpb.Entry) error {
  1216. p.Record(testutil.Action{Name: "Save"})
  1217. return nil
  1218. }
  1219. func (p *storageRecorder) SaveSnap(st raftpb.Snapshot) error {
  1220. if !raft.IsEmptySnap(st) {
  1221. p.Record(testutil.Action{Name: "SaveSnap"})
  1222. }
  1223. return nil
  1224. }
  1225. func (p *storageRecorder) Close() error { return nil }
  1226. type nodeRecorder struct{ testutil.Recorder }
  1227. func (n *nodeRecorder) Tick() { n.Record(testutil.Action{Name: "Tick"}) }
  1228. func (n *nodeRecorder) Campaign(ctx context.Context) error {
  1229. n.Record(testutil.Action{Name: "Campaign"})
  1230. return nil
  1231. }
  1232. func (n *nodeRecorder) Propose(ctx context.Context, data []byte) error {
  1233. n.Record(testutil.Action{Name: "Propose", Params: []interface{}{data}})
  1234. return nil
  1235. }
  1236. func (n *nodeRecorder) ProposeConfChange(ctx context.Context, conf raftpb.ConfChange) error {
  1237. n.Record(testutil.Action{Name: "ProposeConfChange"})
  1238. return nil
  1239. }
  1240. func (n *nodeRecorder) Step(ctx context.Context, msg raftpb.Message) error {
  1241. n.Record(testutil.Action{Name: "Step"})
  1242. return nil
  1243. }
  1244. func (n *nodeRecorder) Status() raft.Status { return raft.Status{} }
  1245. func (n *nodeRecorder) Ready() <-chan raft.Ready { return nil }
  1246. func (n *nodeRecorder) Advance() {}
  1247. func (n *nodeRecorder) ApplyConfChange(conf raftpb.ConfChange) *raftpb.ConfState {
  1248. n.Record(testutil.Action{Name: "ApplyConfChange", Params: []interface{}{conf}})
  1249. return &raftpb.ConfState{}
  1250. }
  1251. func (n *nodeRecorder) Stop() {
  1252. n.Record(testutil.Action{Name: "Stop"})
  1253. }
  1254. func (n *nodeRecorder) ReportUnreachable(id uint64) {}
  1255. func (n *nodeRecorder) ReportSnapshot(id uint64, status raft.SnapshotStatus) {}
  1256. func (n *nodeRecorder) Compact(index uint64, nodes []uint64, d []byte) {
  1257. n.Record(testutil.Action{Name: "Compact"})
  1258. }
  1259. type nodeProposalBlockerRecorder struct {
  1260. nodeRecorder
  1261. }
  1262. func (n *nodeProposalBlockerRecorder) Propose(ctx context.Context, data []byte) error {
  1263. <-ctx.Done()
  1264. n.Record(testutil.Action{Name: "Propose blocked"})
  1265. return nil
  1266. }
  1267. type nodeConfChangeCommitterRecorder struct {
  1268. nodeRecorder
  1269. readyc chan raft.Ready
  1270. index uint64
  1271. }
  1272. func newNodeConfChangeCommitterRecorder() *nodeConfChangeCommitterRecorder {
  1273. readyc := make(chan raft.Ready, 1)
  1274. return &nodeConfChangeCommitterRecorder{readyc: readyc}
  1275. }
  1276. func (n *nodeConfChangeCommitterRecorder) ProposeConfChange(ctx context.Context, conf raftpb.ConfChange) error {
  1277. data, err := conf.Marshal()
  1278. if err != nil {
  1279. return err
  1280. }
  1281. n.index++
  1282. n.Record(testutil.Action{Name: "ProposeConfChange:" + conf.Type.String()})
  1283. n.readyc <- raft.Ready{CommittedEntries: []raftpb.Entry{{Index: n.index, Type: raftpb.EntryConfChange, Data: data}}}
  1284. return nil
  1285. }
  1286. func (n *nodeConfChangeCommitterRecorder) Ready() <-chan raft.Ready {
  1287. return n.readyc
  1288. }
  1289. func (n *nodeConfChangeCommitterRecorder) ApplyConfChange(conf raftpb.ConfChange) *raftpb.ConfState {
  1290. n.Record(testutil.Action{Name: "ApplyConfChange:" + conf.Type.String()})
  1291. return &raftpb.ConfState{}
  1292. }
  1293. // nodeCommitter commits proposed data immediately.
  1294. type nodeCommitter struct {
  1295. nodeRecorder
  1296. readyc chan raft.Ready
  1297. index uint64
  1298. }
  1299. func newNodeCommitter() *nodeCommitter {
  1300. readyc := make(chan raft.Ready, 1)
  1301. return &nodeCommitter{readyc: readyc}
  1302. }
  1303. func (n *nodeCommitter) Propose(ctx context.Context, data []byte) error {
  1304. n.index++
  1305. ents := []raftpb.Entry{{Index: n.index, Data: data}}
  1306. n.readyc <- raft.Ready{
  1307. Entries: ents,
  1308. CommittedEntries: ents,
  1309. }
  1310. return nil
  1311. }
  1312. func (n *nodeCommitter) Ready() <-chan raft.Ready {
  1313. return n.readyc
  1314. }
  1315. type readyNode struct {
  1316. nodeRecorder
  1317. readyc chan raft.Ready
  1318. }
  1319. func newReadyNode() *readyNode {
  1320. readyc := make(chan raft.Ready, 1)
  1321. return &readyNode{readyc: readyc}
  1322. }
  1323. func (n *readyNode) Ready() <-chan raft.Ready { return n.readyc }
  1324. type nopTransporter struct{}
  1325. func (s *nopTransporter) Handler() http.Handler { return nil }
  1326. func (s *nopTransporter) Send(m []raftpb.Message) {}
  1327. func (s *nopTransporter) AddRemote(id types.ID, us []string) {}
  1328. func (s *nopTransporter) AddPeer(id types.ID, us []string) {}
  1329. func (s *nopTransporter) RemovePeer(id types.ID) {}
  1330. func (s *nopTransporter) RemoveAllPeers() {}
  1331. func (s *nopTransporter) UpdatePeer(id types.ID, us []string) {}
  1332. func (s *nopTransporter) Stop() {}
  1333. func (s *nopTransporter) Pause() {}
  1334. func (s *nopTransporter) Resume() {}