server_test.go 39 KB

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