server_test.go 44 KB

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