mirror of
https://github.com/ceph/ceph-csi.git
synced 2024-12-04 20:20:19 +00:00
1657 lines
57 KiB
Go
1657 lines
57 KiB
Go
|
// Copyright 2015 The etcd Authors
|
||
|
//
|
||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||
|
// you may not use this file except in compliance with the License.
|
||
|
// You may obtain a copy of the License at
|
||
|
//
|
||
|
// http://www.apache.org/licenses/LICENSE-2.0
|
||
|
//
|
||
|
// Unless required by applicable law or agreed to in writing, software
|
||
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
||
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||
|
// See the License for the specific language governing permissions and
|
||
|
// limitations under the License.
|
||
|
|
||
|
package raft
|
||
|
|
||
|
import (
|
||
|
"bytes"
|
||
|
"errors"
|
||
|
"fmt"
|
||
|
"math"
|
||
|
"math/rand"
|
||
|
"sort"
|
||
|
"strings"
|
||
|
"sync"
|
||
|
"time"
|
||
|
|
||
|
"go.etcd.io/etcd/raft/confchange"
|
||
|
"go.etcd.io/etcd/raft/quorum"
|
||
|
pb "go.etcd.io/etcd/raft/raftpb"
|
||
|
"go.etcd.io/etcd/raft/tracker"
|
||
|
)
|
||
|
|
||
|
// None is a placeholder node ID used when there is no leader.
|
||
|
const None uint64 = 0
|
||
|
const noLimit = math.MaxUint64
|
||
|
|
||
|
// Possible values for StateType.
|
||
|
const (
|
||
|
StateFollower StateType = iota
|
||
|
StateCandidate
|
||
|
StateLeader
|
||
|
StatePreCandidate
|
||
|
numStates
|
||
|
)
|
||
|
|
||
|
type ReadOnlyOption int
|
||
|
|
||
|
const (
|
||
|
// ReadOnlySafe guarantees the linearizability of the read only request by
|
||
|
// communicating with the quorum. It is the default and suggested option.
|
||
|
ReadOnlySafe ReadOnlyOption = iota
|
||
|
// ReadOnlyLeaseBased ensures linearizability of the read only request by
|
||
|
// relying on the leader lease. It can be affected by clock drift.
|
||
|
// If the clock drift is unbounded, leader might keep the lease longer than it
|
||
|
// should (clock can move backward/pause without any bound). ReadIndex is not safe
|
||
|
// in that case.
|
||
|
ReadOnlyLeaseBased
|
||
|
)
|
||
|
|
||
|
// Possible values for CampaignType
|
||
|
const (
|
||
|
// campaignPreElection represents the first phase of a normal election when
|
||
|
// Config.PreVote is true.
|
||
|
campaignPreElection CampaignType = "CampaignPreElection"
|
||
|
// campaignElection represents a normal (time-based) election (the second phase
|
||
|
// of the election when Config.PreVote is true).
|
||
|
campaignElection CampaignType = "CampaignElection"
|
||
|
// campaignTransfer represents the type of leader transfer
|
||
|
campaignTransfer CampaignType = "CampaignTransfer"
|
||
|
)
|
||
|
|
||
|
// ErrProposalDropped is returned when the proposal is ignored by some cases,
|
||
|
// so that the proposer can be notified and fail fast.
|
||
|
var ErrProposalDropped = errors.New("raft proposal dropped")
|
||
|
|
||
|
// lockedRand is a small wrapper around rand.Rand to provide
|
||
|
// synchronization among multiple raft groups. Only the methods needed
|
||
|
// by the code are exposed (e.g. Intn).
|
||
|
type lockedRand struct {
|
||
|
mu sync.Mutex
|
||
|
rand *rand.Rand
|
||
|
}
|
||
|
|
||
|
func (r *lockedRand) Intn(n int) int {
|
||
|
r.mu.Lock()
|
||
|
v := r.rand.Intn(n)
|
||
|
r.mu.Unlock()
|
||
|
return v
|
||
|
}
|
||
|
|
||
|
var globalRand = &lockedRand{
|
||
|
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
|
||
|
}
|
||
|
|
||
|
// CampaignType represents the type of campaigning
|
||
|
// the reason we use the type of string instead of uint64
|
||
|
// is because it's simpler to compare and fill in raft entries
|
||
|
type CampaignType string
|
||
|
|
||
|
// StateType represents the role of a node in a cluster.
|
||
|
type StateType uint64
|
||
|
|
||
|
var stmap = [...]string{
|
||
|
"StateFollower",
|
||
|
"StateCandidate",
|
||
|
"StateLeader",
|
||
|
"StatePreCandidate",
|
||
|
}
|
||
|
|
||
|
func (st StateType) String() string {
|
||
|
return stmap[uint64(st)]
|
||
|
}
|
||
|
|
||
|
// Config contains the parameters to start a raft.
|
||
|
type Config struct {
|
||
|
// ID is the identity of the local raft. ID cannot be 0.
|
||
|
ID uint64
|
||
|
|
||
|
// peers contains the IDs of all nodes (including self) in the raft cluster. It
|
||
|
// should only be set when starting a new raft cluster. Restarting raft from
|
||
|
// previous configuration will panic if peers is set. peer is private and only
|
||
|
// used for testing right now.
|
||
|
peers []uint64
|
||
|
|
||
|
// learners contains the IDs of all learner nodes (including self if the
|
||
|
// local node is a learner) in the raft cluster. learners only receives
|
||
|
// entries from the leader node. It does not vote or promote itself.
|
||
|
learners []uint64
|
||
|
|
||
|
// ElectionTick is the number of Node.Tick invocations that must pass between
|
||
|
// elections. That is, if a follower does not receive any message from the
|
||
|
// leader of current term before ElectionTick has elapsed, it will become
|
||
|
// candidate and start an election. ElectionTick must be greater than
|
||
|
// HeartbeatTick. We suggest ElectionTick = 10 * HeartbeatTick to avoid
|
||
|
// unnecessary leader switching.
|
||
|
ElectionTick int
|
||
|
// HeartbeatTick is the number of Node.Tick invocations that must pass between
|
||
|
// heartbeats. That is, a leader sends heartbeat messages to maintain its
|
||
|
// leadership every HeartbeatTick ticks.
|
||
|
HeartbeatTick int
|
||
|
|
||
|
// Storage is the storage for raft. raft generates entries and states to be
|
||
|
// stored in storage. raft reads the persisted entries and states out of
|
||
|
// Storage when it needs. raft reads out the previous state and configuration
|
||
|
// out of storage when restarting.
|
||
|
Storage Storage
|
||
|
// Applied is the last applied index. It should only be set when restarting
|
||
|
// raft. raft will not return entries to the application smaller or equal to
|
||
|
// Applied. If Applied is unset when restarting, raft might return previous
|
||
|
// applied entries. This is a very application dependent configuration.
|
||
|
Applied uint64
|
||
|
|
||
|
// MaxSizePerMsg limits the max byte size of each append message. Smaller
|
||
|
// value lowers the raft recovery cost(initial probing and message lost
|
||
|
// during normal operation). On the other side, it might affect the
|
||
|
// throughput during normal replication. Note: math.MaxUint64 for unlimited,
|
||
|
// 0 for at most one entry per message.
|
||
|
MaxSizePerMsg uint64
|
||
|
// MaxCommittedSizePerReady limits the size of the committed entries which
|
||
|
// can be applied.
|
||
|
MaxCommittedSizePerReady uint64
|
||
|
// MaxUncommittedEntriesSize limits the aggregate byte size of the
|
||
|
// uncommitted entries that may be appended to a leader's log. Once this
|
||
|
// limit is exceeded, proposals will begin to return ErrProposalDropped
|
||
|
// errors. Note: 0 for no limit.
|
||
|
MaxUncommittedEntriesSize uint64
|
||
|
// MaxInflightMsgs limits the max number of in-flight append messages during
|
||
|
// optimistic replication phase. The application transportation layer usually
|
||
|
// has its own sending buffer over TCP/UDP. Setting MaxInflightMsgs to avoid
|
||
|
// overflowing that sending buffer. TODO (xiangli): feedback to application to
|
||
|
// limit the proposal rate?
|
||
|
MaxInflightMsgs int
|
||
|
|
||
|
// CheckQuorum specifies if the leader should check quorum activity. Leader
|
||
|
// steps down when quorum is not active for an electionTimeout.
|
||
|
CheckQuorum bool
|
||
|
|
||
|
// PreVote enables the Pre-Vote algorithm described in raft thesis section
|
||
|
// 9.6. This prevents disruption when a node that has been partitioned away
|
||
|
// rejoins the cluster.
|
||
|
PreVote bool
|
||
|
|
||
|
// ReadOnlyOption specifies how the read only request is processed.
|
||
|
//
|
||
|
// ReadOnlySafe guarantees the linearizability of the read only request by
|
||
|
// communicating with the quorum. It is the default and suggested option.
|
||
|
//
|
||
|
// ReadOnlyLeaseBased ensures linearizability of the read only request by
|
||
|
// relying on the leader lease. It can be affected by clock drift.
|
||
|
// If the clock drift is unbounded, leader might keep the lease longer than it
|
||
|
// should (clock can move backward/pause without any bound). ReadIndex is not safe
|
||
|
// in that case.
|
||
|
// CheckQuorum MUST be enabled if ReadOnlyOption is ReadOnlyLeaseBased.
|
||
|
ReadOnlyOption ReadOnlyOption
|
||
|
|
||
|
// Logger is the logger used for raft log. For multinode which can host
|
||
|
// multiple raft group, each raft group can have its own logger
|
||
|
Logger Logger
|
||
|
|
||
|
// DisableProposalForwarding set to true means that followers will drop
|
||
|
// proposals, rather than forwarding them to the leader. One use case for
|
||
|
// this feature would be in a situation where the Raft leader is used to
|
||
|
// compute the data of a proposal, for example, adding a timestamp from a
|
||
|
// hybrid logical clock to data in a monotonically increasing way. Forwarding
|
||
|
// should be disabled to prevent a follower with an inaccurate hybrid
|
||
|
// logical clock from assigning the timestamp and then forwarding the data
|
||
|
// to the leader.
|
||
|
DisableProposalForwarding bool
|
||
|
}
|
||
|
|
||
|
func (c *Config) validate() error {
|
||
|
if c.ID == None {
|
||
|
return errors.New("cannot use none as id")
|
||
|
}
|
||
|
|
||
|
if c.HeartbeatTick <= 0 {
|
||
|
return errors.New("heartbeat tick must be greater than 0")
|
||
|
}
|
||
|
|
||
|
if c.ElectionTick <= c.HeartbeatTick {
|
||
|
return errors.New("election tick must be greater than heartbeat tick")
|
||
|
}
|
||
|
|
||
|
if c.Storage == nil {
|
||
|
return errors.New("storage cannot be nil")
|
||
|
}
|
||
|
|
||
|
if c.MaxUncommittedEntriesSize == 0 {
|
||
|
c.MaxUncommittedEntriesSize = noLimit
|
||
|
}
|
||
|
|
||
|
// default MaxCommittedSizePerReady to MaxSizePerMsg because they were
|
||
|
// previously the same parameter.
|
||
|
if c.MaxCommittedSizePerReady == 0 {
|
||
|
c.MaxCommittedSizePerReady = c.MaxSizePerMsg
|
||
|
}
|
||
|
|
||
|
if c.MaxInflightMsgs <= 0 {
|
||
|
return errors.New("max inflight messages must be greater than 0")
|
||
|
}
|
||
|
|
||
|
if c.Logger == nil {
|
||
|
c.Logger = raftLogger
|
||
|
}
|
||
|
|
||
|
if c.ReadOnlyOption == ReadOnlyLeaseBased && !c.CheckQuorum {
|
||
|
return errors.New("CheckQuorum must be enabled when ReadOnlyOption is ReadOnlyLeaseBased")
|
||
|
}
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
type raft struct {
|
||
|
id uint64
|
||
|
|
||
|
Term uint64
|
||
|
Vote uint64
|
||
|
|
||
|
readStates []ReadState
|
||
|
|
||
|
// the log
|
||
|
raftLog *raftLog
|
||
|
|
||
|
maxMsgSize uint64
|
||
|
maxUncommittedSize uint64
|
||
|
// TODO(tbg): rename to trk.
|
||
|
prs tracker.ProgressTracker
|
||
|
|
||
|
state StateType
|
||
|
|
||
|
// isLearner is true if the local raft node is a learner.
|
||
|
isLearner bool
|
||
|
|
||
|
msgs []pb.Message
|
||
|
|
||
|
// the leader id
|
||
|
lead uint64
|
||
|
// leadTransferee is id of the leader transfer target when its value is not zero.
|
||
|
// Follow the procedure defined in raft thesis 3.10.
|
||
|
leadTransferee uint64
|
||
|
// Only one conf change may be pending (in the log, but not yet
|
||
|
// applied) at a time. This is enforced via pendingConfIndex, which
|
||
|
// is set to a value >= the log index of the latest pending
|
||
|
// configuration change (if any). Config changes are only allowed to
|
||
|
// be proposed if the leader's applied index is greater than this
|
||
|
// value.
|
||
|
pendingConfIndex uint64
|
||
|
// an estimate of the size of the uncommitted tail of the Raft log. Used to
|
||
|
// prevent unbounded log growth. Only maintained by the leader. Reset on
|
||
|
// term changes.
|
||
|
uncommittedSize uint64
|
||
|
|
||
|
readOnly *readOnly
|
||
|
|
||
|
// number of ticks since it reached last electionTimeout when it is leader
|
||
|
// or candidate.
|
||
|
// number of ticks since it reached last electionTimeout or received a
|
||
|
// valid message from current leader when it is a follower.
|
||
|
electionElapsed int
|
||
|
|
||
|
// number of ticks since it reached last heartbeatTimeout.
|
||
|
// only leader keeps heartbeatElapsed.
|
||
|
heartbeatElapsed int
|
||
|
|
||
|
checkQuorum bool
|
||
|
preVote bool
|
||
|
|
||
|
heartbeatTimeout int
|
||
|
electionTimeout int
|
||
|
// randomizedElectionTimeout is a random number between
|
||
|
// [electiontimeout, 2 * electiontimeout - 1]. It gets reset
|
||
|
// when raft changes its state to follower or candidate.
|
||
|
randomizedElectionTimeout int
|
||
|
disableProposalForwarding bool
|
||
|
|
||
|
tick func()
|
||
|
step stepFunc
|
||
|
|
||
|
logger Logger
|
||
|
}
|
||
|
|
||
|
func newRaft(c *Config) *raft {
|
||
|
if err := c.validate(); err != nil {
|
||
|
panic(err.Error())
|
||
|
}
|
||
|
raftlog := newLogWithSize(c.Storage, c.Logger, c.MaxCommittedSizePerReady)
|
||
|
hs, cs, err := c.Storage.InitialState()
|
||
|
if err != nil {
|
||
|
panic(err) // TODO(bdarnell)
|
||
|
}
|
||
|
|
||
|
if len(c.peers) > 0 || len(c.learners) > 0 {
|
||
|
if len(cs.Voters) > 0 || len(cs.Learners) > 0 {
|
||
|
// TODO(bdarnell): the peers argument is always nil except in
|
||
|
// tests; the argument should be removed and these tests should be
|
||
|
// updated to specify their nodes through a snapshot.
|
||
|
panic("cannot specify both newRaft(peers, learners) and ConfState.(Voters, Learners)")
|
||
|
}
|
||
|
cs.Voters = c.peers
|
||
|
cs.Learners = c.learners
|
||
|
}
|
||
|
|
||
|
r := &raft{
|
||
|
id: c.ID,
|
||
|
lead: None,
|
||
|
isLearner: false,
|
||
|
raftLog: raftlog,
|
||
|
maxMsgSize: c.MaxSizePerMsg,
|
||
|
maxUncommittedSize: c.MaxUncommittedEntriesSize,
|
||
|
prs: tracker.MakeProgressTracker(c.MaxInflightMsgs),
|
||
|
electionTimeout: c.ElectionTick,
|
||
|
heartbeatTimeout: c.HeartbeatTick,
|
||
|
logger: c.Logger,
|
||
|
checkQuorum: c.CheckQuorum,
|
||
|
preVote: c.PreVote,
|
||
|
readOnly: newReadOnly(c.ReadOnlyOption),
|
||
|
disableProposalForwarding: c.DisableProposalForwarding,
|
||
|
}
|
||
|
|
||
|
cfg, prs, err := confchange.Restore(confchange.Changer{
|
||
|
Tracker: r.prs,
|
||
|
LastIndex: raftlog.lastIndex(),
|
||
|
}, cs)
|
||
|
if err != nil {
|
||
|
panic(err)
|
||
|
}
|
||
|
assertConfStatesEquivalent(r.logger, cs, r.switchToConfig(cfg, prs))
|
||
|
|
||
|
if !IsEmptyHardState(hs) {
|
||
|
r.loadState(hs)
|
||
|
}
|
||
|
if c.Applied > 0 {
|
||
|
raftlog.appliedTo(c.Applied)
|
||
|
}
|
||
|
r.becomeFollower(r.Term, None)
|
||
|
|
||
|
var nodesStrs []string
|
||
|
for _, n := range r.prs.VoterNodes() {
|
||
|
nodesStrs = append(nodesStrs, fmt.Sprintf("%x", n))
|
||
|
}
|
||
|
|
||
|
r.logger.Infof("newRaft %x [peers: [%s], term: %d, commit: %d, applied: %d, lastindex: %d, lastterm: %d]",
|
||
|
r.id, strings.Join(nodesStrs, ","), r.Term, r.raftLog.committed, r.raftLog.applied, r.raftLog.lastIndex(), r.raftLog.lastTerm())
|
||
|
return r
|
||
|
}
|
||
|
|
||
|
func (r *raft) hasLeader() bool { return r.lead != None }
|
||
|
|
||
|
func (r *raft) softState() *SoftState { return &SoftState{Lead: r.lead, RaftState: r.state} }
|
||
|
|
||
|
func (r *raft) hardState() pb.HardState {
|
||
|
return pb.HardState{
|
||
|
Term: r.Term,
|
||
|
Vote: r.Vote,
|
||
|
Commit: r.raftLog.committed,
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// send persists state to stable storage and then sends to its mailbox.
|
||
|
func (r *raft) send(m pb.Message) {
|
||
|
m.From = r.id
|
||
|
if m.Type == pb.MsgVote || m.Type == pb.MsgVoteResp || m.Type == pb.MsgPreVote || m.Type == pb.MsgPreVoteResp {
|
||
|
if m.Term == 0 {
|
||
|
// All {pre-,}campaign messages need to have the term set when
|
||
|
// sending.
|
||
|
// - MsgVote: m.Term is the term the node is campaigning for,
|
||
|
// non-zero as we increment the term when campaigning.
|
||
|
// - MsgVoteResp: m.Term is the new r.Term if the MsgVote was
|
||
|
// granted, non-zero for the same reason MsgVote is
|
||
|
// - MsgPreVote: m.Term is the term the node will campaign,
|
||
|
// non-zero as we use m.Term to indicate the next term we'll be
|
||
|
// campaigning for
|
||
|
// - MsgPreVoteResp: m.Term is the term received in the original
|
||
|
// MsgPreVote if the pre-vote was granted, non-zero for the
|
||
|
// same reasons MsgPreVote is
|
||
|
panic(fmt.Sprintf("term should be set when sending %s", m.Type))
|
||
|
}
|
||
|
} else {
|
||
|
if m.Term != 0 {
|
||
|
panic(fmt.Sprintf("term should not be set when sending %s (was %d)", m.Type, m.Term))
|
||
|
}
|
||
|
// do not attach term to MsgProp, MsgReadIndex
|
||
|
// proposals are a way to forward to the leader and
|
||
|
// should be treated as local message.
|
||
|
// MsgReadIndex is also forwarded to leader.
|
||
|
if m.Type != pb.MsgProp && m.Type != pb.MsgReadIndex {
|
||
|
m.Term = r.Term
|
||
|
}
|
||
|
}
|
||
|
r.msgs = append(r.msgs, m)
|
||
|
}
|
||
|
|
||
|
// sendAppend sends an append RPC with new entries (if any) and the
|
||
|
// current commit index to the given peer.
|
||
|
func (r *raft) sendAppend(to uint64) {
|
||
|
r.maybeSendAppend(to, true)
|
||
|
}
|
||
|
|
||
|
// maybeSendAppend sends an append RPC with new entries to the given peer,
|
||
|
// if necessary. Returns true if a message was sent. The sendIfEmpty
|
||
|
// argument controls whether messages with no entries will be sent
|
||
|
// ("empty" messages are useful to convey updated Commit indexes, but
|
||
|
// are undesirable when we're sending multiple messages in a batch).
|
||
|
func (r *raft) maybeSendAppend(to uint64, sendIfEmpty bool) bool {
|
||
|
pr := r.prs.Progress[to]
|
||
|
if pr.IsPaused() {
|
||
|
return false
|
||
|
}
|
||
|
m := pb.Message{}
|
||
|
m.To = to
|
||
|
|
||
|
term, errt := r.raftLog.term(pr.Next - 1)
|
||
|
ents, erre := r.raftLog.entries(pr.Next, r.maxMsgSize)
|
||
|
if len(ents) == 0 && !sendIfEmpty {
|
||
|
return false
|
||
|
}
|
||
|
|
||
|
if errt != nil || erre != nil { // send snapshot if we failed to get term or entries
|
||
|
if !pr.RecentActive {
|
||
|
r.logger.Debugf("ignore sending snapshot to %x since it is not recently active", to)
|
||
|
return false
|
||
|
}
|
||
|
|
||
|
m.Type = pb.MsgSnap
|
||
|
snapshot, err := r.raftLog.snapshot()
|
||
|
if err != nil {
|
||
|
if err == ErrSnapshotTemporarilyUnavailable {
|
||
|
r.logger.Debugf("%x failed to send snapshot to %x because snapshot is temporarily unavailable", r.id, to)
|
||
|
return false
|
||
|
}
|
||
|
panic(err) // TODO(bdarnell)
|
||
|
}
|
||
|
if IsEmptySnap(snapshot) {
|
||
|
panic("need non-empty snapshot")
|
||
|
}
|
||
|
m.Snapshot = snapshot
|
||
|
sindex, sterm := snapshot.Metadata.Index, snapshot.Metadata.Term
|
||
|
r.logger.Debugf("%x [firstindex: %d, commit: %d] sent snapshot[index: %d, term: %d] to %x [%s]",
|
||
|
r.id, r.raftLog.firstIndex(), r.raftLog.committed, sindex, sterm, to, pr)
|
||
|
pr.BecomeSnapshot(sindex)
|
||
|
r.logger.Debugf("%x paused sending replication messages to %x [%s]", r.id, to, pr)
|
||
|
} else {
|
||
|
m.Type = pb.MsgApp
|
||
|
m.Index = pr.Next - 1
|
||
|
m.LogTerm = term
|
||
|
m.Entries = ents
|
||
|
m.Commit = r.raftLog.committed
|
||
|
if n := len(m.Entries); n != 0 {
|
||
|
switch pr.State {
|
||
|
// optimistically increase the next when in StateReplicate
|
||
|
case tracker.StateReplicate:
|
||
|
last := m.Entries[n-1].Index
|
||
|
pr.OptimisticUpdate(last)
|
||
|
pr.Inflights.Add(last)
|
||
|
case tracker.StateProbe:
|
||
|
pr.ProbeSent = true
|
||
|
default:
|
||
|
r.logger.Panicf("%x is sending append in unhandled state %s", r.id, pr.State)
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
r.send(m)
|
||
|
return true
|
||
|
}
|
||
|
|
||
|
// sendHeartbeat sends a heartbeat RPC to the given peer.
|
||
|
func (r *raft) sendHeartbeat(to uint64, ctx []byte) {
|
||
|
// Attach the commit as min(to.matched, r.committed).
|
||
|
// When the leader sends out heartbeat message,
|
||
|
// the receiver(follower) might not be matched with the leader
|
||
|
// or it might not have all the committed entries.
|
||
|
// The leader MUST NOT forward the follower's commit to
|
||
|
// an unmatched index.
|
||
|
commit := min(r.prs.Progress[to].Match, r.raftLog.committed)
|
||
|
m := pb.Message{
|
||
|
To: to,
|
||
|
Type: pb.MsgHeartbeat,
|
||
|
Commit: commit,
|
||
|
Context: ctx,
|
||
|
}
|
||
|
|
||
|
r.send(m)
|
||
|
}
|
||
|
|
||
|
// bcastAppend sends RPC, with entries to all peers that are not up-to-date
|
||
|
// according to the progress recorded in r.prs.
|
||
|
func (r *raft) bcastAppend() {
|
||
|
r.prs.Visit(func(id uint64, _ *tracker.Progress) {
|
||
|
if id == r.id {
|
||
|
return
|
||
|
}
|
||
|
r.sendAppend(id)
|
||
|
})
|
||
|
}
|
||
|
|
||
|
// bcastHeartbeat sends RPC, without entries to all the peers.
|
||
|
func (r *raft) bcastHeartbeat() {
|
||
|
lastCtx := r.readOnly.lastPendingRequestCtx()
|
||
|
if len(lastCtx) == 0 {
|
||
|
r.bcastHeartbeatWithCtx(nil)
|
||
|
} else {
|
||
|
r.bcastHeartbeatWithCtx([]byte(lastCtx))
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (r *raft) bcastHeartbeatWithCtx(ctx []byte) {
|
||
|
r.prs.Visit(func(id uint64, _ *tracker.Progress) {
|
||
|
if id == r.id {
|
||
|
return
|
||
|
}
|
||
|
r.sendHeartbeat(id, ctx)
|
||
|
})
|
||
|
}
|
||
|
|
||
|
func (r *raft) advance(rd Ready) {
|
||
|
// If entries were applied (or a snapshot), update our cursor for
|
||
|
// the next Ready. Note that if the current HardState contains a
|
||
|
// new Commit index, this does not mean that we're also applying
|
||
|
// all of the new entries due to commit pagination by size.
|
||
|
if index := rd.appliedCursor(); index > 0 {
|
||
|
r.raftLog.appliedTo(index)
|
||
|
if r.prs.Config.AutoLeave && index >= r.pendingConfIndex && r.state == StateLeader {
|
||
|
// If the current (and most recent, at least for this leader's term)
|
||
|
// configuration should be auto-left, initiate that now.
|
||
|
ccdata, err := (&pb.ConfChangeV2{}).Marshal()
|
||
|
if err != nil {
|
||
|
panic(err)
|
||
|
}
|
||
|
ent := pb.Entry{
|
||
|
Type: pb.EntryConfChangeV2,
|
||
|
Data: ccdata,
|
||
|
}
|
||
|
if !r.appendEntry(ent) {
|
||
|
// If we could not append the entry, bump the pending conf index
|
||
|
// so that we'll try again later.
|
||
|
//
|
||
|
// TODO(tbg): test this case.
|
||
|
r.pendingConfIndex = r.raftLog.lastIndex()
|
||
|
} else {
|
||
|
r.logger.Infof("initiating automatic transition out of joint configuration %s", r.prs.Config)
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
r.reduceUncommittedSize(rd.CommittedEntries)
|
||
|
|
||
|
if len(rd.Entries) > 0 {
|
||
|
e := rd.Entries[len(rd.Entries)-1]
|
||
|
r.raftLog.stableTo(e.Index, e.Term)
|
||
|
}
|
||
|
if !IsEmptySnap(rd.Snapshot) {
|
||
|
r.raftLog.stableSnapTo(rd.Snapshot.Metadata.Index)
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// maybeCommit attempts to advance the commit index. Returns true if
|
||
|
// the commit index changed (in which case the caller should call
|
||
|
// r.bcastAppend).
|
||
|
func (r *raft) maybeCommit() bool {
|
||
|
mci := r.prs.Committed()
|
||
|
return r.raftLog.maybeCommit(mci, r.Term)
|
||
|
}
|
||
|
|
||
|
func (r *raft) reset(term uint64) {
|
||
|
if r.Term != term {
|
||
|
r.Term = term
|
||
|
r.Vote = None
|
||
|
}
|
||
|
r.lead = None
|
||
|
|
||
|
r.electionElapsed = 0
|
||
|
r.heartbeatElapsed = 0
|
||
|
r.resetRandomizedElectionTimeout()
|
||
|
|
||
|
r.abortLeaderTransfer()
|
||
|
|
||
|
r.prs.ResetVotes()
|
||
|
r.prs.Visit(func(id uint64, pr *tracker.Progress) {
|
||
|
*pr = tracker.Progress{
|
||
|
Match: 0,
|
||
|
Next: r.raftLog.lastIndex() + 1,
|
||
|
Inflights: tracker.NewInflights(r.prs.MaxInflight),
|
||
|
IsLearner: pr.IsLearner,
|
||
|
}
|
||
|
if id == r.id {
|
||
|
pr.Match = r.raftLog.lastIndex()
|
||
|
}
|
||
|
})
|
||
|
|
||
|
r.pendingConfIndex = 0
|
||
|
r.uncommittedSize = 0
|
||
|
r.readOnly = newReadOnly(r.readOnly.option)
|
||
|
}
|
||
|
|
||
|
func (r *raft) appendEntry(es ...pb.Entry) (accepted bool) {
|
||
|
li := r.raftLog.lastIndex()
|
||
|
for i := range es {
|
||
|
es[i].Term = r.Term
|
||
|
es[i].Index = li + 1 + uint64(i)
|
||
|
}
|
||
|
// Track the size of this uncommitted proposal.
|
||
|
if !r.increaseUncommittedSize(es) {
|
||
|
r.logger.Debugf(
|
||
|
"%x appending new entries to log would exceed uncommitted entry size limit; dropping proposal",
|
||
|
r.id,
|
||
|
)
|
||
|
// Drop the proposal.
|
||
|
return false
|
||
|
}
|
||
|
// use latest "last" index after truncate/append
|
||
|
li = r.raftLog.append(es...)
|
||
|
r.prs.Progress[r.id].MaybeUpdate(li)
|
||
|
// Regardless of maybeCommit's return, our caller will call bcastAppend.
|
||
|
r.maybeCommit()
|
||
|
return true
|
||
|
}
|
||
|
|
||
|
// tickElection is run by followers and candidates after r.electionTimeout.
|
||
|
func (r *raft) tickElection() {
|
||
|
r.electionElapsed++
|
||
|
|
||
|
if r.promotable() && r.pastElectionTimeout() {
|
||
|
r.electionElapsed = 0
|
||
|
r.Step(pb.Message{From: r.id, Type: pb.MsgHup})
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// tickHeartbeat is run by leaders to send a MsgBeat after r.heartbeatTimeout.
|
||
|
func (r *raft) tickHeartbeat() {
|
||
|
r.heartbeatElapsed++
|
||
|
r.electionElapsed++
|
||
|
|
||
|
if r.electionElapsed >= r.electionTimeout {
|
||
|
r.electionElapsed = 0
|
||
|
if r.checkQuorum {
|
||
|
r.Step(pb.Message{From: r.id, Type: pb.MsgCheckQuorum})
|
||
|
}
|
||
|
// If current leader cannot transfer leadership in electionTimeout, it becomes leader again.
|
||
|
if r.state == StateLeader && r.leadTransferee != None {
|
||
|
r.abortLeaderTransfer()
|
||
|
}
|
||
|
}
|
||
|
|
||
|
if r.state != StateLeader {
|
||
|
return
|
||
|
}
|
||
|
|
||
|
if r.heartbeatElapsed >= r.heartbeatTimeout {
|
||
|
r.heartbeatElapsed = 0
|
||
|
r.Step(pb.Message{From: r.id, Type: pb.MsgBeat})
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (r *raft) becomeFollower(term uint64, lead uint64) {
|
||
|
r.step = stepFollower
|
||
|
r.reset(term)
|
||
|
r.tick = r.tickElection
|
||
|
r.lead = lead
|
||
|
r.state = StateFollower
|
||
|
r.logger.Infof("%x became follower at term %d", r.id, r.Term)
|
||
|
}
|
||
|
|
||
|
func (r *raft) becomeCandidate() {
|
||
|
// TODO(xiangli) remove the panic when the raft implementation is stable
|
||
|
if r.state == StateLeader {
|
||
|
panic("invalid transition [leader -> candidate]")
|
||
|
}
|
||
|
r.step = stepCandidate
|
||
|
r.reset(r.Term + 1)
|
||
|
r.tick = r.tickElection
|
||
|
r.Vote = r.id
|
||
|
r.state = StateCandidate
|
||
|
r.logger.Infof("%x became candidate at term %d", r.id, r.Term)
|
||
|
}
|
||
|
|
||
|
func (r *raft) becomePreCandidate() {
|
||
|
// TODO(xiangli) remove the panic when the raft implementation is stable
|
||
|
if r.state == StateLeader {
|
||
|
panic("invalid transition [leader -> pre-candidate]")
|
||
|
}
|
||
|
// Becoming a pre-candidate changes our step functions and state,
|
||
|
// but doesn't change anything else. In particular it does not increase
|
||
|
// r.Term or change r.Vote.
|
||
|
r.step = stepCandidate
|
||
|
r.prs.ResetVotes()
|
||
|
r.tick = r.tickElection
|
||
|
r.lead = None
|
||
|
r.state = StatePreCandidate
|
||
|
r.logger.Infof("%x became pre-candidate at term %d", r.id, r.Term)
|
||
|
}
|
||
|
|
||
|
func (r *raft) becomeLeader() {
|
||
|
// TODO(xiangli) remove the panic when the raft implementation is stable
|
||
|
if r.state == StateFollower {
|
||
|
panic("invalid transition [follower -> leader]")
|
||
|
}
|
||
|
r.step = stepLeader
|
||
|
r.reset(r.Term)
|
||
|
r.tick = r.tickHeartbeat
|
||
|
r.lead = r.id
|
||
|
r.state = StateLeader
|
||
|
// Followers enter replicate mode when they've been successfully probed
|
||
|
// (perhaps after having received a snapshot as a result). The leader is
|
||
|
// trivially in this state. Note that r.reset() has initialized this
|
||
|
// progress with the last index already.
|
||
|
r.prs.Progress[r.id].BecomeReplicate()
|
||
|
|
||
|
// Conservatively set the pendingConfIndex to the last index in the
|
||
|
// log. There may or may not be a pending config change, but it's
|
||
|
// safe to delay any future proposals until we commit all our
|
||
|
// pending log entries, and scanning the entire tail of the log
|
||
|
// could be expensive.
|
||
|
r.pendingConfIndex = r.raftLog.lastIndex()
|
||
|
|
||
|
emptyEnt := pb.Entry{Data: nil}
|
||
|
if !r.appendEntry(emptyEnt) {
|
||
|
// This won't happen because we just called reset() above.
|
||
|
r.logger.Panic("empty entry was dropped")
|
||
|
}
|
||
|
// As a special case, don't count the initial empty entry towards the
|
||
|
// uncommitted log quota. This is because we want to preserve the
|
||
|
// behavior of allowing one entry larger than quota if the current
|
||
|
// usage is zero.
|
||
|
r.reduceUncommittedSize([]pb.Entry{emptyEnt})
|
||
|
r.logger.Infof("%x became leader at term %d", r.id, r.Term)
|
||
|
}
|
||
|
|
||
|
// campaign transitions the raft instance to candidate state. This must only be
|
||
|
// called after verifying that this is a legitimate transition.
|
||
|
func (r *raft) campaign(t CampaignType) {
|
||
|
if !r.promotable() {
|
||
|
// This path should not be hit (callers are supposed to check), but
|
||
|
// better safe than sorry.
|
||
|
r.logger.Warningf("%x is unpromotable; campaign() should have been called", r.id)
|
||
|
}
|
||
|
var term uint64
|
||
|
var voteMsg pb.MessageType
|
||
|
if t == campaignPreElection {
|
||
|
r.becomePreCandidate()
|
||
|
voteMsg = pb.MsgPreVote
|
||
|
// PreVote RPCs are sent for the next term before we've incremented r.Term.
|
||
|
term = r.Term + 1
|
||
|
} else {
|
||
|
r.becomeCandidate()
|
||
|
voteMsg = pb.MsgVote
|
||
|
term = r.Term
|
||
|
}
|
||
|
if _, _, res := r.poll(r.id, voteRespMsgType(voteMsg), true); res == quorum.VoteWon {
|
||
|
// We won the election after voting for ourselves (which must mean that
|
||
|
// this is a single-node cluster). Advance to the next state.
|
||
|
if t == campaignPreElection {
|
||
|
r.campaign(campaignElection)
|
||
|
} else {
|
||
|
r.becomeLeader()
|
||
|
}
|
||
|
return
|
||
|
}
|
||
|
var ids []uint64
|
||
|
{
|
||
|
idMap := r.prs.Voters.IDs()
|
||
|
ids = make([]uint64, 0, len(idMap))
|
||
|
for id := range idMap {
|
||
|
ids = append(ids, id)
|
||
|
}
|
||
|
sort.Slice(ids, func(i, j int) bool { return ids[i] < ids[j] })
|
||
|
}
|
||
|
for _, id := range ids {
|
||
|
if id == r.id {
|
||
|
continue
|
||
|
}
|
||
|
r.logger.Infof("%x [logterm: %d, index: %d] sent %s request to %x at term %d",
|
||
|
r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), voteMsg, id, r.Term)
|
||
|
|
||
|
var ctx []byte
|
||
|
if t == campaignTransfer {
|
||
|
ctx = []byte(t)
|
||
|
}
|
||
|
r.send(pb.Message{Term: term, To: id, Type: voteMsg, Index: r.raftLog.lastIndex(), LogTerm: r.raftLog.lastTerm(), Context: ctx})
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (r *raft) poll(id uint64, t pb.MessageType, v bool) (granted int, rejected int, result quorum.VoteResult) {
|
||
|
if v {
|
||
|
r.logger.Infof("%x received %s from %x at term %d", r.id, t, id, r.Term)
|
||
|
} else {
|
||
|
r.logger.Infof("%x received %s rejection from %x at term %d", r.id, t, id, r.Term)
|
||
|
}
|
||
|
r.prs.RecordVote(id, v)
|
||
|
return r.prs.TallyVotes()
|
||
|
}
|
||
|
|
||
|
func (r *raft) Step(m pb.Message) error {
|
||
|
// Handle the message term, which may result in our stepping down to a follower.
|
||
|
switch {
|
||
|
case m.Term == 0:
|
||
|
// local message
|
||
|
case m.Term > r.Term:
|
||
|
if m.Type == pb.MsgVote || m.Type == pb.MsgPreVote {
|
||
|
force := bytes.Equal(m.Context, []byte(campaignTransfer))
|
||
|
inLease := r.checkQuorum && r.lead != None && r.electionElapsed < r.electionTimeout
|
||
|
if !force && inLease {
|
||
|
// If a server receives a RequestVote request within the minimum election timeout
|
||
|
// of hearing from a current leader, it does not update its term or grant its vote
|
||
|
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)",
|
||
|
r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term, r.electionTimeout-r.electionElapsed)
|
||
|
return nil
|
||
|
}
|
||
|
}
|
||
|
switch {
|
||
|
case m.Type == pb.MsgPreVote:
|
||
|
// Never change our term in response to a PreVote
|
||
|
case m.Type == pb.MsgPreVoteResp && !m.Reject:
|
||
|
// We send pre-vote requests with a term in our future. If the
|
||
|
// pre-vote is granted, we will increment our term when we get a
|
||
|
// quorum. If it is not, the term comes from the node that
|
||
|
// rejected our vote so we should become a follower at the new
|
||
|
// term.
|
||
|
default:
|
||
|
r.logger.Infof("%x [term: %d] received a %s message with higher term from %x [term: %d]",
|
||
|
r.id, r.Term, m.Type, m.From, m.Term)
|
||
|
if m.Type == pb.MsgApp || m.Type == pb.MsgHeartbeat || m.Type == pb.MsgSnap {
|
||
|
r.becomeFollower(m.Term, m.From)
|
||
|
} else {
|
||
|
r.becomeFollower(m.Term, None)
|
||
|
}
|
||
|
}
|
||
|
|
||
|
case m.Term < r.Term:
|
||
|
if (r.checkQuorum || r.preVote) && (m.Type == pb.MsgHeartbeat || m.Type == pb.MsgApp) {
|
||
|
// We have received messages from a leader at a lower term. It is possible
|
||
|
// that these messages were simply delayed in the network, but this could
|
||
|
// also mean that this node has advanced its term number during a network
|
||
|
// partition, and it is now unable to either win an election or to rejoin
|
||
|
// the majority on the old term. If checkQuorum is false, this will be
|
||
|
// handled by incrementing term numbers in response to MsgVote with a
|
||
|
// higher term, but if checkQuorum is true we may not advance the term on
|
||
|
// MsgVote and must generate other messages to advance the term. The net
|
||
|
// result of these two features is to minimize the disruption caused by
|
||
|
// nodes that have been removed from the cluster's configuration: a
|
||
|
// removed node will send MsgVotes (or MsgPreVotes) which will be ignored,
|
||
|
// but it will not receive MsgApp or MsgHeartbeat, so it will not create
|
||
|
// disruptive term increases, by notifying leader of this node's activeness.
|
||
|
// The above comments also true for Pre-Vote
|
||
|
//
|
||
|
// When follower gets isolated, it soon starts an election ending
|
||
|
// up with a higher term than leader, although it won't receive enough
|
||
|
// votes to win the election. When it regains connectivity, this response
|
||
|
// with "pb.MsgAppResp" of higher term would force leader to step down.
|
||
|
// However, this disruption is inevitable to free this stuck node with
|
||
|
// fresh election. This can be prevented with Pre-Vote phase.
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp})
|
||
|
} else if m.Type == pb.MsgPreVote {
|
||
|
// Before Pre-Vote enable, there may have candidate with higher term,
|
||
|
// but less log. After update to Pre-Vote, the cluster may deadlock if
|
||
|
// we drop messages with a lower term.
|
||
|
r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected %s from %x [logterm: %d, index: %d] at term %d",
|
||
|
r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term)
|
||
|
r.send(pb.Message{To: m.From, Term: r.Term, Type: pb.MsgPreVoteResp, Reject: true})
|
||
|
} else {
|
||
|
// ignore other cases
|
||
|
r.logger.Infof("%x [term: %d] ignored a %s message with lower term from %x [term: %d]",
|
||
|
r.id, r.Term, m.Type, m.From, m.Term)
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
switch m.Type {
|
||
|
case pb.MsgHup:
|
||
|
if r.state != StateLeader {
|
||
|
if !r.promotable() {
|
||
|
r.logger.Warningf("%x is unpromotable and can not campaign; ignoring MsgHup", r.id)
|
||
|
return nil
|
||
|
}
|
||
|
ents, err := r.raftLog.slice(r.raftLog.applied+1, r.raftLog.committed+1, noLimit)
|
||
|
if err != nil {
|
||
|
r.logger.Panicf("unexpected error getting unapplied entries (%v)", err)
|
||
|
}
|
||
|
if n := numOfPendingConf(ents); n != 0 && r.raftLog.committed > r.raftLog.applied {
|
||
|
r.logger.Warningf("%x cannot campaign at term %d since there are still %d pending configuration changes to apply", r.id, r.Term, n)
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
r.logger.Infof("%x is starting a new election at term %d", r.id, r.Term)
|
||
|
if r.preVote {
|
||
|
r.campaign(campaignPreElection)
|
||
|
} else {
|
||
|
r.campaign(campaignElection)
|
||
|
}
|
||
|
} else {
|
||
|
r.logger.Debugf("%x ignoring MsgHup because already leader", r.id)
|
||
|
}
|
||
|
|
||
|
case pb.MsgVote, pb.MsgPreVote:
|
||
|
// We can vote if this is a repeat of a vote we've already cast...
|
||
|
canVote := r.Vote == m.From ||
|
||
|
// ...we haven't voted and we don't think there's a leader yet in this term...
|
||
|
(r.Vote == None && r.lead == None) ||
|
||
|
// ...or this is a PreVote for a future term...
|
||
|
(m.Type == pb.MsgPreVote && m.Term > r.Term)
|
||
|
// ...and we believe the candidate is up to date.
|
||
|
if canVote && r.raftLog.isUpToDate(m.Index, m.LogTerm) {
|
||
|
// Note: it turns out that that learners must be allowed to cast votes.
|
||
|
// This seems counter- intuitive but is necessary in the situation in which
|
||
|
// a learner has been promoted (i.e. is now a voter) but has not learned
|
||
|
// about this yet.
|
||
|
// For example, consider a group in which id=1 is a learner and id=2 and
|
||
|
// id=3 are voters. A configuration change promoting 1 can be committed on
|
||
|
// the quorum `{2,3}` without the config change being appended to the
|
||
|
// learner's log. If the leader (say 2) fails, there are de facto two
|
||
|
// voters remaining. Only 3 can win an election (due to its log containing
|
||
|
// all committed entries), but to do so it will need 1 to vote. But 1
|
||
|
// considers itself a learner and will continue to do so until 3 has
|
||
|
// stepped up as leader, replicates the conf change to 1, and 1 applies it.
|
||
|
// Ultimately, by receiving a request to vote, the learner realizes that
|
||
|
// the candidate believes it to be a voter, and that it should act
|
||
|
// accordingly. The candidate's config may be stale, too; but in that case
|
||
|
// it won't win the election, at least in the absence of the bug discussed
|
||
|
// in:
|
||
|
// https://github.com/etcd-io/etcd/issues/7625#issuecomment-488798263.
|
||
|
r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] cast %s for %x [logterm: %d, index: %d] at term %d",
|
||
|
r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term)
|
||
|
// When responding to Msg{Pre,}Vote messages we include the term
|
||
|
// from the message, not the local term. To see why, consider the
|
||
|
// case where a single node was previously partitioned away and
|
||
|
// it's local term is now out of date. If we include the local term
|
||
|
// (recall that for pre-votes we don't update the local term), the
|
||
|
// (pre-)campaigning node on the other end will proceed to ignore
|
||
|
// the message (it ignores all out of date messages).
|
||
|
// The term in the original message and current local term are the
|
||
|
// same in the case of regular votes, but different for pre-votes.
|
||
|
r.send(pb.Message{To: m.From, Term: m.Term, Type: voteRespMsgType(m.Type)})
|
||
|
if m.Type == pb.MsgVote {
|
||
|
// Only record real votes.
|
||
|
r.electionElapsed = 0
|
||
|
r.Vote = m.From
|
||
|
}
|
||
|
} else {
|
||
|
r.logger.Infof("%x [logterm: %d, index: %d, vote: %x] rejected %s from %x [logterm: %d, index: %d] at term %d",
|
||
|
r.id, r.raftLog.lastTerm(), r.raftLog.lastIndex(), r.Vote, m.Type, m.From, m.LogTerm, m.Index, r.Term)
|
||
|
r.send(pb.Message{To: m.From, Term: r.Term, Type: voteRespMsgType(m.Type), Reject: true})
|
||
|
}
|
||
|
|
||
|
default:
|
||
|
err := r.step(r, m)
|
||
|
if err != nil {
|
||
|
return err
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
type stepFunc func(r *raft, m pb.Message) error
|
||
|
|
||
|
func stepLeader(r *raft, m pb.Message) error {
|
||
|
// These message types do not require any progress for m.From.
|
||
|
switch m.Type {
|
||
|
case pb.MsgBeat:
|
||
|
r.bcastHeartbeat()
|
||
|
return nil
|
||
|
case pb.MsgCheckQuorum:
|
||
|
// The leader should always see itself as active. As a precaution, handle
|
||
|
// the case in which the leader isn't in the configuration any more (for
|
||
|
// example if it just removed itself).
|
||
|
//
|
||
|
// TODO(tbg): I added a TODO in removeNode, it doesn't seem that the
|
||
|
// leader steps down when removing itself. I might be missing something.
|
||
|
if pr := r.prs.Progress[r.id]; pr != nil {
|
||
|
pr.RecentActive = true
|
||
|
}
|
||
|
if !r.prs.QuorumActive() {
|
||
|
r.logger.Warningf("%x stepped down to follower since quorum is not active", r.id)
|
||
|
r.becomeFollower(r.Term, None)
|
||
|
}
|
||
|
// Mark everyone (but ourselves) as inactive in preparation for the next
|
||
|
// CheckQuorum.
|
||
|
r.prs.Visit(func(id uint64, pr *tracker.Progress) {
|
||
|
if id != r.id {
|
||
|
pr.RecentActive = false
|
||
|
}
|
||
|
})
|
||
|
return nil
|
||
|
case pb.MsgProp:
|
||
|
if len(m.Entries) == 0 {
|
||
|
r.logger.Panicf("%x stepped empty MsgProp", r.id)
|
||
|
}
|
||
|
if r.prs.Progress[r.id] == nil {
|
||
|
// If we are not currently a member of the range (i.e. this node
|
||
|
// was removed from the configuration while serving as leader),
|
||
|
// drop any new proposals.
|
||
|
return ErrProposalDropped
|
||
|
}
|
||
|
if r.leadTransferee != None {
|
||
|
r.logger.Debugf("%x [term %d] transfer leadership to %x is in progress; dropping proposal", r.id, r.Term, r.leadTransferee)
|
||
|
return ErrProposalDropped
|
||
|
}
|
||
|
|
||
|
for i := range m.Entries {
|
||
|
e := &m.Entries[i]
|
||
|
var cc pb.ConfChangeI
|
||
|
if e.Type == pb.EntryConfChange {
|
||
|
var ccc pb.ConfChange
|
||
|
if err := ccc.Unmarshal(e.Data); err != nil {
|
||
|
panic(err)
|
||
|
}
|
||
|
cc = ccc
|
||
|
} else if e.Type == pb.EntryConfChangeV2 {
|
||
|
var ccc pb.ConfChangeV2
|
||
|
if err := ccc.Unmarshal(e.Data); err != nil {
|
||
|
panic(err)
|
||
|
}
|
||
|
cc = ccc
|
||
|
}
|
||
|
if cc != nil {
|
||
|
alreadyPending := r.pendingConfIndex > r.raftLog.applied
|
||
|
alreadyJoint := len(r.prs.Config.Voters[1]) > 0
|
||
|
wantsLeaveJoint := len(cc.AsV2().Changes) == 0
|
||
|
|
||
|
var refused string
|
||
|
if alreadyPending {
|
||
|
refused = fmt.Sprintf("possible unapplied conf change at index %d (applied to %d)", r.pendingConfIndex, r.raftLog.applied)
|
||
|
} else if alreadyJoint && !wantsLeaveJoint {
|
||
|
refused = "must transition out of joint config first"
|
||
|
} else if !alreadyJoint && wantsLeaveJoint {
|
||
|
refused = "not in joint state; refusing empty conf change"
|
||
|
}
|
||
|
|
||
|
if refused != "" {
|
||
|
r.logger.Infof("%x ignoring conf change %v at config %s: %s", r.id, cc, r.prs.Config, refused)
|
||
|
m.Entries[i] = pb.Entry{Type: pb.EntryNormal}
|
||
|
} else {
|
||
|
r.pendingConfIndex = r.raftLog.lastIndex() + uint64(i) + 1
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
|
||
|
if !r.appendEntry(m.Entries...) {
|
||
|
return ErrProposalDropped
|
||
|
}
|
||
|
r.bcastAppend()
|
||
|
return nil
|
||
|
case pb.MsgReadIndex:
|
||
|
// If more than the local vote is needed, go through a full broadcast,
|
||
|
// otherwise optimize.
|
||
|
if !r.prs.IsSingleton() {
|
||
|
if r.raftLog.zeroTermOnErrCompacted(r.raftLog.term(r.raftLog.committed)) != r.Term {
|
||
|
// Reject read only request when this leader has not committed any log entry at its term.
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// thinking: use an interally defined context instead of the user given context.
|
||
|
// We can express this in terms of the term and index instead of a user-supplied value.
|
||
|
// This would allow multiple reads to piggyback on the same message.
|
||
|
switch r.readOnly.option {
|
||
|
case ReadOnlySafe:
|
||
|
r.readOnly.addRequest(r.raftLog.committed, m)
|
||
|
// The local node automatically acks the request.
|
||
|
r.readOnly.recvAck(r.id, m.Entries[0].Data)
|
||
|
r.bcastHeartbeatWithCtx(m.Entries[0].Data)
|
||
|
case ReadOnlyLeaseBased:
|
||
|
ri := r.raftLog.committed
|
||
|
if m.From == None || m.From == r.id { // from local member
|
||
|
r.readStates = append(r.readStates, ReadState{Index: ri, RequestCtx: m.Entries[0].Data})
|
||
|
} else {
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgReadIndexResp, Index: ri, Entries: m.Entries})
|
||
|
}
|
||
|
}
|
||
|
} else { // only one voting member (the leader) in the cluster
|
||
|
if m.From == None || m.From == r.id { // from leader itself
|
||
|
r.readStates = append(r.readStates, ReadState{Index: r.raftLog.committed, RequestCtx: m.Entries[0].Data})
|
||
|
} else { // from learner member
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgReadIndexResp, Index: r.raftLog.committed, Entries: m.Entries})
|
||
|
}
|
||
|
}
|
||
|
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// All other message types require a progress for m.From (pr).
|
||
|
pr := r.prs.Progress[m.From]
|
||
|
if pr == nil {
|
||
|
r.logger.Debugf("%x no progress available for %x", r.id, m.From)
|
||
|
return nil
|
||
|
}
|
||
|
switch m.Type {
|
||
|
case pb.MsgAppResp:
|
||
|
pr.RecentActive = true
|
||
|
|
||
|
if m.Reject {
|
||
|
r.logger.Debugf("%x received MsgAppResp(MsgApp was rejected, lastindex: %d) from %x for index %d",
|
||
|
r.id, m.RejectHint, m.From, m.Index)
|
||
|
if pr.MaybeDecrTo(m.Index, m.RejectHint) {
|
||
|
r.logger.Debugf("%x decreased progress of %x to [%s]", r.id, m.From, pr)
|
||
|
if pr.State == tracker.StateReplicate {
|
||
|
pr.BecomeProbe()
|
||
|
}
|
||
|
r.sendAppend(m.From)
|
||
|
}
|
||
|
} else {
|
||
|
oldPaused := pr.IsPaused()
|
||
|
if pr.MaybeUpdate(m.Index) {
|
||
|
switch {
|
||
|
case pr.State == tracker.StateProbe:
|
||
|
pr.BecomeReplicate()
|
||
|
case pr.State == tracker.StateSnapshot && pr.Match >= pr.PendingSnapshot:
|
||
|
// TODO(tbg): we should also enter this branch if a snapshot is
|
||
|
// received that is below pr.PendingSnapshot but which makes it
|
||
|
// possible to use the log again.
|
||
|
r.logger.Debugf("%x recovered from needing snapshot, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
|
||
|
// Transition back to replicating state via probing state
|
||
|
// (which takes the snapshot into account). If we didn't
|
||
|
// move to replicating state, that would only happen with
|
||
|
// the next round of appends (but there may not be a next
|
||
|
// round for a while, exposing an inconsistent RaftStatus).
|
||
|
pr.BecomeProbe()
|
||
|
pr.BecomeReplicate()
|
||
|
case pr.State == tracker.StateReplicate:
|
||
|
pr.Inflights.FreeLE(m.Index)
|
||
|
}
|
||
|
|
||
|
if r.maybeCommit() {
|
||
|
r.bcastAppend()
|
||
|
} else if oldPaused {
|
||
|
// If we were paused before, this node may be missing the
|
||
|
// latest commit index, so send it.
|
||
|
r.sendAppend(m.From)
|
||
|
}
|
||
|
// We've updated flow control information above, which may
|
||
|
// allow us to send multiple (size-limited) in-flight messages
|
||
|
// at once (such as when transitioning from probe to
|
||
|
// replicate, or when freeTo() covers multiple messages). If
|
||
|
// we have more entries to send, send as many messages as we
|
||
|
// can (without sending empty messages for the commit index)
|
||
|
for r.maybeSendAppend(m.From, false) {
|
||
|
}
|
||
|
// Transfer leadership is in progress.
|
||
|
if m.From == r.leadTransferee && pr.Match == r.raftLog.lastIndex() {
|
||
|
r.logger.Infof("%x sent MsgTimeoutNow to %x after received MsgAppResp", r.id, m.From)
|
||
|
r.sendTimeoutNow(m.From)
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
case pb.MsgHeartbeatResp:
|
||
|
pr.RecentActive = true
|
||
|
pr.ProbeSent = false
|
||
|
|
||
|
// free one slot for the full inflights window to allow progress.
|
||
|
if pr.State == tracker.StateReplicate && pr.Inflights.Full() {
|
||
|
pr.Inflights.FreeFirstOne()
|
||
|
}
|
||
|
if pr.Match < r.raftLog.lastIndex() {
|
||
|
r.sendAppend(m.From)
|
||
|
}
|
||
|
|
||
|
if r.readOnly.option != ReadOnlySafe || len(m.Context) == 0 {
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
if r.prs.Voters.VoteResult(r.readOnly.recvAck(m.From, m.Context)) != quorum.VoteWon {
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
rss := r.readOnly.advance(m)
|
||
|
for _, rs := range rss {
|
||
|
req := rs.req
|
||
|
if req.From == None || req.From == r.id { // from local member
|
||
|
r.readStates = append(r.readStates, ReadState{Index: rs.index, RequestCtx: req.Entries[0].Data})
|
||
|
} else {
|
||
|
r.send(pb.Message{To: req.From, Type: pb.MsgReadIndexResp, Index: rs.index, Entries: req.Entries})
|
||
|
}
|
||
|
}
|
||
|
case pb.MsgSnapStatus:
|
||
|
if pr.State != tracker.StateSnapshot {
|
||
|
return nil
|
||
|
}
|
||
|
// TODO(tbg): this code is very similar to the snapshot handling in
|
||
|
// MsgAppResp above. In fact, the code there is more correct than the
|
||
|
// code here and should likely be updated to match (or even better, the
|
||
|
// logic pulled into a newly created Progress state machine handler).
|
||
|
if !m.Reject {
|
||
|
pr.BecomeProbe()
|
||
|
r.logger.Debugf("%x snapshot succeeded, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
|
||
|
} else {
|
||
|
// NB: the order here matters or we'll be probing erroneously from
|
||
|
// the snapshot index, but the snapshot never applied.
|
||
|
pr.PendingSnapshot = 0
|
||
|
pr.BecomeProbe()
|
||
|
r.logger.Debugf("%x snapshot failed, resumed sending replication messages to %x [%s]", r.id, m.From, pr)
|
||
|
}
|
||
|
// If snapshot finish, wait for the MsgAppResp from the remote node before sending
|
||
|
// out the next MsgApp.
|
||
|
// If snapshot failure, wait for a heartbeat interval before next try
|
||
|
pr.ProbeSent = true
|
||
|
case pb.MsgUnreachable:
|
||
|
// During optimistic replication, if the remote becomes unreachable,
|
||
|
// there is huge probability that a MsgApp is lost.
|
||
|
if pr.State == tracker.StateReplicate {
|
||
|
pr.BecomeProbe()
|
||
|
}
|
||
|
r.logger.Debugf("%x failed to send message to %x because it is unreachable [%s]", r.id, m.From, pr)
|
||
|
case pb.MsgTransferLeader:
|
||
|
if pr.IsLearner {
|
||
|
r.logger.Debugf("%x is learner. Ignored transferring leadership", r.id)
|
||
|
return nil
|
||
|
}
|
||
|
leadTransferee := m.From
|
||
|
lastLeadTransferee := r.leadTransferee
|
||
|
if lastLeadTransferee != None {
|
||
|
if lastLeadTransferee == leadTransferee {
|
||
|
r.logger.Infof("%x [term %d] transfer leadership to %x is in progress, ignores request to same node %x",
|
||
|
r.id, r.Term, leadTransferee, leadTransferee)
|
||
|
return nil
|
||
|
}
|
||
|
r.abortLeaderTransfer()
|
||
|
r.logger.Infof("%x [term %d] abort previous transferring leadership to %x", r.id, r.Term, lastLeadTransferee)
|
||
|
}
|
||
|
if leadTransferee == r.id {
|
||
|
r.logger.Debugf("%x is already leader. Ignored transferring leadership to self", r.id)
|
||
|
return nil
|
||
|
}
|
||
|
// Transfer leadership to third party.
|
||
|
r.logger.Infof("%x [term %d] starts to transfer leadership to %x", r.id, r.Term, leadTransferee)
|
||
|
// Transfer leadership should be finished in one electionTimeout, so reset r.electionElapsed.
|
||
|
r.electionElapsed = 0
|
||
|
r.leadTransferee = leadTransferee
|
||
|
if pr.Match == r.raftLog.lastIndex() {
|
||
|
r.sendTimeoutNow(leadTransferee)
|
||
|
r.logger.Infof("%x sends MsgTimeoutNow to %x immediately as %x already has up-to-date log", r.id, leadTransferee, leadTransferee)
|
||
|
} else {
|
||
|
r.sendAppend(leadTransferee)
|
||
|
}
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
// stepCandidate is shared by StateCandidate and StatePreCandidate; the difference is
|
||
|
// whether they respond to MsgVoteResp or MsgPreVoteResp.
|
||
|
func stepCandidate(r *raft, m pb.Message) error {
|
||
|
// Only handle vote responses corresponding to our candidacy (while in
|
||
|
// StateCandidate, we may get stale MsgPreVoteResp messages in this term from
|
||
|
// our pre-candidate state).
|
||
|
var myVoteRespType pb.MessageType
|
||
|
if r.state == StatePreCandidate {
|
||
|
myVoteRespType = pb.MsgPreVoteResp
|
||
|
} else {
|
||
|
myVoteRespType = pb.MsgVoteResp
|
||
|
}
|
||
|
switch m.Type {
|
||
|
case pb.MsgProp:
|
||
|
r.logger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term)
|
||
|
return ErrProposalDropped
|
||
|
case pb.MsgApp:
|
||
|
r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
|
||
|
r.handleAppendEntries(m)
|
||
|
case pb.MsgHeartbeat:
|
||
|
r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
|
||
|
r.handleHeartbeat(m)
|
||
|
case pb.MsgSnap:
|
||
|
r.becomeFollower(m.Term, m.From) // always m.Term == r.Term
|
||
|
r.handleSnapshot(m)
|
||
|
case myVoteRespType:
|
||
|
gr, rj, res := r.poll(m.From, m.Type, !m.Reject)
|
||
|
r.logger.Infof("%x has received %d %s votes and %d vote rejections", r.id, gr, m.Type, rj)
|
||
|
switch res {
|
||
|
case quorum.VoteWon:
|
||
|
if r.state == StatePreCandidate {
|
||
|
r.campaign(campaignElection)
|
||
|
} else {
|
||
|
r.becomeLeader()
|
||
|
r.bcastAppend()
|
||
|
}
|
||
|
case quorum.VoteLost:
|
||
|
// pb.MsgPreVoteResp contains future term of pre-candidate
|
||
|
// m.Term > r.Term; reuse r.Term
|
||
|
r.becomeFollower(r.Term, None)
|
||
|
}
|
||
|
case pb.MsgTimeoutNow:
|
||
|
r.logger.Debugf("%x [term %d state %v] ignored MsgTimeoutNow from %x", r.id, r.Term, r.state, m.From)
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func stepFollower(r *raft, m pb.Message) error {
|
||
|
switch m.Type {
|
||
|
case pb.MsgProp:
|
||
|
if r.lead == None {
|
||
|
r.logger.Infof("%x no leader at term %d; dropping proposal", r.id, r.Term)
|
||
|
return ErrProposalDropped
|
||
|
} else if r.disableProposalForwarding {
|
||
|
r.logger.Infof("%x not forwarding to leader %x at term %d; dropping proposal", r.id, r.lead, r.Term)
|
||
|
return ErrProposalDropped
|
||
|
}
|
||
|
m.To = r.lead
|
||
|
r.send(m)
|
||
|
case pb.MsgApp:
|
||
|
r.electionElapsed = 0
|
||
|
r.lead = m.From
|
||
|
r.handleAppendEntries(m)
|
||
|
case pb.MsgHeartbeat:
|
||
|
r.electionElapsed = 0
|
||
|
r.lead = m.From
|
||
|
r.handleHeartbeat(m)
|
||
|
case pb.MsgSnap:
|
||
|
r.electionElapsed = 0
|
||
|
r.lead = m.From
|
||
|
r.handleSnapshot(m)
|
||
|
case pb.MsgTransferLeader:
|
||
|
if r.lead == None {
|
||
|
r.logger.Infof("%x no leader at term %d; dropping leader transfer msg", r.id, r.Term)
|
||
|
return nil
|
||
|
}
|
||
|
m.To = r.lead
|
||
|
r.send(m)
|
||
|
case pb.MsgTimeoutNow:
|
||
|
if r.promotable() {
|
||
|
r.logger.Infof("%x [term %d] received MsgTimeoutNow from %x and starts an election to get leadership.", r.id, r.Term, m.From)
|
||
|
// Leadership transfers never use pre-vote even if r.preVote is true; we
|
||
|
// know we are not recovering from a partition so there is no need for the
|
||
|
// extra round trip.
|
||
|
r.campaign(campaignTransfer)
|
||
|
} else {
|
||
|
r.logger.Infof("%x received MsgTimeoutNow from %x but is not promotable", r.id, m.From)
|
||
|
}
|
||
|
case pb.MsgReadIndex:
|
||
|
if r.lead == None {
|
||
|
r.logger.Infof("%x no leader at term %d; dropping index reading msg", r.id, r.Term)
|
||
|
return nil
|
||
|
}
|
||
|
m.To = r.lead
|
||
|
r.send(m)
|
||
|
case pb.MsgReadIndexResp:
|
||
|
if len(m.Entries) != 1 {
|
||
|
r.logger.Errorf("%x invalid format of MsgReadIndexResp from %x, entries count: %d", r.id, m.From, len(m.Entries))
|
||
|
return nil
|
||
|
}
|
||
|
r.readStates = append(r.readStates, ReadState{Index: m.Index, RequestCtx: m.Entries[0].Data})
|
||
|
}
|
||
|
return nil
|
||
|
}
|
||
|
|
||
|
func (r *raft) handleAppendEntries(m pb.Message) {
|
||
|
if m.Index < r.raftLog.committed {
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
|
||
|
return
|
||
|
}
|
||
|
|
||
|
if mlastIndex, ok := r.raftLog.maybeAppend(m.Index, m.LogTerm, m.Commit, m.Entries...); ok {
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: mlastIndex})
|
||
|
} else {
|
||
|
r.logger.Debugf("%x [logterm: %d, index: %d] rejected MsgApp [logterm: %d, index: %d] from %x",
|
||
|
r.id, r.raftLog.zeroTermOnErrCompacted(r.raftLog.term(m.Index)), m.Index, m.LogTerm, m.Index, m.From)
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: m.Index, Reject: true, RejectHint: r.raftLog.lastIndex()})
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func (r *raft) handleHeartbeat(m pb.Message) {
|
||
|
r.raftLog.commitTo(m.Commit)
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgHeartbeatResp, Context: m.Context})
|
||
|
}
|
||
|
|
||
|
func (r *raft) handleSnapshot(m pb.Message) {
|
||
|
sindex, sterm := m.Snapshot.Metadata.Index, m.Snapshot.Metadata.Term
|
||
|
if r.restore(m.Snapshot) {
|
||
|
r.logger.Infof("%x [commit: %d] restored snapshot [index: %d, term: %d]",
|
||
|
r.id, r.raftLog.committed, sindex, sterm)
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.lastIndex()})
|
||
|
} else {
|
||
|
r.logger.Infof("%x [commit: %d] ignored snapshot [index: %d, term: %d]",
|
||
|
r.id, r.raftLog.committed, sindex, sterm)
|
||
|
r.send(pb.Message{To: m.From, Type: pb.MsgAppResp, Index: r.raftLog.committed})
|
||
|
}
|
||
|
}
|
||
|
|
||
|
// restore recovers the state machine from a snapshot. It restores the log and the
|
||
|
// configuration of state machine. If this method returns false, the snapshot was
|
||
|
// ignored, either because it was obsolete or because of an error.
|
||
|
func (r *raft) restore(s pb.Snapshot) bool {
|
||
|
if s.Metadata.Index <= r.raftLog.committed {
|
||
|
return false
|
||
|
}
|
||
|
if r.state != StateFollower {
|
||
|
// This is defense-in-depth: if the leader somehow ended up applying a
|
||
|
// snapshot, it could move into a new term without moving into a
|
||
|
// follower state. This should never fire, but if it did, we'd have
|
||
|
// prevented damage by returning early, so log only a loud warning.
|
||
|
//
|
||
|
// At the time of writing, the instance is guaranteed to be in follower
|
||
|
// state when this method is called.
|
||
|
r.logger.Warningf("%x attempted to restore snapshot as leader; should never happen", r.id)
|
||
|
r.becomeFollower(r.Term+1, None)
|
||
|
return false
|
||
|
}
|
||
|
|
||
|
// More defense-in-depth: throw away snapshot if recipient is not in the
|
||
|
// config. This shouuldn't ever happen (at the time of writing) but lots of
|
||
|
// code here and there assumes that r.id is in the progress tracker.
|
||
|
found := false
|
||
|
cs := s.Metadata.ConfState
|
||
|
for _, set := range [][]uint64{
|
||
|
cs.Voters,
|
||
|
cs.Learners,
|
||
|
} {
|
||
|
for _, id := range set {
|
||
|
if id == r.id {
|
||
|
found = true
|
||
|
break
|
||
|
}
|
||
|
}
|
||
|
}
|
||
|
if !found {
|
||
|
r.logger.Warningf(
|
||
|
"%x attempted to restore snapshot but it is not in the ConfState %v; should never happen",
|
||
|
r.id, cs,
|
||
|
)
|
||
|
return false
|
||
|
}
|
||
|
|
||
|
// Now go ahead and actually restore.
|
||
|
|
||
|
if r.raftLog.matchTerm(s.Metadata.Index, s.Metadata.Term) {
|
||
|
r.logger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] fast-forwarded commit to snapshot [index: %d, term: %d]",
|
||
|
r.id, r.raftLog.committed, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term)
|
||
|
r.raftLog.commitTo(s.Metadata.Index)
|
||
|
return false
|
||
|
}
|
||
|
|
||
|
r.raftLog.restore(s)
|
||
|
|
||
|
// Reset the configuration and add the (potentially updated) peers in anew.
|
||
|
r.prs = tracker.MakeProgressTracker(r.prs.MaxInflight)
|
||
|
cfg, prs, err := confchange.Restore(confchange.Changer{
|
||
|
Tracker: r.prs,
|
||
|
LastIndex: r.raftLog.lastIndex(),
|
||
|
}, cs)
|
||
|
|
||
|
if err != nil {
|
||
|
// This should never happen. Either there's a bug in our config change
|
||
|
// handling or the client corrupted the conf change.
|
||
|
panic(fmt.Sprintf("unable to restore config %+v: %s", cs, err))
|
||
|
}
|
||
|
|
||
|
assertConfStatesEquivalent(r.logger, cs, r.switchToConfig(cfg, prs))
|
||
|
|
||
|
pr := r.prs.Progress[r.id]
|
||
|
pr.MaybeUpdate(pr.Next - 1) // TODO(tbg): this is untested and likely unneeded
|
||
|
|
||
|
r.logger.Infof("%x [commit: %d, lastindex: %d, lastterm: %d] restored snapshot [index: %d, term: %d]",
|
||
|
r.id, r.raftLog.committed, r.raftLog.lastIndex(), r.raftLog.lastTerm(), s.Metadata.Index, s.Metadata.Term)
|
||
|
return true
|
||
|
}
|
||
|
|
||
|
// promotable indicates whether state machine can be promoted to leader,
|
||
|
// which is true when its own id is in progress list.
|
||
|
func (r *raft) promotable() bool {
|
||
|
pr := r.prs.Progress[r.id]
|
||
|
return pr != nil && !pr.IsLearner
|
||
|
}
|
||
|
|
||
|
func (r *raft) applyConfChange(cc pb.ConfChangeV2) pb.ConfState {
|
||
|
cfg, prs, err := func() (tracker.Config, tracker.ProgressMap, error) {
|
||
|
changer := confchange.Changer{
|
||
|
Tracker: r.prs,
|
||
|
LastIndex: r.raftLog.lastIndex(),
|
||
|
}
|
||
|
if cc.LeaveJoint() {
|
||
|
return changer.LeaveJoint()
|
||
|
} else if autoLeave, ok := cc.EnterJoint(); ok {
|
||
|
return changer.EnterJoint(autoLeave, cc.Changes...)
|
||
|
}
|
||
|
return changer.Simple(cc.Changes...)
|
||
|
}()
|
||
|
|
||
|
if err != nil {
|
||
|
// TODO(tbg): return the error to the caller.
|
||
|
panic(err)
|
||
|
}
|
||
|
|
||
|
return r.switchToConfig(cfg, prs)
|
||
|
}
|
||
|
|
||
|
// switchToConfig reconfigures this node to use the provided configuration. It
|
||
|
// updates the in-memory state and, when necessary, carries out additional
|
||
|
// actions such as reacting to the removal of nodes or changed quorum
|
||
|
// requirements.
|
||
|
//
|
||
|
// The inputs usually result from restoring a ConfState or applying a ConfChange.
|
||
|
func (r *raft) switchToConfig(cfg tracker.Config, prs tracker.ProgressMap) pb.ConfState {
|
||
|
r.prs.Config = cfg
|
||
|
r.prs.Progress = prs
|
||
|
|
||
|
r.logger.Infof("%x switched to configuration %s", r.id, r.prs.Config)
|
||
|
cs := r.prs.ConfState()
|
||
|
pr, ok := r.prs.Progress[r.id]
|
||
|
|
||
|
// Update whether the node itself is a learner, resetting to false when the
|
||
|
// node is removed.
|
||
|
r.isLearner = ok && pr.IsLearner
|
||
|
|
||
|
if (!ok || r.isLearner) && r.state == StateLeader {
|
||
|
// This node is leader and was removed or demoted. We prevent demotions
|
||
|
// at the time writing but hypothetically we handle them the same way as
|
||
|
// removing the leader: stepping down into the next Term.
|
||
|
//
|
||
|
// TODO(tbg): step down (for sanity) and ask follower with largest Match
|
||
|
// to TimeoutNow (to avoid interruption). This might still drop some
|
||
|
// proposals but it's better than nothing.
|
||
|
//
|
||
|
// TODO(tbg): test this branch. It is untested at the time of writing.
|
||
|
return cs
|
||
|
}
|
||
|
|
||
|
// The remaining steps only make sense if this node is the leader and there
|
||
|
// are other nodes.
|
||
|
if r.state != StateLeader || len(cs.Voters) == 0 {
|
||
|
return cs
|
||
|
}
|
||
|
|
||
|
if r.maybeCommit() {
|
||
|
// If the configuration change means that more entries are committed now,
|
||
|
// broadcast/append to everyone in the updated config.
|
||
|
r.bcastAppend()
|
||
|
} else {
|
||
|
// Otherwise, still probe the newly added replicas; there's no reason to
|
||
|
// let them wait out a heartbeat interval (or the next incoming
|
||
|
// proposal).
|
||
|
r.prs.Visit(func(id uint64, pr *tracker.Progress) {
|
||
|
r.maybeSendAppend(id, false /* sendIfEmpty */)
|
||
|
})
|
||
|
}
|
||
|
// If the the leadTransferee was removed, abort the leadership transfer.
|
||
|
if _, tOK := r.prs.Progress[r.leadTransferee]; !tOK && r.leadTransferee != 0 {
|
||
|
r.abortLeaderTransfer()
|
||
|
}
|
||
|
|
||
|
return cs
|
||
|
}
|
||
|
|
||
|
func (r *raft) loadState(state pb.HardState) {
|
||
|
if state.Commit < r.raftLog.committed || state.Commit > r.raftLog.lastIndex() {
|
||
|
r.logger.Panicf("%x state.commit %d is out of range [%d, %d]", r.id, state.Commit, r.raftLog.committed, r.raftLog.lastIndex())
|
||
|
}
|
||
|
r.raftLog.committed = state.Commit
|
||
|
r.Term = state.Term
|
||
|
r.Vote = state.Vote
|
||
|
}
|
||
|
|
||
|
// pastElectionTimeout returns true iff r.electionElapsed is greater
|
||
|
// than or equal to the randomized election timeout in
|
||
|
// [electiontimeout, 2 * electiontimeout - 1].
|
||
|
func (r *raft) pastElectionTimeout() bool {
|
||
|
return r.electionElapsed >= r.randomizedElectionTimeout
|
||
|
}
|
||
|
|
||
|
func (r *raft) resetRandomizedElectionTimeout() {
|
||
|
r.randomizedElectionTimeout = r.electionTimeout + globalRand.Intn(r.electionTimeout)
|
||
|
}
|
||
|
|
||
|
func (r *raft) sendTimeoutNow(to uint64) {
|
||
|
r.send(pb.Message{To: to, Type: pb.MsgTimeoutNow})
|
||
|
}
|
||
|
|
||
|
func (r *raft) abortLeaderTransfer() {
|
||
|
r.leadTransferee = None
|
||
|
}
|
||
|
|
||
|
// increaseUncommittedSize computes the size of the proposed entries and
|
||
|
// determines whether they would push leader over its maxUncommittedSize limit.
|
||
|
// If the new entries would exceed the limit, the method returns false. If not,
|
||
|
// the increase in uncommitted entry size is recorded and the method returns
|
||
|
// true.
|
||
|
func (r *raft) increaseUncommittedSize(ents []pb.Entry) bool {
|
||
|
var s uint64
|
||
|
for _, e := range ents {
|
||
|
s += uint64(PayloadSize(e))
|
||
|
}
|
||
|
|
||
|
if r.uncommittedSize > 0 && r.uncommittedSize+s > r.maxUncommittedSize {
|
||
|
// If the uncommitted tail of the Raft log is empty, allow any size
|
||
|
// proposal. Otherwise, limit the size of the uncommitted tail of the
|
||
|
// log and drop any proposal that would push the size over the limit.
|
||
|
return false
|
||
|
}
|
||
|
r.uncommittedSize += s
|
||
|
return true
|
||
|
}
|
||
|
|
||
|
// reduceUncommittedSize accounts for the newly committed entries by decreasing
|
||
|
// the uncommitted entry size limit.
|
||
|
func (r *raft) reduceUncommittedSize(ents []pb.Entry) {
|
||
|
if r.uncommittedSize == 0 {
|
||
|
// Fast-path for followers, who do not track or enforce the limit.
|
||
|
return
|
||
|
}
|
||
|
|
||
|
var s uint64
|
||
|
for _, e := range ents {
|
||
|
s += uint64(PayloadSize(e))
|
||
|
}
|
||
|
if s > r.uncommittedSize {
|
||
|
// uncommittedSize may underestimate the size of the uncommitted Raft
|
||
|
// log tail but will never overestimate it. Saturate at 0 instead of
|
||
|
// allowing overflow.
|
||
|
r.uncommittedSize = 0
|
||
|
} else {
|
||
|
r.uncommittedSize -= s
|
||
|
}
|
||
|
}
|
||
|
|
||
|
func numOfPendingConf(ents []pb.Entry) int {
|
||
|
n := 0
|
||
|
for i := range ents {
|
||
|
if ents[i].Type == pb.EntryConfChange {
|
||
|
n++
|
||
|
}
|
||
|
}
|
||
|
return n
|
||
|
}
|