server_test.go 39 KB

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