server_test.go 39 KB

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