server_test.go 40 KB

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