server_test.go 37 KB

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