server_test.go 42 KB

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