rawnode_test.go 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402
  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 raft
  15. import (
  16. "bytes"
  17. "reflect"
  18. "testing"
  19. "github.com/coreos/etcd/raft/raftpb"
  20. )
  21. // TestRawNodeStep ensures that RawNode.Step ignore local message.
  22. func TestRawNodeStep(t *testing.T) {
  23. for i, msgn := range raftpb.MessageType_name {
  24. s := NewMemoryStorage()
  25. rawNode, err := NewRawNode(newTestConfig(1, nil, 10, 1, s), []Peer{{ID: 1}})
  26. if err != nil {
  27. t.Fatal(err)
  28. }
  29. msgt := raftpb.MessageType(i)
  30. err = rawNode.Step(raftpb.Message{Type: msgt})
  31. // LocalMsg should be ignored.
  32. if IsLocalMsg(msgt) {
  33. if err != ErrStepLocalMsg {
  34. t.Errorf("%d: step should ignore %s", msgt, msgn)
  35. }
  36. }
  37. }
  38. }
  39. // TestNodeStepUnblock from node_test.go has no equivalent in rawNode because there is
  40. // no goroutine in RawNode.
  41. // TestRawNodeProposeAndConfChange ensures that RawNode.Propose and RawNode.ProposeConfChange
  42. // send the given proposal and ConfChange to the underlying raft.
  43. func TestRawNodeProposeAndConfChange(t *testing.T) {
  44. s := NewMemoryStorage()
  45. var err error
  46. rawNode, err := NewRawNode(newTestConfig(1, nil, 10, 1, s), []Peer{{ID: 1}})
  47. if err != nil {
  48. t.Fatal(err)
  49. }
  50. rd := rawNode.Ready()
  51. s.Append(rd.Entries)
  52. rawNode.Advance(rd)
  53. rawNode.Campaign()
  54. proposed := false
  55. var (
  56. lastIndex uint64
  57. ccdata []byte
  58. )
  59. for {
  60. rd = rawNode.Ready()
  61. s.Append(rd.Entries)
  62. // Once we are the leader, propose a command and a ConfChange.
  63. if !proposed && rd.SoftState.Lead == rawNode.raft.id {
  64. rawNode.Propose([]byte("somedata"))
  65. cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
  66. ccdata, err = cc.Marshal()
  67. if err != nil {
  68. t.Fatal(err)
  69. }
  70. rawNode.ProposeConfChange(cc)
  71. proposed = true
  72. }
  73. rawNode.Advance(rd)
  74. // Exit when we have four entries: one ConfChange, one no-op for the election,
  75. // our proposed command and proposed ConfChange.
  76. lastIndex, err = s.LastIndex()
  77. if err != nil {
  78. t.Fatal(err)
  79. }
  80. if lastIndex >= 4 {
  81. break
  82. }
  83. }
  84. entries, err := s.Entries(lastIndex-1, lastIndex+1, noLimit)
  85. if err != nil {
  86. t.Fatal(err)
  87. }
  88. if len(entries) != 2 {
  89. t.Fatalf("len(entries) = %d, want %d", len(entries), 2)
  90. }
  91. if !bytes.Equal(entries[0].Data, []byte("somedata")) {
  92. t.Errorf("entries[0].Data = %v, want %v", entries[0].Data, []byte("somedata"))
  93. }
  94. if entries[1].Type != raftpb.EntryConfChange {
  95. t.Fatalf("type = %v, want %v", entries[1].Type, raftpb.EntryConfChange)
  96. }
  97. if !bytes.Equal(entries[1].Data, ccdata) {
  98. t.Errorf("data = %v, want %v", entries[1].Data, ccdata)
  99. }
  100. }
  101. // TestRawNodeProposeAddDuplicateNode ensures that two proposes to add the same node should
  102. // not affect the later propose to add new node.
  103. func TestRawNodeProposeAddDuplicateNode(t *testing.T) {
  104. s := NewMemoryStorage()
  105. rawNode, err := NewRawNode(newTestConfig(1, nil, 10, 1, s), []Peer{{ID: 1}})
  106. if err != nil {
  107. t.Fatal(err)
  108. }
  109. rd := rawNode.Ready()
  110. s.Append(rd.Entries)
  111. rawNode.Advance(rd)
  112. rawNode.Campaign()
  113. for {
  114. rd = rawNode.Ready()
  115. s.Append(rd.Entries)
  116. if rd.SoftState.Lead == rawNode.raft.id {
  117. rawNode.Advance(rd)
  118. break
  119. }
  120. rawNode.Advance(rd)
  121. }
  122. proposeConfChangeAndApply := func(cc raftpb.ConfChange) {
  123. rawNode.ProposeConfChange(cc)
  124. rd = rawNode.Ready()
  125. s.Append(rd.Entries)
  126. for _, entry := range rd.CommittedEntries {
  127. if entry.Type == raftpb.EntryConfChange {
  128. var cc raftpb.ConfChange
  129. cc.Unmarshal(entry.Data)
  130. rawNode.ApplyConfChange(cc)
  131. }
  132. }
  133. rawNode.Advance(rd)
  134. }
  135. cc1 := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
  136. ccdata1, err := cc1.Marshal()
  137. if err != nil {
  138. t.Fatal(err)
  139. }
  140. proposeConfChangeAndApply(cc1)
  141. // try to add the same node again
  142. proposeConfChangeAndApply(cc1)
  143. // the new node join should be ok
  144. cc2 := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 2}
  145. ccdata2, err := cc2.Marshal()
  146. if err != nil {
  147. t.Fatal(err)
  148. }
  149. proposeConfChangeAndApply(cc2)
  150. lastIndex, err := s.LastIndex()
  151. if err != nil {
  152. t.Fatal(err)
  153. }
  154. // the last three entries should be: ConfChange cc1, cc1, cc2
  155. entries, err := s.Entries(lastIndex-2, lastIndex+1, noLimit)
  156. if err != nil {
  157. t.Fatal(err)
  158. }
  159. if len(entries) != 3 {
  160. t.Fatalf("len(entries) = %d, want %d", len(entries), 3)
  161. }
  162. if !bytes.Equal(entries[0].Data, ccdata1) {
  163. t.Errorf("entries[0].Data = %v, want %v", entries[0].Data, ccdata1)
  164. }
  165. if !bytes.Equal(entries[2].Data, ccdata2) {
  166. t.Errorf("entries[2].Data = %v, want %v", entries[2].Data, ccdata2)
  167. }
  168. }
  169. // TestRawNodeReadIndex ensures that Rawnode.ReadIndex sends the MsgReadIndex message
  170. // to the underlying raft. It also ensures that ReadState can be read out.
  171. func TestRawNodeReadIndex(t *testing.T) {
  172. msgs := []raftpb.Message{}
  173. appendStep := func(r *raft, m raftpb.Message) {
  174. msgs = append(msgs, m)
  175. }
  176. wrs := []ReadState{{Index: uint64(1), RequestCtx: []byte("somedata")}}
  177. s := NewMemoryStorage()
  178. c := newTestConfig(1, nil, 10, 1, s)
  179. rawNode, err := NewRawNode(c, []Peer{{ID: 1}})
  180. if err != nil {
  181. t.Fatal(err)
  182. }
  183. rawNode.raft.readStates = wrs
  184. // ensure the ReadStates can be read out
  185. hasReady := rawNode.HasReady()
  186. if !hasReady {
  187. t.Errorf("HasReady() returns %t, want %t", hasReady, true)
  188. }
  189. rd := rawNode.Ready()
  190. if !reflect.DeepEqual(rd.ReadStates, wrs) {
  191. t.Errorf("ReadStates = %d, want %d", rd.ReadStates, wrs)
  192. }
  193. s.Append(rd.Entries)
  194. rawNode.Advance(rd)
  195. // ensure raft.readStates is reset after advance
  196. if rawNode.raft.readStates != nil {
  197. t.Errorf("readStates = %v, want %v", rawNode.raft.readStates, nil)
  198. }
  199. wrequestCtx := []byte("somedata2")
  200. rawNode.Campaign()
  201. for {
  202. rd = rawNode.Ready()
  203. s.Append(rd.Entries)
  204. if rd.SoftState.Lead == rawNode.raft.id {
  205. rawNode.Advance(rd)
  206. // Once we are the leader, issue a ReadIndex request
  207. rawNode.raft.step = appendStep
  208. rawNode.ReadIndex(wrequestCtx)
  209. break
  210. }
  211. rawNode.Advance(rd)
  212. }
  213. // ensure that MsgReadIndex message is sent to the underlying raft
  214. if len(msgs) != 1 {
  215. t.Fatalf("len(msgs) = %d, want %d", len(msgs), 1)
  216. }
  217. if msgs[0].Type != raftpb.MsgReadIndex {
  218. t.Errorf("msg type = %d, want %d", msgs[0].Type, raftpb.MsgReadIndex)
  219. }
  220. if !bytes.Equal(msgs[0].Entries[0].Data, wrequestCtx) {
  221. t.Errorf("data = %v, want %v", msgs[0].Entries[0].Data, wrequestCtx)
  222. }
  223. }
  224. // TestBlockProposal from node_test.go has no equivalent in rawNode because there is
  225. // no leader check in RawNode.
  226. // TestNodeTick from node_test.go has no equivalent in rawNode because
  227. // it reaches into the raft object which is not exposed.
  228. // TestNodeStop from node_test.go has no equivalent in rawNode because there is
  229. // no goroutine in RawNode.
  230. // TestRawNodeStart ensures that a node can be started correctly. The node should
  231. // start with correct configuration change entries, and can accept and commit
  232. // proposals.
  233. func TestRawNodeStart(t *testing.T) {
  234. cc := raftpb.ConfChange{Type: raftpb.ConfChangeAddNode, NodeID: 1}
  235. ccdata, err := cc.Marshal()
  236. if err != nil {
  237. t.Fatalf("unexpected marshal error: %v", err)
  238. }
  239. wants := []Ready{
  240. {
  241. HardState: raftpb.HardState{Term: 1, Commit: 1, Vote: 0},
  242. Entries: []raftpb.Entry{
  243. {Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
  244. },
  245. CommittedEntries: []raftpb.Entry{
  246. {Type: raftpb.EntryConfChange, Term: 1, Index: 1, Data: ccdata},
  247. },
  248. MustSync: true,
  249. },
  250. {
  251. HardState: raftpb.HardState{Term: 2, Commit: 3, Vote: 1},
  252. Entries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
  253. CommittedEntries: []raftpb.Entry{{Term: 2, Index: 3, Data: []byte("foo")}},
  254. MustSync: true,
  255. },
  256. }
  257. storage := NewMemoryStorage()
  258. rawNode, err := NewRawNode(newTestConfig(1, nil, 10, 1, storage), []Peer{{ID: 1}})
  259. if err != nil {
  260. t.Fatal(err)
  261. }
  262. rd := rawNode.Ready()
  263. t.Logf("rd %v", rd)
  264. if !reflect.DeepEqual(rd, wants[0]) {
  265. t.Fatalf("#%d: g = %+v,\n w %+v", 1, rd, wants[0])
  266. } else {
  267. storage.Append(rd.Entries)
  268. rawNode.Advance(rd)
  269. }
  270. storage.Append(rd.Entries)
  271. rawNode.Advance(rd)
  272. rawNode.Campaign()
  273. rd = rawNode.Ready()
  274. storage.Append(rd.Entries)
  275. rawNode.Advance(rd)
  276. rawNode.Propose([]byte("foo"))
  277. if rd = rawNode.Ready(); !reflect.DeepEqual(rd, wants[1]) {
  278. t.Errorf("#%d: g = %+v,\n w %+v", 2, rd, wants[1])
  279. } else {
  280. storage.Append(rd.Entries)
  281. rawNode.Advance(rd)
  282. }
  283. if rawNode.HasReady() {
  284. t.Errorf("unexpected Ready: %+v", rawNode.Ready())
  285. }
  286. }
  287. func TestRawNodeRestart(t *testing.T) {
  288. entries := []raftpb.Entry{
  289. {Term: 1, Index: 1},
  290. {Term: 1, Index: 2, Data: []byte("foo")},
  291. }
  292. st := raftpb.HardState{Term: 1, Commit: 1}
  293. want := Ready{
  294. HardState: emptyState,
  295. // commit up to commit index in st
  296. CommittedEntries: entries[:st.Commit],
  297. MustSync: true,
  298. }
  299. storage := NewMemoryStorage()
  300. storage.SetHardState(st)
  301. storage.Append(entries)
  302. rawNode, err := NewRawNode(newTestConfig(1, nil, 10, 1, storage), nil)
  303. if err != nil {
  304. t.Fatal(err)
  305. }
  306. rd := rawNode.Ready()
  307. if !reflect.DeepEqual(rd, want) {
  308. t.Errorf("g = %+v,\n w %+v", rd, want)
  309. }
  310. rawNode.Advance(rd)
  311. if rawNode.HasReady() {
  312. t.Errorf("unexpected Ready: %+v", rawNode.Ready())
  313. }
  314. }
  315. func TestRawNodeRestartFromSnapshot(t *testing.T) {
  316. snap := raftpb.Snapshot{
  317. Metadata: raftpb.SnapshotMetadata{
  318. ConfState: raftpb.ConfState{Nodes: []uint64{1, 2}},
  319. Index: 2,
  320. Term: 1,
  321. },
  322. }
  323. entries := []raftpb.Entry{
  324. {Term: 1, Index: 3, Data: []byte("foo")},
  325. }
  326. st := raftpb.HardState{Term: 1, Commit: 3}
  327. want := Ready{
  328. HardState: emptyState,
  329. // commit up to commit index in st
  330. CommittedEntries: entries,
  331. MustSync: true,
  332. }
  333. s := NewMemoryStorage()
  334. s.SetHardState(st)
  335. s.ApplySnapshot(snap)
  336. s.Append(entries)
  337. rawNode, err := NewRawNode(newTestConfig(1, nil, 10, 1, s), nil)
  338. if err != nil {
  339. t.Fatal(err)
  340. }
  341. if rd := rawNode.Ready(); !reflect.DeepEqual(rd, want) {
  342. t.Errorf("g = %+v,\n w %+v", rd, want)
  343. } else {
  344. rawNode.Advance(rd)
  345. }
  346. if rawNode.HasReady() {
  347. t.Errorf("unexpected Ready: %+v", rawNode.HasReady())
  348. }
  349. }
  350. // TestNodeAdvance from node_test.go has no equivalent in rawNode because there is
  351. // no dependency check between Ready() and Advance()
  352. func TestRawNodeStatus(t *testing.T) {
  353. storage := NewMemoryStorage()
  354. rawNode, err := NewRawNode(newTestConfig(1, nil, 10, 1, storage), []Peer{{ID: 1}})
  355. if err != nil {
  356. t.Fatal(err)
  357. }
  358. status := rawNode.Status()
  359. if status == nil {
  360. t.Errorf("expected status struct, got nil")
  361. }
  362. }