server_test.go 44 KB

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