server_test.go 42 KB

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