raft.go
func (r *Raft) requestVote(rpc RPC, req *RequestVoteRequest) {
	defer metrics.MeasureSince([]string{"raft", "rpc", "requestVote"}, time.Now())
	r.observe(*req)

	// Setup a response
	resp := &RequestVoteResponse{
		RPCHeader: r.getRPCHeader(),
		Term:      r.getCurrentTerm(),
		Granted:   false,
	}
	var rpcErr error
	defer func() {
		rpc.Respond(resp, rpcErr)
	}()

	// Version 0 servers will panic unless the peers is present. It's only
	// used on them to produce a warning message.
	if r.protocolVersion < 2 {
		resp.Peers = encodePeers(r.configurations.latest, r.trans)
	}

	// Check if we have an existing leader [who's not the candidate]
	candidate := r.trans.DecodePeer(req.Candidate)
	if leader := r.Leader(); leader != "" && leader != candidate {
		r.logger.Warn(fmt.Sprintf("Rejecting vote request from %v since we have a leader: %v",
			candidate, leader))
		return
	}

	// Ignore an older term
	if req.Term < r.getCurrentTerm() {
		return
	}

	// Increase the term if we see a newer one
	if req.Term > r.getCurrentTerm() {
		// Ensure transition to follower
		r.setState(Follower)
		r.setCurrentTerm(req.Term)
		resp.Term = req.Term
	}

	// Check if we have voted yet
	lastVoteTerm, err := r.stable.GetUint64(keyLastVoteTerm)
	if err != nil && err.Error() != "not found" {
		r.logger.Error(fmt.Sprintf("Failed to get last vote term: %v", err))
		return
	}
	lastVoteCandBytes, err := r.stable.Get(keyLastVoteCand)
	if err != nil && err.Error() != "not found" {
		r.logger.Error(fmt.Sprintf("Failed to get last vote candidate: %v", err))
		return
	}

	// Check if we've voted in this election before
	if lastVoteTerm == req.Term && lastVoteCandBytes != nil {
		r.logger.Info(fmt.Sprintf("Duplicate RequestVote for same term: %d", req.Term))
		if bytes.Compare(lastVoteCandBytes, req.Candidate) == 0 {
			r.logger.Warn(fmt.Sprintf("Duplicate RequestVote from candidate: %s", req.Candidate))
			resp.Granted = true
		}
		return
	}

	// Reject if their term is older
	lastIdx, lastTerm := r.getLastEntry()
	if lastTerm > req.LastLogTerm {
		r.logger.Warn(fmt.Sprintf("Rejecting vote request from %v since our last term is greater (%d, %d)",
			candidate, lastTerm, req.LastLogTerm))
		return
	}

	if lastTerm == req.LastLogTerm && lastIdx > req.LastLogIndex {
		r.logger.Warn(fmt.Sprintf("Rejecting vote request from %v since our last index is greater (%d, %d)",
			candidate, lastIdx, req.LastLogIndex))
		return
	}

	// Persist a vote for safety
	if err := r.persistVote(req.Term, req.Candidate); err != nil {
		r.logger.Error(fmt.Sprintf("Failed to persist vote: %v", err))
		return
	}

	resp.Granted = true
	r.setLastContact()
	return
}
func (r *Raft) requestVote(rpc RPC, req *RequestVoteRequest) {
	defer metrics.MeasureSince([]string{"raft", "rpc", "requestVote"}, time.Now())
	r.observe(*req)

	// Setup a response
	resp := &RequestVoteResponse{
		RPCHeader: r.getRPCHeader(),
		Term:      r.getCurrentTerm(),
		Granted:   false,
	}
	var rpcErr error
	defer func() {
		rpc.Respond(resp, rpcErr)
	}()

	// Version 0 servers will panic unless the peers is present. It's only
	// used on them to produce a warning message.
	if r.protocolVersion < 2 {
		resp.Peers = encodePeers(r.configurations.latest, r.trans)
	}

	// Check if we have an existing leader [who's not the candidate]
	candidate := r.trans.DecodePeer(req.Candidate)
	if leader := r.Leader(); leader != "" && leader != candidate {
		r.logger.Warn(fmt.Sprintf("Rejecting vote request from %v since we have a leader: %v",
			candidate, leader))
		return
	}

	// Ignore an older term
	if req.Term < r.getCurrentTerm() {
		return
	}

	// Increase the term if we see a newer one
	if req.Term > r.getCurrentTerm() {
		// Ensure transition to follower
		r.setState(Follower)
		r.setCurrentTerm(req.Term)
		resp.Term = req.Term
	}

	// Check if we have voted yet
	lastVoteTerm, err := r.stable.GetUint64(keyLastVoteTerm)
	if err != nil && err.Error() != "not found" {
		r.logger.Error(fmt.Sprintf("Failed to get last vote term: %v", err))
		return
	}
	lastVoteCandBytes, err := r.stable.Get(keyLastVoteCand)
	if err != nil && err.Error() != "not found" {
		r.logger.Error(fmt.Sprintf("Failed to get last vote candidate: %v", err))
		return
	}

	// Check if we've voted in this election before
	if lastVoteTerm == req.Term && lastVoteCandBytes != nil {
		r.logger.Info(fmt.Sprintf("Duplicate RequestVote for same term: %d", req.Term))
		if bytes.Compare(lastVoteCandBytes, req.Candidate) == 0 {
			r.logger.Warn(fmt.Sprintf("Duplicate RequestVote from candidate: %s", req.Candidate))
			resp.Granted = true
		}
		return
	}

	// Reject if their term is older
	lastIdx, lastTerm := r.getLastEntry()
	if lastTerm > req.LastLogTerm {
		r.logger.Warn(fmt.Sprintf("Rejecting vote request from %v since our last term is greater (%d, %d)",
			candidate, lastTerm, req.LastLogTerm))
		return
	}

	if lastTerm == req.LastLogTerm && lastIdx > req.LastLogIndex {
		r.logger.Warn(fmt.Sprintf("Rejecting vote request from %v since our last index is greater (%d, %d)",
			candidate, lastIdx, req.LastLogIndex))
		return
	}

	// Persist a vote for safety
	if err := r.persistVote(req.Term, req.Candidate); err != nil {
		r.logger.Error(fmt.Sprintf("Failed to persist vote: %v", err))
		return
	}

	resp.Granted = true
	r.setLastContact()
	return
}
func ( r * Raft ) requestVote ( rpc RPC , req * RequestVoteRequest ) { defer metrics . MeasureSince ( [ ] string { " " , " " , " " } , time . Now ( ) ) r . observe ( * req ) // Setup a response resp := & RequestVoteResponse { RPCHeader : r . getRPCHeader ( ) , Term : r . getCurrentTerm ( ) , Granted : false , } var rpcErr error defer func ( ) { rpc . Respond ( resp , rpcErr ) } ( ) // Version 0 servers will panic unless the peers is present. It's only // used on them to produce a warning message. if r . protocolVersion < 2 { resp . Peers = encodePeers ( r . configurations . latest , r . trans ) } // Check if we have an existing leader [who's not the candidate] candidate := r . trans . DecodePeer ( req . Candidate ) if leader := r . Leader ( ) ; leader != " " && leader != candidate { r . logger . Warn ( fmt . Sprintf ( " " , candidate , leader ) ) return } // Ignore an older term if req . Term < r . getCurrentTerm ( ) { return } // Increase the term if we see a newer one if req . Term > r . getCurrentTerm ( ) { // Ensure transition to follower r . setState ( Follower ) r . setCurrentTerm ( req . Term ) resp . Term = req . Term } // Check if we have voted yet lastVoteTerm , err := r . stable . GetUint64 ( keyLastVoteTerm ) if err != nil && err . Error ( ) != " " { r . logger . Error ( fmt . Sprintf ( " " , err ) ) return } lastVoteCandBytes , err := r . stable . Get ( keyLastVoteCand ) if err != nil && err . Error ( ) != " " { r . logger . Error ( fmt . Sprintf ( " " , err ) ) return } // Check if we've voted in this election before if lastVoteTerm == req . Term && lastVoteCandBytes != nil { r . logger . Info ( fmt . Sprintf ( " " , req . Term ) ) if bytes . Compare ( lastVoteCandBytes , req . Candidate ) == 0 { r . logger . Warn ( fmt . Sprintf ( " " , req . Candidate ) ) resp . Granted = true } return } // Reject if their term is older lastIdx , lastTerm := r . getLastEntry ( ) if lastTerm > req . LastLogTerm { r . logger . Warn ( fmt . Sprintf ( " " , candidate , lastTerm , req . LastLogTerm ) ) return } if lastTerm == req . LastLogTerm && lastIdx > req . LastLogIndex { r . logger . Warn ( fmt . Sprintf ( " " , candidate , lastIdx , req . LastLogIndex ) ) return } // Persist a vote for safety if err := r . persistVote ( req . Term , req . Candidate ) ; err != nil { r . logger . Error ( fmt . Sprintf ( " " , err ) ) return } resp . Granted = true r . setLastContact ( ) return }
// requestVote is invoked when we get an request vote RPC call.
requestVote is invoked when we get an request vote RPC call .