William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 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 | |
| 15 | package raft |
| 16 | |
| 17 | import ( |
| 18 | "bytes" |
| 19 | "errors" |
| 20 | "fmt" |
| 21 | "math" |
| 22 | "math/rand" |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 23 | "strings" |
| 24 | "sync" |
| 25 | "time" |
| 26 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 27 | "go.etcd.io/etcd/raft/quorum" |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 28 | pb "go.etcd.io/etcd/raft/raftpb" |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 29 | "go.etcd.io/etcd/raft/tracker" |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 30 | ) |
| 31 | |
| 32 | // None is a placeholder node ID used when there is no leader. |
| 33 | const None uint64 = 0 |
| 34 | const noLimit = math.MaxUint64 |
| 35 | |
| 36 | // Possible values for StateType. |
| 37 | const ( |
| 38 | StateFollower StateType = iota |
| 39 | StateCandidate |
| 40 | StateLeader |
| 41 | StatePreCandidate |
| 42 | numStates |
| 43 | ) |
| 44 | |
| 45 | type ReadOnlyOption int |
| 46 | |
| 47 | const ( |
| 48 | // ReadOnlySafe guarantees the linearizability of the read only request by |
| 49 | // communicating with the quorum. It is the default and suggested option. |
| 50 | ReadOnlySafe ReadOnlyOption = iota |
| 51 | // ReadOnlyLeaseBased ensures linearizability of the read only request by |
| 52 | // relying on the leader lease. It can be affected by clock drift. |
| 53 | // If the clock drift is unbounded, leader might keep the lease longer than it |
| 54 | // should (clock can move backward/pause without any bound). ReadIndex is not safe |
| 55 | // in that case. |
| 56 | ReadOnlyLeaseBased |
| 57 | ) |
| 58 | |
| 59 | // Possible values for CampaignType |
| 60 | const ( |
| 61 | // campaignPreElection represents the first phase of a normal election when |
| 62 | // Config.PreVote is true. |
| 63 | campaignPreElection CampaignType = "CampaignPreElection" |
| 64 | // campaignElection represents a normal (time-based) election (the second phase |
| 65 | // of the election when Config.PreVote is true). |
| 66 | campaignElection CampaignType = "CampaignElection" |
| 67 | // campaignTransfer represents the type of leader transfer |
| 68 | campaignTransfer CampaignType = "CampaignTransfer" |
| 69 | ) |
| 70 | |
| 71 | // ErrProposalDropped is returned when the proposal is ignored by some cases, |
| 72 | // so that the proposer can be notified and fail fast. |
| 73 | var ErrProposalDropped = errors.New("raft proposal dropped") |
| 74 | |
| 75 | // lockedRand is a small wrapper around rand.Rand to provide |
| 76 | // synchronization among multiple raft groups. Only the methods needed |
| 77 | // by the code are exposed (e.g. Intn). |
| 78 | type lockedRand struct { |
| 79 | mu sync.Mutex |
| 80 | rand *rand.Rand |
| 81 | } |
| 82 | |
| 83 | func (r *lockedRand) Intn(n int) int { |
| 84 | r.mu.Lock() |
| 85 | v := r.rand.Intn(n) |
| 86 | r.mu.Unlock() |
| 87 | return v |
| 88 | } |
| 89 | |
| 90 | var globalRand = &lockedRand{ |
| 91 | rand: rand.New(rand.NewSource(time.Now().UnixNano())), |
| 92 | } |
| 93 | |
| 94 | // CampaignType represents the type of campaigning |
| 95 | // the reason we use the type of string instead of uint64 |
| 96 | // is because it's simpler to compare and fill in raft entries |
| 97 | type CampaignType string |
| 98 | |
| 99 | // StateType represents the role of a node in a cluster. |
| 100 | type StateType uint64 |
| 101 | |
| 102 | var stmap = [...]string{ |
| 103 | "StateFollower", |
| 104 | "StateCandidate", |
| 105 | "StateLeader", |
| 106 | "StatePreCandidate", |
| 107 | } |
| 108 | |
| 109 | func (st StateType) String() string { |
| 110 | return stmap[uint64(st)] |
| 111 | } |
| 112 | |
| 113 | // Config contains the parameters to start a raft. |
| 114 | type Config struct { |
| 115 | // ID is the identity of the local raft. ID cannot be 0. |
| 116 | ID uint64 |
| 117 | |
| 118 | // peers contains the IDs of all nodes (including self) in the raft cluster. It |
| 119 | // should only be set when starting a new raft cluster. Restarting raft from |
| 120 | // previous configuration will panic if peers is set. peer is private and only |
| 121 | // used for testing right now. |
| 122 | peers []uint64 |
| 123 | |
| 124 | // learners contains the IDs of all learner nodes (including self if the |
| 125 | // local node is a learner) in the raft cluster. learners only receives |
| 126 | // entries from the leader node. It does not vote or promote itself. |
| 127 | learners []uint64 |
| 128 | |
| 129 | // ElectionTick is the number of Node.Tick invocations that must pass between |
| 130 | // elections. That is, if a follower does not receive any message from the |
| 131 | // leader of current term before ElectionTick has elapsed, it will become |
| 132 | // candidate and start an election. ElectionTick must be greater than |
| 133 | // HeartbeatTick. We suggest ElectionTick = 10 * HeartbeatTick to avoid |
| 134 | // unnecessary leader switching. |
| 135 | ElectionTick int |
| 136 | // HeartbeatTick is the number of Node.Tick invocations that must pass between |
| 137 | // heartbeats. That is, a leader sends heartbeat messages to maintain its |
| 138 | // leadership every HeartbeatTick ticks. |
| 139 | HeartbeatTick int |
| 140 | |
| 141 | // Storage is the storage for raft. raft generates entries and states to be |
| 142 | // stored in storage. raft reads the persisted entries and states out of |
| 143 | // Storage when it needs. raft reads out the previous state and configuration |
| 144 | // out of storage when restarting. |
| 145 | Storage Storage |
| 146 | // Applied is the last applied index. It should only be set when restarting |
| 147 | // raft. raft will not return entries to the application smaller or equal to |
| 148 | // Applied. If Applied is unset when restarting, raft might return previous |
| 149 | // applied entries. This is a very application dependent configuration. |
| 150 | Applied uint64 |
| 151 | |
| 152 | // MaxSizePerMsg limits the max byte size of each append message. Smaller |
| 153 | // value lowers the raft recovery cost(initial probing and message lost |
| 154 | // during normal operation). On the other side, it might affect the |
| 155 | // throughput during normal replication. Note: math.MaxUint64 for unlimited, |
| 156 | // 0 for at most one entry per message. |
| 157 | MaxSizePerMsg uint64 |
| 158 | // MaxCommittedSizePerReady limits the size of the committed entries which |
| 159 | // can be applied. |
| 160 | MaxCommittedSizePerReady uint64 |
| 161 | // MaxUncommittedEntriesSize limits the aggregate byte size of the |
| 162 | // uncommitted entries that may be appended to a leader's log. Once this |
| 163 | // limit is exceeded, proposals will begin to return ErrProposalDropped |
| 164 | // errors. Note: 0 for no limit. |
| 165 | MaxUncommittedEntriesSize uint64 |
| 166 | // MaxInflightMsgs limits the max number of in-flight append messages during |
| 167 | // optimistic replication phase. The application transportation layer usually |
| 168 | // has its own sending buffer over TCP/UDP. Setting MaxInflightMsgs to avoid |
| 169 | // overflowing that sending buffer. TODO (xiangli): feedback to application to |
| 170 | // limit the proposal rate? |
| 171 | MaxInflightMsgs int |
| 172 | |
| 173 | // CheckQuorum specifies if the leader should check quorum activity. Leader |
| 174 | // steps down when quorum is not active for an electionTimeout. |
| 175 | CheckQuorum bool |
| 176 | |
| 177 | // PreVote enables the Pre-Vote algorithm described in raft thesis section |
| 178 | // 9.6. This prevents disruption when a node that has been partitioned away |
| 179 | // rejoins the cluster. |
| 180 | PreVote bool |
| 181 | |
| 182 | // ReadOnlyOption specifies how the read only request is processed. |
| 183 | // |
| 184 | // ReadOnlySafe guarantees the linearizability of the read only request by |
| 185 | // communicating with the quorum. It is the default and suggested option. |
| 186 | // |
| 187 | // ReadOnlyLeaseBased ensures linearizability of the read only request by |
| 188 | // relying on the leader lease. It can be affected by clock drift. |
| 189 | // If the clock drift is unbounded, leader might keep the lease longer than it |
| 190 | // should (clock can move backward/pause without any bound). ReadIndex is not safe |
| 191 | // in that case. |
| 192 | // CheckQuorum MUST be enabled if ReadOnlyOption is ReadOnlyLeaseBased. |
| 193 | ReadOnlyOption ReadOnlyOption |
| 194 | |
| 195 | // Logger is the logger used for raft log. For multinode which can host |
| 196 | // multiple raft group, each raft group can have its own logger |
| 197 | Logger Logger |
| 198 | |
| 199 | // DisableProposalForwarding set to true means that followers will drop |
| 200 | // proposals, rather than forwarding them to the leader. One use case for |
| 201 | // this feature would be in a situation where the Raft leader is used to |
| 202 | // compute the data of a proposal, for example, adding a timestamp from a |
| 203 | // hybrid logical clock to data in a monotonically increasing way. Forwarding |
| 204 | // should be disabled to prevent a follower with an inaccurate hybrid |
| 205 | // logical clock from assigning the timestamp and then forwarding the data |
| 206 | // to the leader. |
| 207 | DisableProposalForwarding bool |
| 208 | } |
| 209 | |
| 210 | func (c *Config) validate() error { |
| 211 | if c.ID == None { |
| 212 | return errors.New("cannot use none as id") |
| 213 | } |
| 214 | |
| 215 | if c.HeartbeatTick <= 0 { |
| 216 | return errors.New("heartbeat tick must be greater than 0") |
| 217 | } |
| 218 | |
| 219 | if c.ElectionTick <= c.HeartbeatTick { |
| 220 | return errors.New("election tick must be greater than heartbeat tick") |
| 221 | } |
| 222 | |
| 223 | if c.Storage == nil { |
| 224 | return errors.New("storage cannot be nil") |
| 225 | } |
| 226 | |
| 227 | if c.MaxUncommittedEntriesSize == 0 { |
| 228 | c.MaxUncommittedEntriesSize = noLimit |
| 229 | } |
| 230 | |
| 231 | // default MaxCommittedSizePerReady to MaxSizePerMsg because they were |
| 232 | // previously the same parameter. |
| 233 | if c.MaxCommittedSizePerReady == 0 { |
| 234 | c.MaxCommittedSizePerReady = c.MaxSizePerMsg |
| 235 | } |
| 236 | |
| 237 | if c.MaxInflightMsgs <= 0 { |
| 238 | return errors.New("max inflight messages must be greater than 0") |
| 239 | } |
| 240 | |
| 241 | if c.Logger == nil { |
| 242 | c.Logger = raftLogger |
| 243 | } |
| 244 | |
| 245 | if c.ReadOnlyOption == ReadOnlyLeaseBased && !c.CheckQuorum { |
| 246 | return errors.New("CheckQuorum must be enabled when ReadOnlyOption is ReadOnlyLeaseBased") |
| 247 | } |
| 248 | |
| 249 | return nil |
| 250 | } |
| 251 | |
| 252 | type raft struct { |
| 253 | id uint64 |
| 254 | |
| 255 | Term uint64 |
| 256 | Vote uint64 |
| 257 | |
| 258 | readStates []ReadState |
| 259 | |
| 260 | // the log |
| 261 | raftLog *raftLog |
| 262 | |
| 263 | maxMsgSize uint64 |
| 264 | maxUncommittedSize uint64 |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 265 | prs tracker.ProgressTracker |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 266 | |
| 267 | state StateType |
| 268 | |
| 269 | // isLearner is true if the local raft node is a learner. |
| 270 | isLearner bool |
| 271 | |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 272 | msgs []pb.Message |
| 273 | |
| 274 | // the leader id |
| 275 | lead uint64 |
| 276 | // leadTransferee is id of the leader transfer target when its value is not zero. |
| 277 | // Follow the procedure defined in raft thesis 3.10. |
| 278 | leadTransferee uint64 |
| 279 | // Only one conf change may be pending (in the log, but not yet |
| 280 | // applied) at a time. This is enforced via pendingConfIndex, which |
| 281 | // is set to a value >= the log index of the latest pending |
| 282 | // configuration change (if any). Config changes are only allowed to |
| 283 | // be proposed if the leader's applied index is greater than this |
| 284 | // value. |
| 285 | pendingConfIndex uint64 |
| 286 | // an estimate of the size of the uncommitted tail of the Raft log. Used to |
| 287 | // prevent unbounded log growth. Only maintained by the leader. Reset on |
| 288 | // term changes. |
| 289 | uncommittedSize uint64 |
| 290 | |
| 291 | readOnly *readOnly |
| 292 | |
| 293 | // number of ticks since it reached last electionTimeout when it is leader |
| 294 | // or candidate. |
| 295 | // number of ticks since it reached last electionTimeout or received a |
| 296 | // valid message from current leader when it is a follower. |
| 297 | electionElapsed int |
| 298 | |
| 299 | // number of ticks since it reached last heartbeatTimeout. |
| 300 | // only leader keeps heartbeatElapsed. |
| 301 | heartbeatElapsed int |
| 302 | |
| 303 | checkQuorum bool |
| 304 | preVote bool |
| 305 | |
| 306 | heartbeatTimeout int |
| 307 | electionTimeout int |
| 308 | // randomizedElectionTimeout is a random number between |
| 309 | // [electiontimeout, 2 * electiontimeout - 1]. It gets reset |
| 310 | // when raft changes its state to follower or candidate. |
| 311 | randomizedElectionTimeout int |
| 312 | disableProposalForwarding bool |
| 313 | |
| 314 | tick func() |
| 315 | step stepFunc |
| 316 | |
| 317 | logger Logger |
| 318 | } |
| 319 | |
| 320 | func newRaft(c *Config) *raft { |
| 321 | if err := c.validate(); err != nil { |
| 322 | panic(err.Error()) |
| 323 | } |
| 324 | raftlog := newLogWithSize(c.Storage, c.Logger, c.MaxCommittedSizePerReady) |
| 325 | hs, cs, err := c.Storage.InitialState() |
| 326 | if err != nil { |
| 327 | panic(err) // TODO(bdarnell) |
| 328 | } |
| 329 | peers := c.peers |
| 330 | learners := c.learners |
| 331 | if len(cs.Nodes) > 0 || len(cs.Learners) > 0 { |
| 332 | if len(peers) > 0 || len(learners) > 0 { |
| 333 | // TODO(bdarnell): the peers argument is always nil except in |
| 334 | // tests; the argument should be removed and these tests should be |
| 335 | // updated to specify their nodes through a snapshot. |
| 336 | panic("cannot specify both newRaft(peers, learners) and ConfState.(Nodes, Learners)") |
| 337 | } |
| 338 | peers = cs.Nodes |
| 339 | learners = cs.Learners |
| 340 | } |
| 341 | r := &raft{ |
| 342 | id: c.ID, |
| 343 | lead: None, |
| 344 | isLearner: false, |
| 345 | raftLog: raftlog, |
| 346 | maxMsgSize: c.MaxSizePerMsg, |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 347 | maxUncommittedSize: c.MaxUncommittedEntriesSize, |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 348 | prs: tracker.MakeProgressTracker(c.MaxInflightMsgs), |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 349 | electionTimeout: c.ElectionTick, |
| 350 | heartbeatTimeout: c.HeartbeatTick, |
| 351 | logger: c.Logger, |
| 352 | checkQuorum: c.CheckQuorum, |
| 353 | preVote: c.PreVote, |
| 354 | readOnly: newReadOnly(c.ReadOnlyOption), |
| 355 | disableProposalForwarding: c.DisableProposalForwarding, |
| 356 | } |
| 357 | for _, p := range peers { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 358 | // Add node to active config. |
| 359 | r.prs.InitProgress(p, 0 /* match */, 1 /* next */, false /* isLearner */) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 360 | } |
| 361 | for _, p := range learners { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 362 | // Add learner to active config. |
| 363 | r.prs.InitProgress(p, 0 /* match */, 1 /* next */, true /* isLearner */) |
| 364 | |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 365 | if r.id == p { |
| 366 | r.isLearner = true |
| 367 | } |
| 368 | } |
| 369 | |
| 370 | if !isHardStateEqual(hs, emptyState) { |
| 371 | r.loadState(hs) |
| 372 | } |
| 373 | if c.Applied > 0 { |
| 374 | raftlog.appliedTo(c.Applied) |
| 375 | } |
| 376 | r.becomeFollower(r.Term, None) |
| 377 | |
| 378 | var nodesStrs []string |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 379 | for _, n := range r.prs.VoterNodes() { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 380 | nodesStrs = append(nodesStrs, fmt.Sprintf("%x", n)) |
| 381 | } |
| 382 | |
| 383 | r.logger.Infof("newRaft %x [peers: [%s], term: %d, commit: %d, applied: %d, lastindex: %d, lastterm: %d]", |
| 384 | r.id, strings.Join(nodesStrs, ","), r.Term, r.raftLog.committed, r.raftLog.applied, r.raftLog.lastIndex(), r.raftLog.lastTerm()) |
| 385 | return r |
| 386 | } |
| 387 | |
| 388 | func (r *raft) hasLeader() bool { return r.lead != None } |
| 389 | |
| 390 | func (r *raft) softState() *SoftState { return &SoftState{Lead: r.lead, RaftState: r.state} } |
| 391 | |
| 392 | func (r *raft) hardState() pb.HardState { |
| 393 | return pb.HardState{ |
| 394 | Term: r.Term, |
| 395 | Vote: r.Vote, |
| 396 | Commit: r.raftLog.committed, |
| 397 | } |
| 398 | } |
| 399 | |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 400 | // send persists state to stable storage and then sends to its mailbox. |
| 401 | func (r *raft) send(m pb.Message) { |
| 402 | m.From = r.id |
| 403 | if m.Type == pb.MsgVote || m.Type == pb.MsgVoteResp || m.Type == pb.MsgPreVote || m.Type == pb.MsgPreVoteResp { |
| 404 | if m.Term == 0 { |
| 405 | // All {pre-,}campaign messages need to have the term set when |
| 406 | // sending. |
| 407 | // - MsgVote: m.Term is the term the node is campaigning for, |
| 408 | // non-zero as we increment the term when campaigning. |
| 409 | // - MsgVoteResp: m.Term is the new r.Term if the MsgVote was |
| 410 | // granted, non-zero for the same reason MsgVote is |
| 411 | // - MsgPreVote: m.Term is the term the node will campaign, |
| 412 | // non-zero as we use m.Term to indicate the next term we'll be |
| 413 | // campaigning for |
| 414 | // - MsgPreVoteResp: m.Term is the term received in the original |
| 415 | // MsgPreVote if the pre-vote was granted, non-zero for the |
| 416 | // same reasons MsgPreVote is |
| 417 | panic(fmt.Sprintf("term should be set when sending %s", m.Type)) |
| 418 | } |
| 419 | } else { |
| 420 | if m.Term != 0 { |
| 421 | panic(fmt.Sprintf("term should not be set when sending %s (was %d)", m.Type, m.Term)) |
| 422 | } |
| 423 | // do not attach term to MsgProp, MsgReadIndex |
| 424 | // proposals are a way to forward to the leader and |
| 425 | // should be treated as local message. |
| 426 | // MsgReadIndex is also forwarded to leader. |
| 427 | if m.Type != pb.MsgProp && m.Type != pb.MsgReadIndex { |
| 428 | m.Term = r.Term |
| 429 | } |
| 430 | } |
| 431 | r.msgs = append(r.msgs, m) |
| 432 | } |
| 433 | |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 434 | // sendAppend sends an append RPC with new entries (if any) and the |
| 435 | // current commit index to the given peer. |
| 436 | func (r *raft) sendAppend(to uint64) { |
| 437 | r.maybeSendAppend(to, true) |
| 438 | } |
| 439 | |
| 440 | // maybeSendAppend sends an append RPC with new entries to the given peer, |
| 441 | // if necessary. Returns true if a message was sent. The sendIfEmpty |
| 442 | // argument controls whether messages with no entries will be sent |
| 443 | // ("empty" messages are useful to convey updated Commit indexes, but |
| 444 | // are undesirable when we're sending multiple messages in a batch). |
| 445 | func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 446 | pr := r.prs.Progress[to] |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 447 | if pr.IsPaused() { |
| 448 | return false |
| 449 | } |
| 450 | m := pb.Message{} |
| 451 | m.To = to |
| 452 | |
| 453 | term, errt := r.raftLog.term(pr.Next - 1) |
| 454 | ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize) |
| 455 | if len(ents) == 0 && !sendIfEmpty { |
| 456 | return false |
| 457 | } |
| 458 | |
| 459 | if errt != nil || erre != nil { // send snapshot if we failed to get term or entries |
| 460 | if !pr.RecentActive { |
| 461 | r.logger.Debugf("ignore sending snapshot to %x since it is not recently active", to) |
| 462 | return false |
| 463 | } |
| 464 | |
| 465 | m.Type = pb.MsgSnap |
| 466 | snapshot, err := r.raftLog.snapshot() |
| 467 | if err != nil { |
| 468 | if err == ErrSnapshotTemporarilyUnavailable { |
| 469 | r.logger.Debugf("%x failed to send snapshot to %x because snapshot is temporarily unavailable", r.id, to) |
| 470 | return false |
| 471 | } |
| 472 | panic(err) // TODO(bdarnell) |
| 473 | } |
| 474 | if IsEmptySnap(snapshot) { |
| 475 | panic("need non-empty snapshot") |
| 476 | } |
| 477 | m.Snapshot = snapshot |
| 478 | sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term |
| 479 | r.logger.Debugf("%x [firstindex: %d, commit: %d] sent snapshot[index: %d, term: %d] to %x [%s]", |
| 480 | r.id, r.raftLog.firstIndex(), r.raftLog.committed, sindex, sterm, to, pr) |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 481 | pr.BecomeSnapshot(sindex) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 482 | r.logger.Debugf("%x paused sending replication messages to %x [%s]", r.id, to, pr) |
| 483 | } else { |
| 484 | m.Type = pb.MsgApp |
| 485 | m.Index = pr.Next - 1 |
| 486 | m.LogTerm = term |
| 487 | m.Entries = ents |
| 488 | m.Commit = r.raftLog.committed |
| 489 | if n := len(m.Entries); n != 0 { |
| 490 | switch pr.State { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 491 | // optimistically increase the next when in StateReplicate |
| 492 | case tracker.StateReplicate: |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 493 | last := m.Entries[n-1].Index |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 494 | pr.OptimisticUpdate(last) |
| 495 | pr.Inflights.Add(last) |
| 496 | case tracker.StateProbe: |
| 497 | pr.ProbeSent = true |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 498 | default: |
| 499 | r.logger.Panicf("%x is sending append in unhandled state %s", r.id, pr.State) |
| 500 | } |
| 501 | } |
| 502 | } |
| 503 | r.send(m) |
| 504 | return true |
| 505 | } |
| 506 | |
| 507 | // sendHeartbeat sends a heartbeat RPC to the given peer. |
| 508 | func (r *raft) sendHeartbeat(to uint64, ctx []byte) { |
| 509 | // Attach the commit as min(to.matched, r.committed). |
| 510 | // When the leader sends out heartbeat message, |
| 511 | // the receiver(follower) might not be matched with the leader |
| 512 | // or it might not have all the committed entries. |
| 513 | // The leader MUST NOT forward the follower's commit to |
| 514 | // an unmatched index. |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 515 | commit := min(r.prs.Progress[to].Match, r.raftLog.committed) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 516 | m := pb.Message{ |
| 517 | To: to, |
| 518 | Type: pb.MsgHeartbeat, |
| 519 | Commit: commit, |
| 520 | Context: ctx, |
| 521 | } |
| 522 | |
| 523 | r.send(m) |
| 524 | } |
| 525 | |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 526 | // bcastAppend sends RPC, with entries to all peers that are not up-to-date |
| 527 | // according to the progress recorded in r.prs. |
| 528 | func (r *raft) bcastAppend() { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 529 | r.prs.Visit(func(id uint64, _ *tracker.Progress) { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 530 | if id == r.id { |
| 531 | return |
| 532 | } |
| 533 | |
| 534 | r.sendAppend(id) |
| 535 | }) |
| 536 | } |
| 537 | |
| 538 | // bcastHeartbeat sends RPC, without entries to all the peers. |
| 539 | func (r *raft) bcastHeartbeat() { |
| 540 | lastCtx := r.readOnly.lastPendingRequestCtx() |
| 541 | if len(lastCtx) == 0 { |
| 542 | r.bcastHeartbeatWithCtx(nil) |
| 543 | } else { |
| 544 | r.bcastHeartbeatWithCtx([]byte(lastCtx)) |
| 545 | } |
| 546 | } |
| 547 | |
| 548 | func (r *raft) bcastHeartbeatWithCtx(ctx []byte) { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 549 | r.prs.Visit(func(id uint64, _ *tracker.Progress) { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 550 | if id == r.id { |
| 551 | return |
| 552 | } |
| 553 | r.sendHeartbeat(id, ctx) |
| 554 | }) |
| 555 | } |
| 556 | |
| 557 | // maybeCommit attempts to advance the commit index. Returns true if |
| 558 | // the commit index changed (in which case the caller should call |
| 559 | // r.bcastAppend). |
| 560 | func (r *raft) maybeCommit() bool { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 561 | mci := r.prs.Committed() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 562 | return r.raftLog.maybeCommit(mci, r.Term) |
| 563 | } |
| 564 | |
| 565 | func (r *raft) reset(term uint64) { |
| 566 | if r.Term != term { |
| 567 | r.Term = term |
| 568 | r.Vote = None |
| 569 | } |
| 570 | r.lead = None |
| 571 | |
| 572 | r.electionElapsed = 0 |
| 573 | r.heartbeatElapsed = 0 |
| 574 | r.resetRandomizedElectionTimeout() |
| 575 | |
| 576 | r.abortLeaderTransfer() |
| 577 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 578 | r.prs.ResetVotes() |
| 579 | r.prs.Visit(func(id uint64, pr *tracker.Progress) { |
| 580 | *pr = tracker.Progress{ |
| 581 | Match: 0, |
| 582 | Next: r.raftLog.lastIndex() + 1, |
| 583 | Inflights: tracker.NewInflights(r.prs.MaxInflight), |
| 584 | IsLearner: pr.IsLearner, |
| 585 | } |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 586 | if id == r.id { |
| 587 | pr.Match = r.raftLog.lastIndex() |
| 588 | } |
| 589 | }) |
| 590 | |
| 591 | r.pendingConfIndex = 0 |
| 592 | r.uncommittedSize = 0 |
| 593 | r.readOnly = newReadOnly(r.readOnly.option) |
| 594 | } |
| 595 | |
| 596 | func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) { |
| 597 | li := r.raftLog.lastIndex() |
| 598 | for i := range es { |
| 599 | es[i].Term = r.Term |
| 600 | es[i].Index = li + 1 + uint64(i) |
| 601 | } |
| 602 | // Track the size of this uncommitted proposal. |
| 603 | if !r.increaseUncommittedSize(es) { |
| 604 | r.logger.Debugf( |
| 605 | "%x appending new entries to log would exceed uncommitted entry size limit; dropping proposal", |
| 606 | r.id, |
| 607 | ) |
| 608 | // Drop the proposal. |
| 609 | return false |
| 610 | } |
| 611 | // use latest "last" index after truncate/append |
| 612 | li = r.raftLog.append(es...) |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 613 | r.prs.Progress[r.id].MaybeUpdate(li) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 614 | // Regardless of maybeCommit's return, our caller will call bcastAppend. |
| 615 | r.maybeCommit() |
| 616 | return true |
| 617 | } |
| 618 | |
| 619 | // tickElection is run by followers and candidates after r.electionTimeout. |
| 620 | func (r *raft) tickElection() { |
| 621 | r.electionElapsed++ |
| 622 | |
| 623 | if r.promotable() && r.pastElectionTimeout() { |
| 624 | r.electionElapsed = 0 |
| 625 | r.Step(pb.Message{From: r.id, Type: pb.MsgHup}) |
| 626 | } |
| 627 | } |
| 628 | |
| 629 | // tickHeartbeat is run by leaders to send a MsgBeat after r.heartbeatTimeout. |
| 630 | func (r *raft) tickHeartbeat() { |
| 631 | r.heartbeatElapsed++ |
| 632 | r.electionElapsed++ |
| 633 | |
| 634 | if r.electionElapsed >= r.electionTimeout { |
| 635 | r.electionElapsed = 0 |
| 636 | if r.checkQuorum { |
| 637 | r.Step(pb.Message{From: r.id, Type: pb.MsgCheckQuorum}) |
| 638 | } |
| 639 | // If current leader cannot transfer leadership in electionTimeout, it becomes leader again. |
| 640 | if r.state == StateLeader && r.leadTransferee != None { |
| 641 | r.abortLeaderTransfer() |
| 642 | } |
| 643 | } |
| 644 | |
| 645 | if r.state != StateLeader { |
| 646 | return |
| 647 | } |
| 648 | |
| 649 | if r.heartbeatElapsed >= r.heartbeatTimeout { |
| 650 | r.heartbeatElapsed = 0 |
| 651 | r.Step(pb.Message{From: r.id, Type: pb.MsgBeat}) |
| 652 | } |
| 653 | } |
| 654 | |
| 655 | func (r *raft) becomeFollower(term uint64, lead uint64) { |
| 656 | r.step = stepFollower |
| 657 | r.reset(term) |
| 658 | r.tick = r.tickElection |
| 659 | r.lead = lead |
| 660 | r.state = StateFollower |
| 661 | r.logger.Infof("%x became follower at term %d", r.id, r.Term) |
| 662 | } |
| 663 | |
| 664 | func (r *raft) becomeCandidate() { |
| 665 | // TODO(xiangli) remove the panic when the raft implementation is stable |
| 666 | if r.state == StateLeader { |
| 667 | panic("invalid transition [leader -> candidate]") |
| 668 | } |
| 669 | r.step = stepCandidate |
| 670 | r.reset(r.Term + 1) |
| 671 | r.tick = r.tickElection |
| 672 | r.Vote = r.id |
| 673 | r.state = StateCandidate |
| 674 | r.logger.Infof("%x became candidate at term %d", r.id, r.Term) |
| 675 | } |
| 676 | |
| 677 | func (r *raft) becomePreCandidate() { |
| 678 | // TODO(xiangli) remove the panic when the raft implementation is stable |
| 679 | if r.state == StateLeader { |
| 680 | panic("invalid transition [leader -> pre-candidate]") |
| 681 | } |
| 682 | // Becoming a pre-candidate changes our step functions and state, |
| 683 | // but doesn't change anything else. In particular it does not increase |
| 684 | // r.Term or change r.Vote. |
| 685 | r.step = stepCandidate |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 686 | r.prs.ResetVotes() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 687 | r.tick = r.tickElection |
| 688 | r.lead = None |
| 689 | r.state = StatePreCandidate |
| 690 | r.logger.Infof("%x became pre-candidate at term %d", r.id, r.Term) |
| 691 | } |
| 692 | |
| 693 | func (r *raft) becomeLeader() { |
| 694 | // TODO(xiangli) remove the panic when the raft implementation is stable |
| 695 | if r.state == StateFollower { |
| 696 | panic("invalid transition [follower -> leader]") |
| 697 | } |
| 698 | r.step = stepLeader |
| 699 | r.reset(r.Term) |
| 700 | r.tick = r.tickHeartbeat |
| 701 | r.lead = r.id |
| 702 | r.state = StateLeader |
| 703 | // Followers enter replicate mode when they've been successfully probed |
| 704 | // (perhaps after having received a snapshot as a result). The leader is |
| 705 | // trivially in this state. Note that r.reset() has initialized this |
| 706 | // progress with the last index already. |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 707 | r.prs.Progress[r.id].BecomeReplicate() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 708 | |
| 709 | // Conservatively set the pendingConfIndex to the last index in the |
| 710 | // log. There may or may not be a pending config change, but it's |
| 711 | // safe to delay any future proposals until we commit all our |
| 712 | // pending log entries, and scanning the entire tail of the log |
| 713 | // could be expensive. |
| 714 | r.pendingConfIndex = r.raftLog.lastIndex() |
| 715 | |
| 716 | emptyEnt := pb.Entry{Data: nil} |
| 717 | if !r.appendEntry(emptyEnt) { |
| 718 | // This won't happen because we just called reset() above. |
| 719 | r.logger.Panic("empty entry was dropped") |
| 720 | } |
| 721 | // As a special case, don't count the initial empty entry towards the |
| 722 | // uncommitted log quota. This is because we want to preserve the |
| 723 | // behavior of allowing one entry larger than quota if the current |
| 724 | // usage is zero. |
| 725 | r.reduceUncommittedSize([]pb.Entry{emptyEnt}) |
| 726 | r.logger.Infof("%x became leader at term %d", r.id, r.Term) |
| 727 | } |
| 728 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 729 | // campaign transitions the raft instance to candidate state. This must only be |
| 730 | // called after verifying that this is a legitimate transition. |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 731 | func (r *raft) campaign(t CampaignType) { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 732 | if !r.promotable() { |
| 733 | // This path should not be hit (callers are supposed to check), but |
| 734 | // better safe than sorry. |
| 735 | r.logger.Warningf("%x is unpromotable; campaign() should have been called", r.id) |
| 736 | } |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 737 | var term uint64 |
| 738 | var voteMsg pb.MessageType |
| 739 | if t == campaignPreElection { |
| 740 | r.becomePreCandidate() |
| 741 | voteMsg = pb.MsgPreVote |
| 742 | // PreVote RPCs are sent for the next term before we've incremented r.Term. |
| 743 | term = r.Term + 1 |
| 744 | } else { |
| 745 | r.becomeCandidate() |
| 746 | voteMsg = pb.MsgVote |
| 747 | term = r.Term |
| 748 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 749 | if _, _, res := r.poll(r.id, voteRespMsgType(voteMsg), true); res == quorum.VoteWon { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 750 | // We won the election after voting for ourselves (which must mean that |
| 751 | // this is a single-node cluster). Advance to the next state. |
| 752 | if t == campaignPreElection { |
| 753 | r.campaign(campaignElection) |
| 754 | } else { |
| 755 | r.becomeLeader() |
| 756 | } |
| 757 | return |
| 758 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 759 | for id := range r.prs.Voters.IDs() { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 760 | if id == r.id { |
| 761 | continue |
| 762 | } |
| 763 | r.logger.Infof("%x [logterm: %d, index: %d] sent %s request to %x at term %d", |
| 764 | r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), voteMsg, id, r.Term) |
| 765 | |
| 766 | var ctx []byte |
| 767 | if t == campaignTransfer { |
| 768 | ctx = []byte(t) |
| 769 | } |
| 770 | r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx}) |
| 771 | } |
| 772 | } |
| 773 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 774 | func (r *raft) poll(id uint64, t pb.MessageType, v bool) (granted int, rejected int, result quorum.VoteResult) { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 775 | if v { |
| 776 | r.logger.Infof("%x received %s from %x at term %d", r.id, t, id, r.Term) |
| 777 | } else { |
| 778 | r.logger.Infof("%x received %s rejection from %x at term %d", r.id, t, id, r.Term) |
| 779 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 780 | r.prs.RecordVote(id, v) |
| 781 | return r.prs.TallyVotes() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 782 | } |
| 783 | |
| 784 | func (r *raft) Step(m pb.Message) error { |
| 785 | // Handle the message term, which may result in our stepping down to a follower. |
| 786 | switch { |
| 787 | case m.Term == 0: |
| 788 | // local message |
| 789 | case m.Term > r.Term: |
| 790 | if m.Type == pb.MsgVote || m.Type == pb.MsgPreVote { |
| 791 | force := bytes.Equal(m.Context, []byte(campaignTransfer)) |
| 792 | inLease := r.checkQuorum && r.lead != None && r.electionElapsed < r.electionTimeout |
| 793 | if !force && inLease { |
| 794 | // If a server receives a RequestVote request within the minimum election timeout |
| 795 | // of hearing from a current leader, it does not update its term or grant its vote |
| 796 | r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] ignored %s from %x [logterm: %d, index: %d] at term %d: lease is not expired (remaining ticks: %d)", |
| 797 | r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term, r.electionTimeout-r.electionElapsed) |
| 798 | return nil |
| 799 | } |
| 800 | } |
| 801 | switch { |
| 802 | case m.Type == pb.MsgPreVote: |
| 803 | // Never change our term in response to a PreVote |
| 804 | case m.Type == pb.MsgPreVoteResp && !m.Reject: |
| 805 | // We send pre-vote requests with a term in our future. If the |
| 806 | // pre-vote is granted, we will increment our term when we get a |
| 807 | // quorum. If it is not, the term comes from the node that |
| 808 | // rejected our vote so we should become a follower at the new |
| 809 | // term. |
| 810 | default: |
| 811 | r.logger.Infof("%x [term: %d] received a %s message with higher term from %x [term: %d]", |
| 812 | r.id, r.Term, m.Type, m.From, m.Term) |
| 813 | if m.Type == pb.MsgApp || m.Type == pb.MsgHeartbeat || m.Type == pb.MsgSnap { |
| 814 | r.becomeFollower(m.Term, m.From) |
| 815 | } else { |
| 816 | r.becomeFollower(m.Term, None) |
| 817 | } |
| 818 | } |
| 819 | |
| 820 | case m.Term < r.Term: |
| 821 | if (r.checkQuorum || r.preVote) && (m.Type == pb.MsgHeartbeat || m.Type == pb.MsgApp) { |
| 822 | // We have received messages from a leader at a lower term. It is possible |
| 823 | // that these messages were simply delayed in the network, but this could |
| 824 | // also mean that this node has advanced its term number during a network |
| 825 | // partition, and it is now unable to either win an election or to rejoin |
| 826 | // the majority on the old term. If checkQuorum is false, this will be |
| 827 | // handled by incrementing term numbers in response to MsgVote with a |
| 828 | // higher term, but if checkQuorum is true we may not advance the term on |
| 829 | // MsgVote and must generate other messages to advance the term. The net |
| 830 | // result of these two features is to minimize the disruption caused by |
| 831 | // nodes that have been removed from the cluster's configuration: a |
| 832 | // removed node will send MsgVotes (or MsgPreVotes) which will be ignored, |
| 833 | // but it will not receive MsgApp or MsgHeartbeat, so it will not create |
| 834 | // disruptive term increases, by notifying leader of this node's activeness. |
| 835 | // The above comments also true for Pre-Vote |
| 836 | // |
| 837 | // When follower gets isolated, it soon starts an election ending |
| 838 | // up with a higher term than leader, although it won't receive enough |
| 839 | // votes to win the election. When it regains connectivity, this response |
| 840 | // with "pb.MsgAppResp" of higher term would force leader to step down. |
| 841 | // However, this disruption is inevitable to free this stuck node with |
| 842 | // fresh election. This can be prevented with Pre-Vote phase. |
| 843 | r.send(pb.Message{To: m.From, Type: pb.MsgAppResp}) |
| 844 | } else if m.Type == pb.MsgPreVote { |
| 845 | // Before Pre-Vote enable, there may have candidate with higher term, |
| 846 | // but less log. After update to Pre-Vote, the cluster may deadlock if |
| 847 | // we drop messages with a lower term. |
| 848 | r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected %s from %x [logterm: %d, index: %d] at term %d", |
| 849 | r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term) |
| 850 | r.send(pb.Message{To: m.From, Term: r.Term, Type: pb.MsgPreVoteResp, Reject: true}) |
| 851 | } else { |
| 852 | // ignore other cases |
| 853 | r.logger.Infof("%x [term: %d] ignored a %s message with lower term from %x [term: %d]", |
| 854 | r.id, r.Term, m.Type, m.From, m.Term) |
| 855 | } |
| 856 | return nil |
| 857 | } |
| 858 | |
| 859 | switch m.Type { |
| 860 | case pb.MsgHup: |
| 861 | if r.state != StateLeader { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 862 | if !r.promotable() { |
| 863 | r.logger.Warningf("%x is unpromotable and can not campaign; ignoring MsgHup", r.id) |
| 864 | return nil |
| 865 | } |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 866 | ents, err := r.raftLog.slice(r.raftLog.applied+1, r.raftLog.committed+1, noLimit) |
| 867 | if err != nil { |
| 868 | r.logger.Panicf("unexpected error getting unapplied entries (%v)", err) |
| 869 | } |
| 870 | if n := numOfPendingConf(ents); n != 0 && r.raftLog.committed > r.raftLog.applied { |
| 871 | r.logger.Warningf("%x cannot campaign at term %d since there are still %d pending configuration changes to apply", r.id, r.Term, n) |
| 872 | return nil |
| 873 | } |
| 874 | |
| 875 | r.logger.Infof("%x is starting a new election at term %d", r.id, r.Term) |
| 876 | if r.preVote { |
| 877 | r.campaign(campaignPreElection) |
| 878 | } else { |
| 879 | r.campaign(campaignElection) |
| 880 | } |
| 881 | } else { |
| 882 | r.logger.Debugf("%x ignoring MsgHup because already leader", r.id) |
| 883 | } |
| 884 | |
| 885 | case pb.MsgVote, pb.MsgPreVote: |
| 886 | if r.isLearner { |
| 887 | // TODO: learner may need to vote, in case of node down when confchange. |
| 888 | r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] ignored %s from %x [logterm: %d, index: %d] at term %d: learner can not vote", |
| 889 | r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term) |
| 890 | return nil |
| 891 | } |
| 892 | // We can vote if this is a repeat of a vote we've already cast... |
| 893 | canVote := r.Vote == m.From || |
| 894 | // ...we haven't voted and we don't think there's a leader yet in this term... |
| 895 | (r.Vote == None && r.lead == None) || |
| 896 | // ...or this is a PreVote for a future term... |
| 897 | (m.Type == pb.MsgPreVote && m.Term > r.Term) |
| 898 | // ...and we believe the candidate is up to date. |
| 899 | if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) { |
| 900 | r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] cast %s for %x [logterm: %d, index: %d] at term %d", |
| 901 | r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term) |
| 902 | // When responding to Msg{Pre,}Vote messages we include the term |
| 903 | // from the message, not the local term. To see why, consider the |
| 904 | // case where a single node was previously partitioned away and |
| 905 | // it's local term is now out of date. If we include the local term |
| 906 | // (recall that for pre-votes we don't update the local term), the |
| 907 | // (pre-)campaigning node on the other end will proceed to ignore |
| 908 | // the message (it ignores all out of date messages). |
| 909 | // The term in the original message and current local term are the |
| 910 | // same in the case of regular votes, but different for pre-votes. |
| 911 | r.send(pb.Message{To: m.From, Term: m.Term, Type: voteRespMsgType(m.Type)}) |
| 912 | if m.Type == pb.MsgVote { |
| 913 | // Only record real votes. |
| 914 | r.electionElapsed = 0 |
| 915 | r.Vote = m.From |
| 916 | } |
| 917 | } else { |
| 918 | r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected %s from %x [logterm: %d, index: %d] at term %d", |
| 919 | r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term) |
| 920 | r.send(pb.Message{To: m.From, Term: r.Term, Type: voteRespMsgType(m.Type), Reject: true}) |
| 921 | } |
| 922 | |
| 923 | default: |
| 924 | err := r.step(r, m) |
| 925 | if err != nil { |
| 926 | return err |
| 927 | } |
| 928 | } |
| 929 | return nil |
| 930 | } |
| 931 | |
| 932 | type stepFunc func(r *raft, m pb.Message) error |
| 933 | |
| 934 | func stepLeader(r *raft, m pb.Message) error { |
| 935 | // These message types do not require any progress for m.From. |
| 936 | switch m.Type { |
| 937 | case pb.MsgBeat: |
| 938 | r.bcastHeartbeat() |
| 939 | return nil |
| 940 | case pb.MsgCheckQuorum: |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 941 | // The leader should always see itself as active. As a precaution, handle |
| 942 | // the case in which the leader isn't in the configuration any more (for |
| 943 | // example if it just removed itself). |
| 944 | // |
| 945 | // TODO(tbg): I added a TODO in removeNode, it doesn't seem that the |
| 946 | // leader steps down when removing itself. I might be missing something. |
| 947 | if pr := r.prs.Progress[r.id]; pr != nil { |
| 948 | pr.RecentActive = true |
| 949 | } |
| 950 | if !r.prs.QuorumActive() { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 951 | r.logger.Warningf("%x stepped down to follower since quorum is not active", r.id) |
| 952 | r.becomeFollower(r.Term, None) |
| 953 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 954 | // Mark everyone (but ourselves) as inactive in preparation for the next |
| 955 | // CheckQuorum. |
| 956 | r.prs.Visit(func(id uint64, pr *tracker.Progress) { |
| 957 | if id != r.id { |
| 958 | pr.RecentActive = false |
| 959 | } |
| 960 | }) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 961 | return nil |
| 962 | case pb.MsgProp: |
| 963 | if len(m.Entries) == 0 { |
| 964 | r.logger.Panicf("%x stepped empty MsgProp", r.id) |
| 965 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 966 | if r.prs.Progress[r.id] == nil { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 967 | // If we are not currently a member of the range (i.e. this node |
| 968 | // was removed from the configuration while serving as leader), |
| 969 | // drop any new proposals. |
| 970 | return ErrProposalDropped |
| 971 | } |
| 972 | if r.leadTransferee != None { |
| 973 | r.logger.Debugf("%x [term %d] transfer leadership to %x is in progress; dropping proposal", r.id, r.Term, r.leadTransferee) |
| 974 | return ErrProposalDropped |
| 975 | } |
| 976 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 977 | for i := range m.Entries { |
| 978 | e := &m.Entries[i] |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 979 | if e.Type == pb.EntryConfChange { |
| 980 | if r.pendingConfIndex > r.raftLog.applied { |
| 981 | r.logger.Infof("propose conf %s ignored since pending unapplied configuration [index %d, applied %d]", |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 982 | e, r.pendingConfIndex, r.raftLog.applied) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 983 | m.Entries[i] = pb.Entry{Type: pb.EntryNormal} |
| 984 | } else { |
| 985 | r.pendingConfIndex = r.raftLog.lastIndex() + uint64(i) + 1 |
| 986 | } |
| 987 | } |
| 988 | } |
| 989 | |
| 990 | if !r.appendEntry(m.Entries...) { |
| 991 | return ErrProposalDropped |
| 992 | } |
| 993 | r.bcastAppend() |
| 994 | return nil |
| 995 | case pb.MsgReadIndex: |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 996 | // If more than the local vote is needed, go through a full broadcast, |
| 997 | // otherwise optimize. |
| 998 | if !r.prs.IsSingleton() { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 999 | if r.raftLog.zeroTermOnErrCompacted(r.raftLog.term(r.raftLog.committed)) != r.Term { |
| 1000 | // Reject read only request when this leader has not committed any log entry at its term. |
| 1001 | return nil |
| 1002 | } |
| 1003 | |
| 1004 | // thinking: use an interally defined context instead of the user given context. |
| 1005 | // We can express this in terms of the term and index instead of a user-supplied value. |
| 1006 | // This would allow multiple reads to piggyback on the same message. |
| 1007 | switch r.readOnly.option { |
| 1008 | case ReadOnlySafe: |
| 1009 | r.readOnly.addRequest(r.raftLog.committed, m) |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1010 | // The local node automatically acks the request. |
| 1011 | r.readOnly.recvAck(r.id, m.Entries[0].Data) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1012 | r.bcastHeartbeatWithCtx(m.Entries[0].Data) |
| 1013 | case ReadOnlyLeaseBased: |
| 1014 | ri := r.raftLog.committed |
| 1015 | if m.From == None || m.From == r.id { // from local member |
| 1016 | r.readStates = append(r.readStates, ReadState{Index: r.raftLog.committed, RequestCtx: m.Entries[0].Data}) |
| 1017 | } else { |
| 1018 | r.send(pb.Message{To: m.From, Type: pb.MsgReadIndexResp, Index: ri, Entries: m.Entries}) |
| 1019 | } |
| 1020 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1021 | } else { // only one voting member (the leader) in the cluster |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1022 | if m.From == None || m.From == r.id { // from leader itself |
| 1023 | r.readStates = append(r.readStates, ReadState{Index: r.raftLog.committed, RequestCtx: m.Entries[0].Data}) |
| 1024 | } else { // from learner member |
| 1025 | r.send(pb.Message{To: m.From, Type: pb.MsgReadIndexResp, Index: r.raftLog.committed, Entries: m.Entries}) |
| 1026 | } |
| 1027 | } |
| 1028 | |
| 1029 | return nil |
| 1030 | } |
| 1031 | |
| 1032 | // All other message types require a progress for m.From (pr). |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1033 | pr := r.prs.Progress[m.From] |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1034 | if pr == nil { |
| 1035 | r.logger.Debugf("%x no progress available for %x", r.id, m.From) |
| 1036 | return nil |
| 1037 | } |
| 1038 | switch m.Type { |
| 1039 | case pb.MsgAppResp: |
| 1040 | pr.RecentActive = true |
| 1041 | |
| 1042 | if m.Reject { |
| 1043 | r.logger.Debugf("%x received msgApp rejection(lastindex: %d) from %x for index %d", |
| 1044 | r.id, m.RejectHint, m.From, m.Index) |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1045 | if pr.MaybeDecrTo(m.Index, m.RejectHint) { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1046 | r.logger.Debugf("%x decreased progress of %x to [%s]", r.id, m.From, pr) |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1047 | if pr.State == tracker.StateReplicate { |
| 1048 | pr.BecomeProbe() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1049 | } |
| 1050 | r.sendAppend(m.From) |
| 1051 | } |
| 1052 | } else { |
| 1053 | oldPaused := pr.IsPaused() |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1054 | if pr.MaybeUpdate(m.Index) { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1055 | switch { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1056 | case pr.State == tracker.StateProbe: |
| 1057 | pr.BecomeReplicate() |
| 1058 | case pr.State == tracker.StateSnapshot && pr.Match >= pr.PendingSnapshot: |
| 1059 | r.logger.Debugf("%x recovered from needing snapshot, resumed sending replication messages to %x [%s]", r.id, m.From, pr) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1060 | // Transition back to replicating state via probing state |
| 1061 | // (which takes the snapshot into account). If we didn't |
| 1062 | // move to replicating state, that would only happen with |
| 1063 | // the next round of appends (but there may not be a next |
| 1064 | // round for a while, exposing an inconsistent RaftStatus). |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1065 | pr.BecomeProbe() |
| 1066 | pr.BecomeReplicate() |
| 1067 | case pr.State == tracker.StateReplicate: |
| 1068 | pr.Inflights.FreeLE(m.Index) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1069 | } |
| 1070 | |
| 1071 | if r.maybeCommit() { |
| 1072 | r.bcastAppend() |
| 1073 | } else if oldPaused { |
| 1074 | // If we were paused before, this node may be missing the |
| 1075 | // latest commit index, so send it. |
| 1076 | r.sendAppend(m.From) |
| 1077 | } |
| 1078 | // We've updated flow control information above, which may |
| 1079 | // allow us to send multiple (size-limited) in-flight messages |
| 1080 | // at once (such as when transitioning from probe to |
| 1081 | // replicate, or when freeTo() covers multiple messages). If |
| 1082 | // we have more entries to send, send as many messages as we |
| 1083 | // can (without sending empty messages for the commit index) |
| 1084 | for r.maybeSendAppend(m.From, false) { |
| 1085 | } |
| 1086 | // Transfer leadership is in progress. |
| 1087 | if m.From == r.leadTransferee && pr.Match == r.raftLog.lastIndex() { |
| 1088 | r.logger.Infof("%x sent MsgTimeoutNow to %x after received MsgAppResp", r.id, m.From) |
| 1089 | r.sendTimeoutNow(m.From) |
| 1090 | } |
| 1091 | } |
| 1092 | } |
| 1093 | case pb.MsgHeartbeatResp: |
| 1094 | pr.RecentActive = true |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1095 | pr.ProbeSent = false |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1096 | |
| 1097 | // free one slot for the full inflights window to allow progress. |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1098 | if pr.State == tracker.StateReplicate && pr.Inflights.Full() { |
| 1099 | pr.Inflights.FreeFirstOne() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1100 | } |
| 1101 | if pr.Match < r.raftLog.lastIndex() { |
| 1102 | r.sendAppend(m.From) |
| 1103 | } |
| 1104 | |
| 1105 | if r.readOnly.option != ReadOnlySafe || len(m.Context) == 0 { |
| 1106 | return nil |
| 1107 | } |
| 1108 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1109 | if r.prs.Voters.VoteResult(r.readOnly.recvAck(m.From, m.Context)) != quorum.VoteWon { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1110 | return nil |
| 1111 | } |
| 1112 | |
| 1113 | rss := r.readOnly.advance(m) |
| 1114 | for _, rs := range rss { |
| 1115 | req := rs.req |
| 1116 | if req.From == None || req.From == r.id { // from local member |
| 1117 | r.readStates = append(r.readStates, ReadState{Index: rs.index, RequestCtx: req.Entries[0].Data}) |
| 1118 | } else { |
| 1119 | r.send(pb.Message{To: req.From, Type: pb.MsgReadIndexResp, Index: rs.index, Entries: req.Entries}) |
| 1120 | } |
| 1121 | } |
| 1122 | case pb.MsgSnapStatus: |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1123 | if pr.State != tracker.StateSnapshot { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1124 | return nil |
| 1125 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1126 | // TODO(tbg): this code is very similar to the snapshot handling in |
| 1127 | // MsgAppResp above. In fact, the code there is more correct than the |
| 1128 | // code here and should likely be updated to match (or even better, the |
| 1129 | // logic pulled into a newly created Progress state machine handler). |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1130 | if !m.Reject { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1131 | pr.BecomeProbe() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1132 | r.logger.Debugf("%x snapshot succeeded, resumed sending replication messages to %x [%s]", r.id, m.From, pr) |
| 1133 | } else { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1134 | // NB: the order here matters or we'll be probing erroneously from |
| 1135 | // the snapshot index, but the snapshot never applied. |
| 1136 | pr.PendingSnapshot = 0 |
| 1137 | pr.BecomeProbe() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1138 | r.logger.Debugf("%x snapshot failed, resumed sending replication messages to %x [%s]", r.id, m.From, pr) |
| 1139 | } |
| 1140 | // If snapshot finish, wait for the msgAppResp from the remote node before sending |
| 1141 | // out the next msgApp. |
| 1142 | // If snapshot failure, wait for a heartbeat interval before next try |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1143 | pr.ProbeSent = true |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1144 | case pb.MsgUnreachable: |
| 1145 | // During optimistic replication, if the remote becomes unreachable, |
| 1146 | // there is huge probability that a MsgApp is lost. |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1147 | if pr.State == tracker.StateReplicate { |
| 1148 | pr.BecomeProbe() |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1149 | } |
| 1150 | r.logger.Debugf("%x failed to send message to %x because it is unreachable [%s]", r.id, m.From, pr) |
| 1151 | case pb.MsgTransferLeader: |
| 1152 | if pr.IsLearner { |
| 1153 | r.logger.Debugf("%x is learner. Ignored transferring leadership", r.id) |
| 1154 | return nil |
| 1155 | } |
| 1156 | leadTransferee := m.From |
| 1157 | lastLeadTransferee := r.leadTransferee |
| 1158 | if lastLeadTransferee != None { |
| 1159 | if lastLeadTransferee == leadTransferee { |
| 1160 | r.logger.Infof("%x [term %d] transfer leadership to %x is in progress, ignores request to same node %x", |
| 1161 | r.id, r.Term, leadTransferee, leadTransferee) |
| 1162 | return nil |
| 1163 | } |
| 1164 | r.abortLeaderTransfer() |
| 1165 | r.logger.Infof("%x [term %d] abort previous transferring leadership to %x", r.id, r.Term, lastLeadTransferee) |
| 1166 | } |
| 1167 | if leadTransferee == r.id { |
| 1168 | r.logger.Debugf("%x is already leader. Ignored transferring leadership to self", r.id) |
| 1169 | return nil |
| 1170 | } |
| 1171 | // Transfer leadership to third party. |
| 1172 | r.logger.Infof("%x [term %d] starts to transfer leadership to %x", r.id, r.Term, leadTransferee) |
| 1173 | // Transfer leadership should be finished in one electionTimeout, so reset r.electionElapsed. |
| 1174 | r.electionElapsed = 0 |
| 1175 | r.leadTransferee = leadTransferee |
| 1176 | if pr.Match == r.raftLog.lastIndex() { |
| 1177 | r.sendTimeoutNow(leadTransferee) |
| 1178 | r.logger.Infof("%x sends MsgTimeoutNow to %x immediately as %x already has up-to-date log", r.id, leadTransferee, leadTransferee) |
| 1179 | } else { |
| 1180 | r.sendAppend(leadTransferee) |
| 1181 | } |
| 1182 | } |
| 1183 | return nil |
| 1184 | } |
| 1185 | |
| 1186 | // stepCandidate is shared by StateCandidate and StatePreCandidate; the difference is |
| 1187 | // whether they respond to MsgVoteResp or MsgPreVoteResp. |
| 1188 | func stepCandidate(r *raft, m pb.Message) error { |
| 1189 | // Only handle vote responses corresponding to our candidacy (while in |
| 1190 | // StateCandidate, we may get stale MsgPreVoteResp messages in this term from |
| 1191 | // our pre-candidate state). |
| 1192 | var myVoteRespType pb.MessageType |
| 1193 | if r.state == StatePreCandidate { |
| 1194 | myVoteRespType = pb.MsgPreVoteResp |
| 1195 | } else { |
| 1196 | myVoteRespType = pb.MsgVoteResp |
| 1197 | } |
| 1198 | switch m.Type { |
| 1199 | case pb.MsgProp: |
| 1200 | r.logger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term) |
| 1201 | return ErrProposalDropped |
| 1202 | case pb.MsgApp: |
| 1203 | r.becomeFollower(m.Term, m.From) // always m.Term == r.Term |
| 1204 | r.handleAppendEntries(m) |
| 1205 | case pb.MsgHeartbeat: |
| 1206 | r.becomeFollower(m.Term, m.From) // always m.Term == r.Term |
| 1207 | r.handleHeartbeat(m) |
| 1208 | case pb.MsgSnap: |
| 1209 | r.becomeFollower(m.Term, m.From) // always m.Term == r.Term |
| 1210 | r.handleSnapshot(m) |
| 1211 | case myVoteRespType: |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1212 | gr, rj, res := r.poll(m.From, m.Type, !m.Reject) |
| 1213 | r.logger.Infof("%x has received %d %s votes and %d vote rejections", r.id, gr, m.Type, rj) |
| 1214 | switch res { |
| 1215 | case quorum.VoteWon: |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1216 | if r.state == StatePreCandidate { |
| 1217 | r.campaign(campaignElection) |
| 1218 | } else { |
| 1219 | r.becomeLeader() |
| 1220 | r.bcastAppend() |
| 1221 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1222 | case quorum.VoteLost: |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1223 | // pb.MsgPreVoteResp contains future term of pre-candidate |
| 1224 | // m.Term > r.Term; reuse r.Term |
| 1225 | r.becomeFollower(r.Term, None) |
| 1226 | } |
| 1227 | case pb.MsgTimeoutNow: |
| 1228 | r.logger.Debugf("%x [term %d state %v] ignored MsgTimeoutNow from %x", r.id, r.Term, r.state, m.From) |
| 1229 | } |
| 1230 | return nil |
| 1231 | } |
| 1232 | |
| 1233 | func stepFollower(r *raft, m pb.Message) error { |
| 1234 | switch m.Type { |
| 1235 | case pb.MsgProp: |
| 1236 | if r.lead == None { |
| 1237 | r.logger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term) |
| 1238 | return ErrProposalDropped |
| 1239 | } else if r.disableProposalForwarding { |
| 1240 | r.logger.Infof("%x not forwarding to leader %x at term %d; dropping proposal", r.id, r.lead, r.Term) |
| 1241 | return ErrProposalDropped |
| 1242 | } |
| 1243 | m.To = r.lead |
| 1244 | r.send(m) |
| 1245 | case pb.MsgApp: |
| 1246 | r.electionElapsed = 0 |
| 1247 | r.lead = m.From |
| 1248 | r.handleAppendEntries(m) |
| 1249 | case pb.MsgHeartbeat: |
| 1250 | r.electionElapsed = 0 |
| 1251 | r.lead = m.From |
| 1252 | r.handleHeartbeat(m) |
| 1253 | case pb.MsgSnap: |
| 1254 | r.electionElapsed = 0 |
| 1255 | r.lead = m.From |
| 1256 | r.handleSnapshot(m) |
| 1257 | case pb.MsgTransferLeader: |
| 1258 | if r.lead == None { |
| 1259 | r.logger.Infof("%x no leader at term %d; dropping leader transfer msg", r.id, r.Term) |
| 1260 | return nil |
| 1261 | } |
| 1262 | m.To = r.lead |
| 1263 | r.send(m) |
| 1264 | case pb.MsgTimeoutNow: |
| 1265 | if r.promotable() { |
| 1266 | r.logger.Infof("%x [term %d] received MsgTimeoutNow from %x and starts an election to get leadership.", r.id, r.Term, m.From) |
| 1267 | // Leadership transfers never use pre-vote even if r.preVote is true; we |
| 1268 | // know we are not recovering from a partition so there is no need for the |
| 1269 | // extra round trip. |
| 1270 | r.campaign(campaignTransfer) |
| 1271 | } else { |
| 1272 | r.logger.Infof("%x received MsgTimeoutNow from %x but is not promotable", r.id, m.From) |
| 1273 | } |
| 1274 | case pb.MsgReadIndex: |
| 1275 | if r.lead == None { |
| 1276 | r.logger.Infof("%x no leader at term %d; dropping index reading msg", r.id, r.Term) |
| 1277 | return nil |
| 1278 | } |
| 1279 | m.To = r.lead |
| 1280 | r.send(m) |
| 1281 | case pb.MsgReadIndexResp: |
| 1282 | if len(m.Entries) != 1 { |
| 1283 | r.logger.Errorf("%x invalid format of MsgReadIndexResp from %x, entries count: %d", r.id, m.From, len(m.Entries)) |
| 1284 | return nil |
| 1285 | } |
| 1286 | r.readStates = append(r.readStates, ReadState{Index: m.Index, RequestCtx: m.Entries[0].Data}) |
| 1287 | } |
| 1288 | return nil |
| 1289 | } |
| 1290 | |
| 1291 | func (r *raft) handleAppendEntries(m pb.Message) { |
| 1292 | if m.Index < r.raftLog.committed { |
| 1293 | r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed}) |
| 1294 | return |
| 1295 | } |
| 1296 | |
| 1297 | if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok { |
| 1298 | r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex}) |
| 1299 | } else { |
| 1300 | r.logger.Debugf("%x [logterm: %d, index: %d] rejected msgApp [logterm: %d, index: %d] from %x", |
| 1301 | r.id, r.raftLog.zeroTermOnErrCompacted(r.raftLog.term(m.Index)), m.Index, m.LogTerm, m.Index, m.From) |
| 1302 | r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: m.Index, Reject: true, RejectHint: r.raftLog.lastIndex()}) |
| 1303 | } |
| 1304 | } |
| 1305 | |
| 1306 | func (r *raft) handleHeartbeat(m pb.Message) { |
| 1307 | r.raftLog.commitTo(m.Commit) |
| 1308 | r.send(pb.Message{To: m.From, Type: pb.MsgHeartbeatResp, Context: m.Context}) |
| 1309 | } |
| 1310 | |
| 1311 | func (r *raft) handleSnapshot(m pb.Message) { |
| 1312 | sindex, sterm := m.Snapshot.Metadata.Index, m.Snapshot.Metadata.Term |
| 1313 | if r.restore(m.Snapshot) { |
| 1314 | r.logger.Infof("%x [commit: %d] restored snapshot [index: %d, term: %d]", |
| 1315 | r.id, r.raftLog.committed, sindex, sterm) |
| 1316 | r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.lastIndex()}) |
| 1317 | } else { |
| 1318 | r.logger.Infof("%x [commit: %d] ignored snapshot [index: %d, term: %d]", |
| 1319 | r.id, r.raftLog.committed, sindex, sterm) |
| 1320 | r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed}) |
| 1321 | } |
| 1322 | } |
| 1323 | |
| 1324 | // restore recovers the state machine from a snapshot. It restores the log and the |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1325 | // configuration of state machine. If this method returns false, the snapshot was |
| 1326 | // ignored, either because it was obsolete or because of an error. |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1327 | func (r *raft) restore(s pb.Snapshot) bool { |
| 1328 | if s.Metadata.Index <= r.raftLog.committed { |
| 1329 | return false |
| 1330 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1331 | if r.state != StateFollower { |
| 1332 | // This is defense-in-depth: if the leader somehow ended up applying a |
| 1333 | // snapshot, it could move into a new term without moving into a |
| 1334 | // follower state. This should never fire, but if it did, we'd have |
| 1335 | // prevented damage by returning early, so log only a loud warning. |
| 1336 | // |
| 1337 | // At the time of writing, the instance is guaranteed to be in follower |
| 1338 | // state when this method is called. |
| 1339 | r.logger.Warningf("%x attempted to restore snapshot as leader; should never happen", r.id) |
| 1340 | r.becomeFollower(r.Term+1, None) |
| 1341 | return false |
| 1342 | } |
| 1343 | |
| 1344 | // More defense-in-depth: throw away snapshot if recipient is not in the |
| 1345 | // config. This shouuldn't ever happen (at the time of writing) but lots of |
| 1346 | // code here and there assumes that r.id is in the progress tracker. |
| 1347 | found := false |
| 1348 | cs := s.Metadata.ConfState |
| 1349 | for _, set := range [][]uint64{ |
| 1350 | cs.Nodes, |
| 1351 | cs.Learners, |
| 1352 | } { |
| 1353 | for _, id := range set { |
| 1354 | if id == r.id { |
| 1355 | found = true |
| 1356 | break |
| 1357 | } |
| 1358 | } |
| 1359 | } |
| 1360 | if !found { |
| 1361 | r.logger.Warningf( |
| 1362 | "%x attempted to restore snapshot but it is not in the ConfState %v; should never happen", |
| 1363 | r.id, cs, |
| 1364 | ) |
| 1365 | return false |
| 1366 | } |
| 1367 | |
| 1368 | // Now go ahead and actually restore. |
| 1369 | |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1370 | if r.raftLog.matchTerm(s.Metadata.Index, s.Metadata.Term) { |
| 1371 | r.logger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] fast-forwarded commit to snapshot [index: %d, term: %d]", |
| 1372 | r.id, r.raftLog.committed, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term) |
| 1373 | r.raftLog.commitTo(s.Metadata.Index) |
| 1374 | return false |
| 1375 | } |
| 1376 | |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1377 | r.raftLog.restore(s) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1378 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1379 | // Reset the configuration and add the (potentially updated) peers in anew. |
| 1380 | r.prs = tracker.MakeProgressTracker(r.prs.MaxInflight) |
| 1381 | for _, id := range s.Metadata.ConfState.Nodes { |
| 1382 | r.applyConfChange(pb.ConfChange{NodeID: id, Type: pb.ConfChangeAddNode}) |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1383 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1384 | for _, id := range s.Metadata.ConfState.Learners { |
| 1385 | r.applyConfChange(pb.ConfChange{NodeID: id, Type: pb.ConfChangeAddLearnerNode}) |
| 1386 | } |
| 1387 | |
| 1388 | pr := r.prs.Progress[r.id] |
| 1389 | pr.MaybeUpdate(pr.Next - 1) // TODO(tbg): this is untested and likely unneeded |
| 1390 | |
| 1391 | r.logger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] restored snapshot [index: %d, term: %d]", |
| 1392 | r.id, r.raftLog.committed, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term) |
| 1393 | return true |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1394 | } |
| 1395 | |
| 1396 | // promotable indicates whether state machine can be promoted to leader, |
| 1397 | // which is true when its own id is in progress list. |
| 1398 | func (r *raft) promotable() bool { |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1399 | pr := r.prs.Progress[r.id] |
| 1400 | return pr != nil && !pr.IsLearner |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1401 | } |
| 1402 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1403 | func (r *raft) applyConfChange(cc pb.ConfChange) pb.ConfState { |
| 1404 | addNodeOrLearnerNode := func(id uint64, isLearner bool) { |
| 1405 | // NB: this method is intentionally hidden from view. All mutations of |
| 1406 | // the conf state must call applyConfChange directly. |
| 1407 | pr := r.prs.Progress[id] |
| 1408 | if pr == nil { |
| 1409 | r.prs.InitProgress(id, 0, r.raftLog.lastIndex()+1, isLearner) |
| 1410 | } else { |
| 1411 | if isLearner && !pr.IsLearner { |
| 1412 | // Can only change Learner to Voter. |
| 1413 | // |
| 1414 | // TODO(tbg): why? |
| 1415 | r.logger.Infof("%x ignored addLearner: do not support changing %x from raft peer to learner.", r.id, id) |
| 1416 | return |
| 1417 | } |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1418 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1419 | if isLearner == pr.IsLearner { |
| 1420 | // Ignore any redundant addNode calls (which can happen because the |
| 1421 | // initial bootstrapping entries are applied twice). |
| 1422 | return |
| 1423 | } |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1424 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1425 | // Change Learner to Voter, use origin Learner progress. |
| 1426 | r.prs.RemoveAny(id) |
| 1427 | r.prs.InitProgress(id, 0 /* match */, 1 /* next */, false /* isLearner */) |
| 1428 | pr.IsLearner = false |
| 1429 | *r.prs.Progress[id] = *pr |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1430 | } |
| 1431 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1432 | // When a node is first added, we should mark it as recently active. |
| 1433 | // Otherwise, CheckQuorum may cause us to step down if it is invoked |
| 1434 | // before the added node has had a chance to communicate with us. |
| 1435 | r.prs.Progress[id].RecentActive = true |
| 1436 | } |
| 1437 | |
| 1438 | var removed int |
| 1439 | if cc.NodeID != None { |
| 1440 | switch cc.Type { |
| 1441 | case pb.ConfChangeAddNode: |
| 1442 | addNodeOrLearnerNode(cc.NodeID, false /* isLearner */) |
| 1443 | case pb.ConfChangeAddLearnerNode: |
| 1444 | addNodeOrLearnerNode(cc.NodeID, true /* isLearner */) |
| 1445 | case pb.ConfChangeRemoveNode: |
| 1446 | removed++ |
| 1447 | r.prs.RemoveAny(cc.NodeID) |
| 1448 | case pb.ConfChangeUpdateNode: |
| 1449 | default: |
| 1450 | panic("unexpected conf type") |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1451 | } |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1452 | } |
| 1453 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1454 | r.logger.Infof("%x switched to configuration %s", r.id, r.prs.Config) |
| 1455 | // Now that the configuration is updated, handle any side effects. |
| 1456 | |
| 1457 | cs := pb.ConfState{Nodes: r.prs.VoterNodes(), Learners: r.prs.LearnerNodes()} |
| 1458 | pr, ok := r.prs.Progress[r.id] |
| 1459 | |
| 1460 | // Update whether the node itself is a learner, resetting to false when the |
| 1461 | // node is removed. |
| 1462 | r.isLearner = ok && pr.IsLearner |
| 1463 | |
| 1464 | if (!ok || r.isLearner) && r.state == StateLeader { |
| 1465 | // This node is leader and was removed or demoted. We prevent demotions |
| 1466 | // at the time writing but hypothetically we handle them the same way as |
| 1467 | // removing the leader: stepping down into the next Term. |
| 1468 | // |
| 1469 | // TODO(tbg): step down (for sanity) and ask follower with largest Match |
| 1470 | // to TimeoutNow (to avoid interruption). This might still drop some |
| 1471 | // proposals but it's better than nothing. |
| 1472 | // |
| 1473 | // TODO(tbg): test this branch. It is untested at the time of writing. |
| 1474 | return cs |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1475 | } |
| 1476 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1477 | // The remaining steps only make sense if this node is the leader and there |
| 1478 | // are other nodes. |
| 1479 | if r.state != StateLeader || len(cs.Nodes) == 0 { |
| 1480 | return cs |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1481 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1482 | if removed > 0 { |
| 1483 | // The quorum size may have been reduced (but not to zero), so see if |
| 1484 | // any pending entries can be committed. |
| 1485 | if r.maybeCommit() { |
| 1486 | r.bcastAppend() |
| 1487 | } |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1488 | } |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1489 | // If the the leadTransferee was removed, abort the leadership transfer. |
| 1490 | if _, tOK := r.prs.Progress[r.leadTransferee]; !tOK && r.leadTransferee != 0 { |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1491 | r.abortLeaderTransfer() |
| 1492 | } |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1493 | |
Abhilash S.L | 3b49463 | 2019-07-16 15:51:09 +0530 | [diff] [blame^] | 1494 | return cs |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1495 | } |
| 1496 | |
| 1497 | func (r *raft) loadState(state pb.HardState) { |
| 1498 | if state.Commit < r.raftLog.committed || state.Commit > r.raftLog.lastIndex() { |
| 1499 | r.logger.Panicf("%x state.commit %d is out of range [%d, %d]", r.id, state.Commit, r.raftLog.committed, r.raftLog.lastIndex()) |
| 1500 | } |
| 1501 | r.raftLog.committed = state.Commit |
| 1502 | r.Term = state.Term |
| 1503 | r.Vote = state.Vote |
| 1504 | } |
| 1505 | |
| 1506 | // pastElectionTimeout returns true iff r.electionElapsed is greater |
| 1507 | // than or equal to the randomized election timeout in |
| 1508 | // [electiontimeout, 2 * electiontimeout - 1]. |
| 1509 | func (r *raft) pastElectionTimeout() bool { |
| 1510 | return r.electionElapsed >= r.randomizedElectionTimeout |
| 1511 | } |
| 1512 | |
| 1513 | func (r *raft) resetRandomizedElectionTimeout() { |
| 1514 | r.randomizedElectionTimeout = r.electionTimeout + globalRand.Intn(r.electionTimeout) |
| 1515 | } |
| 1516 | |
William Kurkian | ea86948 | 2019-04-09 15:16:11 -0400 | [diff] [blame] | 1517 | func (r *raft) sendTimeoutNow(to uint64) { |
| 1518 | r.send(pb.Message{To: to, Type: pb.MsgTimeoutNow}) |
| 1519 | } |
| 1520 | |
| 1521 | func (r *raft) abortLeaderTransfer() { |
| 1522 | r.leadTransferee = None |
| 1523 | } |
| 1524 | |
| 1525 | // increaseUncommittedSize computes the size of the proposed entries and |
| 1526 | // determines whether they would push leader over its maxUncommittedSize limit. |
| 1527 | // If the new entries would exceed the limit, the method returns false. If not, |
| 1528 | // the increase in uncommitted entry size is recorded and the method returns |
| 1529 | // true. |
| 1530 | func (r *raft) increaseUncommittedSize(ents []pb.Entry) bool { |
| 1531 | var s uint64 |
| 1532 | for _, e := range ents { |
| 1533 | s += uint64(PayloadSize(e)) |
| 1534 | } |
| 1535 | |
| 1536 | if r.uncommittedSize > 0 && r.uncommittedSize+s > r.maxUncommittedSize { |
| 1537 | // If the uncommitted tail of the Raft log is empty, allow any size |
| 1538 | // proposal. Otherwise, limit the size of the uncommitted tail of the |
| 1539 | // log and drop any proposal that would push the size over the limit. |
| 1540 | return false |
| 1541 | } |
| 1542 | r.uncommittedSize += s |
| 1543 | return true |
| 1544 | } |
| 1545 | |
| 1546 | // reduceUncommittedSize accounts for the newly committed entries by decreasing |
| 1547 | // the uncommitted entry size limit. |
| 1548 | func (r *raft) reduceUncommittedSize(ents []pb.Entry) { |
| 1549 | if r.uncommittedSize == 0 { |
| 1550 | // Fast-path for followers, who do not track or enforce the limit. |
| 1551 | return |
| 1552 | } |
| 1553 | |
| 1554 | var s uint64 |
| 1555 | for _, e := range ents { |
| 1556 | s += uint64(PayloadSize(e)) |
| 1557 | } |
| 1558 | if s > r.uncommittedSize { |
| 1559 | // uncommittedSize may underestimate the size of the uncommitted Raft |
| 1560 | // log tail but will never overestimate it. Saturate at 0 instead of |
| 1561 | // allowing overflow. |
| 1562 | r.uncommittedSize = 0 |
| 1563 | } else { |
| 1564 | r.uncommittedSize -= s |
| 1565 | } |
| 1566 | } |
| 1567 | |
| 1568 | func numOfPendingConf(ents []pb.Entry) int { |
| 1569 | n := 0 |
| 1570 | for i := range ents { |
| 1571 | if ents[i].Type == pb.EntryConfChange { |
| 1572 | n++ |
| 1573 | } |
| 1574 | } |
| 1575 | return n |
| 1576 | } |