server_test.go 37 KB

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