server_test.go 44 KB

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