Google Git
Sign in
apache / kudu / refs/tags/1.14.0-RC1 / . / src / kudu / consensus / raft_consensus.cc
blob: 0daf78a89e34df0550700be8fc4733ce61c2b703 [file] [log] [blame]
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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.
#include "kudu/consensus/raft_consensus.h"
#include <algorithm>
#include <cmath>
#include <cstdint>
#include <functional>
#include <iterator>
#include <memory>
#include <mutex>
#include <ostream>
#include <unordered_map>
#include <unordered_set>
#include <boost/optional/optional.hpp>
#include <gflags/gflags.h>
#include <google/protobuf/util/message_differencer.h>
#include "kudu/common/timestamp.h"
#include "kudu/common/wire_protocol.h"
#include "kudu/consensus/consensus.pb.h"
#include "kudu/consensus/consensus_meta.h"
#include "kudu/consensus/consensus_meta_manager.h"
#include "kudu/consensus/consensus_peers.h"
#include "kudu/consensus/leader_election.h"
#include "kudu/consensus/log.h"
#include "kudu/consensus/metadata.pb.h"
#include "kudu/consensus/opid_util.h"
#include "kudu/consensus/peer_manager.h"
#include "kudu/consensus/pending_rounds.h"
#include "kudu/consensus/quorum_util.h"
#include "kudu/consensus/time_manager.h"
#include "kudu/gutil/macros.h"
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/port.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/stringpiece.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/walltime.h"
#include "kudu/rpc/periodic.h"
#include "kudu/util/async_util.h"
#include "kudu/util/debug/trace_event.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/logging.h"
#include "kudu/util/metrics.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/process_memory.h"
#include "kudu/util/random.h"
#include "kudu/util/random_util.h"
#include "kudu/util/scoped_cleanup.h"
#include "kudu/util/status.h"
#include "kudu/util/thread_restrictions.h"
#include "kudu/util/threadpool.h"
#include "kudu/util/trace.h"
#include "kudu/util/url-coding.h"
DEFINE_int32(raft_heartbeat_interval_ms, 500,
"The heartbeat interval for Raft replication. The leader produces heartbeats "
"to followers at this interval. The followers expect a heartbeat at this interval "
"and consider a leader to have failed if it misses several in a row.");
TAG_FLAG(raft_heartbeat_interval_ms, advanced);
DEFINE_double(leader_failure_max_missed_heartbeat_periods, 3.0,
"Maximum heartbeat periods that the leader can fail to heartbeat in before we "
"consider the leader to be failed. The total failure timeout in milliseconds is "
"raft_heartbeat_interval_ms times leader_failure_max_missed_heartbeat_periods. "
"The value passed to this flag may be fractional.");
TAG_FLAG(leader_failure_max_missed_heartbeat_periods, advanced);
DEFINE_int32(leader_failure_exp_backoff_max_delta_ms, 20 * 1000,
"Maximum time to sleep in between leader election retries, in addition to the "
"regular timeout. When leader election fails the interval in between retries "
"increases exponentially, up to this value.");
TAG_FLAG(leader_failure_exp_backoff_max_delta_ms, experimental);
DEFINE_bool(enable_leader_failure_detection, true,
"Whether to enable failure detection of tablet leaders. If enabled, attempts will be "
"made to elect a follower as a new leader when the leader is detected to have failed.");
TAG_FLAG(enable_leader_failure_detection, unsafe);
DEFINE_bool(evict_failed_followers, true,
"Whether to evict followers from the Raft config that have fallen "
"too far behind the leader's log to catch up normally or have been "
"unreachable by the leader for longer than "
"follower_unavailable_considered_failed_sec");
TAG_FLAG(evict_failed_followers, advanced);
DEFINE_bool(follower_reject_update_consensus_requests, false,
"Whether a follower will return an error for all UpdateConsensus() requests. "
"Warning! This is only intended for testing.");
TAG_FLAG(follower_reject_update_consensus_requests, unsafe);
DEFINE_bool(follower_fail_all_prepare, false,
"Whether a follower will fail preparing all ops. "
"Warning! This is only intended for testing.");
TAG_FLAG(follower_fail_all_prepare, unsafe);
DEFINE_bool(raft_enable_pre_election, true,
"When enabled, candidates will call a pre-election before "
"running a real leader election.");
TAG_FLAG(raft_enable_pre_election, experimental);
TAG_FLAG(raft_enable_pre_election, runtime);
DEFINE_bool(raft_enable_tombstoned_voting, true,
"When enabled, tombstoned tablets may vote in elections.");
TAG_FLAG(raft_enable_tombstoned_voting, experimental);
TAG_FLAG(raft_enable_tombstoned_voting, runtime);
// Enable improved re-replication (KUDU-1097).
DEFINE_bool(raft_prepare_replacement_before_eviction, true,
"When enabled, failed replicas will only be evicted after a "
"replacement has been prepared for them.");
TAG_FLAG(raft_prepare_replacement_before_eviction, advanced);
TAG_FLAG(raft_prepare_replacement_before_eviction, experimental);
DECLARE_int32(memory_limit_warn_threshold_percentage);
// Metrics
// ---------
METRIC_DEFINE_counter(tablet, follower_memory_pressure_rejections,
"Follower Memory Pressure Rejections",
kudu::MetricUnit::kRequests,
"Number of RPC requests rejected due to "
"memory pressure while FOLLOWER.",
kudu::MetricLevel::kWarn);
METRIC_DEFINE_gauge_int64(tablet, raft_term,
"Current Raft Consensus Term",
kudu::MetricUnit::kUnits,
"Current Term of the Raft Consensus algorithm. This number increments "
"each time a leader election is started.",
kudu::MetricLevel::kDebug);
METRIC_DEFINE_gauge_int64(tablet, failed_elections_since_stable_leader,
"Failed Elections Since Stable Leader",
kudu::MetricUnit::kUnits,
"Number of failed elections on this node since there was a stable "
"leader. This number increments on each failed election and resets on "
"each successful one.",
kudu::MetricLevel::kWarn);
METRIC_DEFINE_gauge_int64(tablet, time_since_last_leader_heartbeat,
"Time Since Last Leader Heartbeat",
kudu::MetricUnit::kMilliseconds,
"The time elapsed since the last heartbeat from the leader "
"in milliseconds. This metric is identically zero on a leader replica.",
kudu::MetricLevel::kDebug);
using boost::optional;
using google::protobuf::util::MessageDifferencer;
using kudu::pb_util::SecureShortDebugString;
using kudu::rpc::PeriodicTimer;
using kudu::tserver::TabletServerErrorPB;
using std::shared_ptr;
using std::string;
using std::unique_ptr;
using std::unordered_set;
using std::weak_ptr;
using strings::Substitute;
namespace kudu {
namespace consensus {
RaftConsensus::RaftConsensus(
ConsensusOptions options,
RaftPeerPB local_peer_pb,
scoped_refptr<ConsensusMetadataManager> cmeta_manager,
ServerContext server_ctx)
: options_(std::move(options)),
local_peer_pb_(std::move(local_peer_pb)),
cmeta_manager_(DCHECK_NOTNULL(std::move(cmeta_manager))),
server_ctx_(std::move(server_ctx)),
state_(kNew),
rng_(GetRandomSeed32()),
leader_is_ready_(false),
leader_transfer_in_progress_(false),
withhold_votes_until_(MonoTime::Min()),
last_received_cur_leader_(MinimumOpId()),
failed_elections_since_stable_leader_(0),
shutdown_(false),
update_calls_for_tests_(0) {
DCHECK(local_peer_pb_.has_permanent_uuid());
}
Status RaftConsensus::Init() {
LockGuard l(lock_);
DCHECK_EQ(kNew, state_) << State_Name(state_);
RETURN_NOT_OK(cmeta_manager_->Load(options_.tablet_id, &cmeta_));
SetStateUnlocked(kInitialized);
return Status::OK();
}
RaftConsensus::~RaftConsensus() {
Shutdown();
}
Status RaftConsensus::Create(ConsensusOptions options,
RaftPeerPB local_peer_pb,
scoped_refptr<ConsensusMetadataManager> cmeta_manager,
ServerContext server_ctx,
shared_ptr<RaftConsensus>* consensus_out) {
shared_ptr<RaftConsensus> consensus(RaftConsensus::make_shared(std::move(options),
std::move(local_peer_pb),
std::move(cmeta_manager),
std::move(server_ctx)));
RETURN_NOT_OK_PREPEND(consensus->Init(), "Unable to initialize Raft consensus");
*consensus_out = std::move(consensus);
return Status::OK();
}
Status RaftConsensus::Start(const ConsensusBootstrapInfo& info,
unique_ptr<PeerProxyFactory> peer_proxy_factory,
scoped_refptr<log::Log> log,
unique_ptr<TimeManager> time_manager,
ConsensusRoundHandler* round_handler,
const scoped_refptr<MetricEntity>& metric_entity,
MarkDirtyCallback cb) {
DCHECK(metric_entity);
peer_proxy_factory_ = DCHECK_NOTNULL(std::move(peer_proxy_factory));
log_ = DCHECK_NOTNULL(std::move(log));
time_manager_ = DCHECK_NOTNULL(std::move(time_manager));
round_handler_ = DCHECK_NOTNULL(round_handler);
mark_dirty_clbk_ = std::move(cb);
term_metric_ = metric_entity->FindOrCreateGauge(&METRIC_raft_term,
CurrentTerm(),
MergeType::kMax);
follower_memory_pressure_rejections_ =
metric_entity->FindOrCreateCounter(&METRIC_follower_memory_pressure_rejections);
num_failed_elections_metric_ =
metric_entity->FindOrCreateGauge(&METRIC_failed_elections_since_stable_leader,
failed_elections_since_stable_leader_);
METRIC_time_since_last_leader_heartbeat.InstantiateFunctionGauge(
metric_entity, [this]() { return this->GetMillisSinceLastLeaderHeartbeat(); },
MergeType::kMax)
->AutoDetach(&metric_detacher_);
// A single Raft thread pool token is shared between RaftConsensus and
// PeerManager. Because PeerManager is owned by RaftConsensus, it receives a
// raw pointer to the token, to emphasize that RaftConsensus is responsible
// for destroying the token.
ThreadPool* raft_pool = server_ctx_.raft_pool;
raft_pool_token_ = raft_pool->NewToken(ThreadPool::ExecutionMode::CONCURRENT);
// The message queue that keeps track of which operations need to be replicated
// where.
//
// Note: the message queue receives a dedicated Raft thread pool token so that
// its submissions don't block other submissions by RaftConsensus (such as
// heartbeat processing).
//
// TODO(adar): the token is SERIAL to match the previous single-thread
// observer pool behavior, but CONCURRENT may be safe here.
unique_ptr<PeerMessageQueue> queue(new PeerMessageQueue(
metric_entity,
log_,
time_manager_.get(),
local_peer_pb_,
options_.tablet_id,
raft_pool->NewToken(ThreadPool::ExecutionMode::SERIAL),
server_ctx_.quiescing,
info.last_id,
info.last_committed_id,
server_ctx_.allow_status_msg_for_failed_peer));
// A manager for the set of peers that actually send the operations both remotely
// and to the local wal.
unique_ptr<PeerManager> peer_manager(new PeerManager(options_.tablet_id,
peer_uuid(),
peer_proxy_factory_.get(),
queue.get(),
raft_pool_token_.get(),
log_));
unique_ptr<PendingRounds> pending(new PendingRounds(
LogPrefixThreadSafe(), time_manager_.get()));
// Capture a weak_ptr reference into the functor so it can safely handle
// outliving the consensus instance.
weak_ptr<RaftConsensus> w = shared_from_this();
PeriodicTimer::Options opts;
opts.one_shot = true;
failure_detector_ = PeriodicTimer::Create(
peer_proxy_factory_->messenger(),
[w]() {
if (auto consensus = w.lock()) {
consensus->ReportFailureDetected();
}
},
MinimumElectionTimeout(),
opts);
transfer_period_timer_ = PeriodicTimer::Create(
peer_proxy_factory_->messenger(),
[w]() {
if (auto consensus = w.lock()) {
consensus->EndLeaderTransferPeriod();
}
},
MinimumElectionTimeout(),
opts);
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
CHECK_EQ(kInitialized, state_) << LogPrefixUnlocked() << "Illegal state for Start(): "
<< State_Name(state_);
queue_ = std::move(queue);
peer_manager_ = std::move(peer_manager);
pending_ = std::move(pending);
ClearLeaderUnlocked();
// Our last persisted term can be higher than the last persisted operation
// (i.e. if we called an election) but reverse should never happen.
if (info.last_id.term() > CurrentTermUnlocked()) {
return Status::Corruption(Substitute("Unable to start RaftConsensus: "
"The last op in the WAL with id $0 has a term ($1) that is greater "
"than the latest recorded term, which is $2",
OpIdToString(info.last_id),
info.last_id.term(),
CurrentTermUnlocked()));
}
// Append any uncommitted replicate messages found during log replay to the queue.
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Replica starting. Triggering "
<< info.orphaned_replicates.size()
<< " pending ops. Active config: "
<< SecureShortDebugString(cmeta_->ActiveConfig());
for (ReplicateMsg* replicate : info.orphaned_replicates) {
ReplicateRefPtr replicate_ptr = make_scoped_refptr_replicate(new ReplicateMsg(*replicate));
RETURN_NOT_OK(StartFollowerOpUnlocked(replicate_ptr));
}
// Set the initial committed opid for the PendingRounds only after
// appending any uncommitted replicate messages to the queue.
pending_->SetInitialCommittedOpId(info.last_committed_id);
// If this is the first term expire the FD immediately so that we have a
// fast first election, otherwise we just let the timer expire normally.
boost::optional<MonoDelta> fd_initial_delta;
if (CurrentTermUnlocked() == 0) {
// The failure detector is initialized to a low value to trigger an early
// election (unless someone else requested a vote from us first, which
// resets the election timer).
//
// We do it this way instead of immediately running an election to get a
// higher likelihood of enough servers being available when the first one
// attempts an election to avoid multiple election cycles on startup,
// while keeping that "waiting period" random.
if (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Consensus starting up: Expiring failure detector timer "
"to make a prompt election more likely";
fd_initial_delta = MonoDelta::FromMilliseconds(
rng_.Uniform(FLAGS_raft_heartbeat_interval_ms));
}
}
// Now assume non-leader replica duties.
RETURN_NOT_OK(BecomeReplicaUnlocked(fd_initial_delta));
SetStateUnlocked(kRunning);
}
if (IsSingleVoterConfig() && FLAGS_enable_leader_failure_detection) {
LOG_WITH_PREFIX(INFO) << "Only one voter in the Raft config. Triggering election immediately";
WARN_NOT_OK_EVERY_N_SECS(StartElection(NORMAL_ELECTION, INITIAL_SINGLE_NODE_ELECTION),
"Couldn't start leader election", 10);
}
// Report become visible to the Master.
MarkDirty("RaftConsensus started");
return Status::OK();
}
bool RaftConsensus::IsRunning() const {
return state_ == kRunning;
}
Status RaftConsensus::EmulateElectionForTests() {
TRACE_EVENT2("consensus", "RaftConsensus::EmulateElectionForTests",
"peer", peer_uuid(),
"tablet", options_.tablet_id);
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
RETURN_NOT_OK(CheckRunningUnlocked());
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Emulating election...";
// Assume leadership of new term.
RETURN_NOT_OK(HandleTermAdvanceUnlocked(CurrentTermUnlocked() + 1));
SetLeaderUuidUnlocked(peer_uuid());
return BecomeLeaderUnlocked();
}
namespace {
const char* ModeString(RaftConsensus::ElectionMode mode) {
switch (mode) {
case RaftConsensus::NORMAL_ELECTION:
return "leader election";
case RaftConsensus::PRE_ELECTION:
return "pre-election";
case RaftConsensus::ELECT_EVEN_IF_LEADER_IS_ALIVE:
return "forced leader election";
}
__builtin_unreachable(); // silence gcc warnings
}
string ReasonString(RaftConsensus::ElectionReason reason, StringPiece leader_uuid) {
switch (reason) {
case RaftConsensus::INITIAL_SINGLE_NODE_ELECTION:
return "initial election of a single-replica configuration";
case RaftConsensus::EXTERNAL_REQUEST:
return "received explicit request";
case RaftConsensus::ELECTION_TIMEOUT_EXPIRED:
if (leader_uuid.empty()) {
return "no leader contacted us within the election timeout";
}
return Substitute("detected failure of leader $0", leader_uuid);
}
__builtin_unreachable(); // silence gcc warnings
}
} // anonymous namespace
Status RaftConsensus::StartElection(ElectionMode mode, ElectionReason reason) {
if (server_ctx_.quiescing && server_ctx_.quiescing->load()) {
return Status::IllegalState("leader elections are disabled");
}
const char* const mode_str = ModeString(mode);
TRACE_EVENT2("consensus", "RaftConsensus::StartElection",
"peer", LogPrefixThreadSafe(),
"mode", mode_str);
scoped_refptr<LeaderElection> election;
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
RETURN_NOT_OK(CheckRunningUnlocked());
RaftPeerPB::Role active_role = cmeta_->active_role();
if (active_role == RaftPeerPB::LEADER) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << Substitute(
"Not starting $0 -- already a leader", mode_str);
return Status::OK();
}
if (PREDICT_FALSE(!consensus::IsVoterRole(active_role))) {
// A non-voter should not start leader elections. The leader failure
// detector should be re-enabled once the non-voter replica is promoted
// to voter replica.
return Status::IllegalState("only voting members can start elections",
SecureShortDebugString(cmeta_->ActiveConfig()));
}
if (PREDICT_FALSE(active_role == RaftPeerPB::NON_PARTICIPANT)) {
SnoozeFailureDetector();
return Status::IllegalState("Not starting election: node is currently "
"a non-participant in the Raft config",
SecureShortDebugString(cmeta_->ActiveConfig()));
}
LOG_WITH_PREFIX_UNLOCKED(INFO)
<< "Starting " << mode_str
<< " (" << ReasonString(reason, GetLeaderUuidUnlocked()) << ")";
// Snooze to avoid the election timer firing again as much as possible.
// We do not disable the election timer while running an election, so that
// if the election times out, we will try again.
MonoDelta timeout = LeaderElectionExpBackoffDeltaUnlocked();
SnoozeFailureDetector(string("starting election"), timeout);
// Increment the term and vote for ourselves, unless it's a pre-election.
if (mode != PRE_ELECTION) {
// TODO(mpercy): Consider using a separate Mutex for voting, which must sync to disk.
// We skip flushing the term to disk because setting the vote just below also
// flushes to disk, and the double fsync doesn't buy us anything.
RETURN_NOT_OK(HandleTermAdvanceUnlocked(CurrentTermUnlocked() + 1,
SKIP_FLUSH_TO_DISK));
RETURN_NOT_OK(SetVotedForCurrentTermUnlocked(peer_uuid()));
}
RaftConfigPB active_config = cmeta_->ActiveConfig();
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Starting " << mode_str << " with config: "
<< SecureShortDebugString(active_config);
// Initialize the VoteCounter.
int num_voters = CountVoters(active_config);
int majority_size = MajoritySize(num_voters);
VoteCounter counter(num_voters, majority_size);
// Vote for ourselves.
bool duplicate;
RETURN_NOT_OK(counter.RegisterVote(peer_uuid(), VOTE_GRANTED, &duplicate));
CHECK(!duplicate) << LogPrefixUnlocked()
<< "Inexplicable duplicate self-vote for term "
<< CurrentTermUnlocked();
VoteRequestPB request;
request.set_ignore_live_leader(mode == ELECT_EVEN_IF_LEADER_IS_ALIVE);
request.set_candidate_uuid(peer_uuid());
if (mode == PRE_ELECTION) {
// In a pre-election, we haven't bumped our own term yet, so we need to be
// asking for votes for the next term.
request.set_is_pre_election(true);
request.set_candidate_term(CurrentTermUnlocked() + 1);
} else {
request.set_candidate_term(CurrentTermUnlocked());
}
request.set_tablet_id(options_.tablet_id);
*request.mutable_candidate_status()->mutable_last_received() =
queue_->GetLastOpIdInLog();
auto self = shared_from_this();
election.reset(new LeaderElection(
std::move(active_config),
// The RaftConsensus ref passed below ensures that this raw pointer
// remains safe to use for the entirety of LeaderElection's life.
peer_proxy_factory_.get(),
std::move(request), std::move(counter), timeout,
[self, reason](const ElectionResult& result) {
self->ElectionCallback(reason, result);
}));
}
// Start the election outside the lock.
election->Run();
return Status::OK();
}
Status RaftConsensus::WaitUntilLeaderForTests(const MonoDelta& timeout) {
MonoTime deadline = MonoTime::Now() + timeout;
while (role() != consensus::RaftPeerPB::LEADER) {
if (MonoTime::Now() >= deadline) {
return Status::TimedOut(Substitute("Peer $0 is not leader of tablet $1 after $2. Role: $3",
peer_uuid(), options_.tablet_id, timeout.ToString(),
role()));
}
SleepFor(MonoDelta::FromMilliseconds(10));
}
return Status::OK();
}
Status RaftConsensus::StepDown(LeaderStepDownResponsePB* resp) {
TRACE_EVENT0("consensus", "RaftConsensus::StepDown");
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
DCHECK((queue_->IsInLeaderMode() && cmeta_->active_role() == RaftPeerPB::LEADER) ||
(!queue_->IsInLeaderMode() && cmeta_->active_role() != RaftPeerPB::LEADER));
RETURN_NOT_OK(CheckRunningUnlocked());
if (cmeta_->active_role() != RaftPeerPB::LEADER) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Rejecting request to step down while not leader";
resp->mutable_error()->set_code(TabletServerErrorPB::NOT_THE_LEADER);
StatusToPB(Status::IllegalState("Not currently leader"),
resp->mutable_error()->mutable_status());
// We return OK so that the tablet service won't overwrite the error code.
return Status::OK();
}
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Received request to step down";
RETURN_NOT_OK(HandleTermAdvanceUnlocked(CurrentTermUnlocked() + 1,
SKIP_FLUSH_TO_DISK));
// Snooze the failure detector for an extra leader failure timeout.
// This should ensure that a different replica is elected leader after this one steps down.
SnoozeFailureDetector(string("explicit stepdown request"), MonoDelta::FromMilliseconds(
2 * MinimumElectionTimeout().ToMilliseconds()));
return Status::OK();
}
Status RaftConsensus::TransferLeadership(const boost::optional<string>& new_leader_uuid,
LeaderStepDownResponsePB* resp) {
TRACE_EVENT0("consensus", "RaftConsensus::TransferLeadership");
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Received request to transfer leadership"
<< (new_leader_uuid ?
Substitute(" to TS $0", *new_leader_uuid) :
"");
DCHECK((queue_->IsInLeaderMode() && cmeta_->active_role() == RaftPeerPB::LEADER) ||
(!queue_->IsInLeaderMode() && cmeta_->active_role() != RaftPeerPB::LEADER));
RETURN_NOT_OK(CheckRunningUnlocked());
if (cmeta_->active_role() != RaftPeerPB::LEADER) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Rejecting request to transer leadership while not leader";
resp->mutable_error()->set_code(TabletServerErrorPB::NOT_THE_LEADER);
StatusToPB(Status::IllegalState("not currently leader"),
resp->mutable_error()->mutable_status());
// We return OK so that the tablet service won't overwrite the error code.
return Status::OK();
}
if (new_leader_uuid) {
if (*new_leader_uuid == peer_uuid()) {
// Short-circuit as we are transferring leadership to ourselves and we
// already checked that we are leader.
return Status::OK();
}
if (!IsRaftConfigVoter(*new_leader_uuid, cmeta_->ActiveConfig())) {
const string msg = Substitute("tablet server $0 is not a voter in the active config",
*new_leader_uuid);
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Rejecting request to transfer leadership "
<< "because " << msg;
return Status::InvalidArgument(msg);
}
}
return BeginLeaderTransferPeriodUnlocked(new_leader_uuid);
}
Status RaftConsensus::BeginLeaderTransferPeriodUnlocked(
const boost::optional<string>& successor_uuid) {
DCHECK(lock_.is_locked());
if (std::atomic_exchange(&leader_transfer_in_progress_, true)) {
return Status::ServiceUnavailable(
Substitute("leadership transfer for $0 already in progress",
options_.tablet_id));
}
queue_->BeginWatchForSuccessor(successor_uuid);
transfer_period_timer_->Start();
return Status::OK();
}
void RaftConsensus::EndLeaderTransferPeriod() {
transfer_period_timer_->Stop();
queue_->EndWatchForSuccessor();
leader_transfer_in_progress_ = false;
}
scoped_refptr<ConsensusRound> RaftConsensus::NewRound(
unique_ptr<ReplicateMsg> replicate_msg,
ConsensusReplicatedCallback replicated_cb) {
return make_scoped_refptr(new ConsensusRound(this,
std::move(replicate_msg),
std::move(replicated_cb)));
}
void RaftConsensus::ReportFailureDetectedTask() {
Status s = StartElection(FLAGS_raft_enable_pre_election ?
PRE_ELECTION : NORMAL_ELECTION, ELECTION_TIMEOUT_EXPIRED);
if (PREDICT_FALSE(!s.ok())) {
WARN_NOT_OK_EVERY_N_SECS(
s, LogPrefixThreadSafe() + "failed to trigger leader election", 10);
// Normally the failure detector would be enabled at the end of the election,
// but since the election failed to start, we must reenable explicitly.
EnableFailureDetector();
}
}
void RaftConsensus::ReportFailureDetected() {
// We're running on a timer thread; start an election on a different thread.
//
// There's no need to reenable the failure detector; if this fails, it's a
// sign that RaftConsensus has stopped and we no longer need failure detection.
auto self = shared_from_this();
Status s = raft_pool_token_->Submit([self]() { self->ReportFailureDetectedTask(); });
if (PREDICT_FALSE(!s.ok())) {
static const char* msg = "failed to submit failure detected task";
CHECK(s.IsServiceUnavailable()) << LogPrefixThreadSafe() << msg;
WARN_NOT_OK(s, LogPrefixThreadSafe() + msg);
}
}
Status RaftConsensus::BecomeLeaderUnlocked() {
DCHECK(lock_.is_locked());
TRACE_EVENT2("consensus", "RaftConsensus::BecomeLeaderUnlocked",
"peer", peer_uuid(),
"tablet", options_.tablet_id);
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Becoming Leader. State: " << ToStringUnlocked();
// Disable FD while we are leader.
DisableFailureDetector();
// Don't vote for anyone if we're a leader.
withhold_votes_until_ = MonoTime::Max();
// Leadership never starts in a transfer period.
EndLeaderTransferPeriod();
queue_->RegisterObserver(this);
bool was_leader = queue_->IsInLeaderMode();
RETURN_NOT_OK(RefreshConsensusQueueAndPeersUnlocked());
if (!was_leader && server_ctx_.num_leaders) {
server_ctx_.num_leaders->Increment();
}
// Initiate a NO_OP op that is sent at the beginning of every term
// change in raft.
auto replicate = new ReplicateMsg;
replicate->set_op_type(NO_OP);
replicate->mutable_noop_request(); // Define the no-op request field.
CHECK_OK(time_manager_->AssignTimestamp(replicate));
scoped_refptr<ConsensusRound> round(
new ConsensusRound(this, make_scoped_refptr(new RefCountedReplicate(replicate))));
auto* round_raw = round.get();
round->SetConsensusReplicatedCallback(
[this, round_raw](const Status& s) {
this->NonTxRoundReplicationFinished(round_raw, &DoNothingStatusCB, s);
});
last_leader_communication_time_micros_ = 0;
RETURN_NOT_OK(AppendNewRoundToQueueUnlocked(round));
leader_is_ready_ = true;
return Status::OK();
}
Status RaftConsensus::BecomeReplicaUnlocked(boost::optional<MonoDelta> fd_delta) {
DCHECK(lock_.is_locked());
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Becoming Follower/Learner. State: "
<< ToStringUnlocked();
ClearLeaderUnlocked();
// Enable/disable leader failure detection if becoming VOTER/NON_VOTER replica
// correspondingly.
UpdateFailureDetectorState(std::move(fd_delta));
// Now that we're a replica, we can allow voting for other nodes.
withhold_votes_until_ = MonoTime::Min();
// Deregister ourselves from the queue. We no longer need to track what gets
// replicated since we're stepping down.
queue_->UnRegisterObserver(this);
bool was_leader = queue_->IsInLeaderMode();
queue_->SetNonLeaderMode(cmeta_->ActiveConfig());
if (was_leader && server_ctx_.num_leaders) {
server_ctx_.num_leaders->IncrementBy(-1);
}
peer_manager_->Close();
return Status::OK();
}
Status RaftConsensus::Replicate(const scoped_refptr<ConsensusRound>& round) {
std::lock_guard<simple_spinlock> lock(update_lock_);
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
RETURN_NOT_OK(CheckSafeToReplicateUnlocked(*round->replicate_msg()));
RETURN_NOT_OK(round->CheckBoundTerm(CurrentTermUnlocked()));
RETURN_NOT_OK(AppendNewRoundToQueueUnlocked(round));
}
peer_manager_->SignalRequest();
return Status::OK();
}
Status RaftConsensus::CheckLeadershipAndBindTerm(const scoped_refptr<ConsensusRound>& round) {
#ifndef NDEBUG
const auto& msg = *round->replicate_msg();
DCHECK(!msg.has_id()) << "should not have an ID yet: "
<< SecureShortDebugString(msg);
#endif
// Get a snapshot of an atomic member for consistency between the condition
// and the error message, if any.
const State state = state_;
if (PREDICT_FALSE(state != kRunning)) {
return Status::IllegalState("RaftConsensus is not running",
Substitute("state $0", State_Name(state)));
}
// The order of reading role_and_term and leader_is_ready is essential to
// deal with situations when leader_is_ready and role_and_term are read
// in different Raft terms.
// * It's safe to bind to a stale term even if we've asserted leadership
// in a newer term: the stale op will be rejected on followers anyway
// because Raft guarantees that a majority of replicas have accepted
// the new term.
// * It's safe for the term to be stale if we haven't asserted leadership
// for a newer term because we'll exit out below.
// * It's unsafe to bind to a newer term if we've asserted leadership
// for a stale term, hence this ordering.
const auto role_and_term = cmeta_->GetRoleAndTerm();
const bool leader_is_ready = leader_is_ready_;
const auto role = role_and_term.first;
switch (role) {
case RaftPeerPB::LEADER:
if (leader_transfer_in_progress_) {
return Status::ServiceUnavailable("leader transfer in progress");
}
if (!leader_is_ready) {
// Leader replica should not accept write operations before scheduling
// the replication of a NO_OP to assert its leadership in the current
// term. Otherwise there might be a race, so the very first accepted
// write operation might have timestamp lower than the timestamp
// of the NO_OP. That would trigger an assertion in MVCC.
return Status::ServiceUnavailable("leader is not yet ready");
}
break;
default:
return Status::IllegalState(
Substitute("replica $0 is not leader of this config: current role $1",
peer_uuid(), RaftPeerPB::Role_Name(role)));
}
const auto term = role_and_term.second;
round->BindToTerm(term);
return Status::OK();
}
Status RaftConsensus::AppendNewRoundToQueueUnlocked(const scoped_refptr<ConsensusRound>& round) {
DCHECK(lock_.is_locked());
*round->replicate_msg()->mutable_id() = queue_->GetNextOpId();
RETURN_NOT_OK(AddPendingOperationUnlocked(round));
// The only reasons for a bad status would be if the log itself were shut down,
// or if we had an actual IO error, which we currently don't handle.
CHECK_OK_PREPEND(queue_->AppendOperation(round->replicate_scoped_refptr()),
Substitute("$0: could not append to queue", LogPrefixUnlocked()));
return Status::OK();
}
Status RaftConsensus::AddPendingOperationUnlocked(const scoped_refptr<ConsensusRound>& round) {
DCHECK(lock_.is_locked());
DCHECK(pending_);
// If we are adding a pending config change, we need to propagate it to the
// metadata.
if (PREDICT_FALSE(round->replicate_msg()->op_type() == CHANGE_CONFIG_OP)) {
// Fill in the opid for the proposed new configuration. This has to be done
// here rather than when it's first created because we didn't yet have an
// OpId assigned at creation time.
ChangeConfigRecordPB* change_record = round->replicate_msg()->mutable_change_config_record();
change_record->mutable_new_config()->set_opid_index(round->replicate_msg()->id().index());
DCHECK(change_record->IsInitialized())
<< "change_config_record missing required fields: "
<< change_record->InitializationErrorString();
const RaftConfigPB& new_config = change_record->new_config();
if (!new_config.unsafe_config_change()) {
Status s = CheckNoConfigChangePendingUnlocked();
if (PREDICT_FALSE(!s.ok())) {
s = s.CloneAndAppend(Substitute("\n New config: $0", SecureShortDebugString(new_config)));
LOG_WITH_PREFIX_UNLOCKED(INFO) << s.ToString();
return s;
}
}
// Check if the pending Raft config has an OpId less than the committed
// config. If so, this is a replay at startup in which the COMMIT
// messages were delayed.
int64_t committed_config_opid_index = cmeta_->GetConfigOpIdIndex(COMMITTED_CONFIG);
if (round->replicate_msg()->id().index() > committed_config_opid_index) {
RETURN_NOT_OK(SetPendingConfigUnlocked(new_config));
if (cmeta_->active_role() == RaftPeerPB::LEADER) {
RETURN_NOT_OK(RefreshConsensusQueueAndPeersUnlocked());
}
} else {
LOG_WITH_PREFIX_UNLOCKED(INFO)
<< "Ignoring setting pending config change with OpId "
<< round->replicate_msg()->id() << " because the committed config has OpId index "
<< committed_config_opid_index << ". The config change we are ignoring is: "
<< "Old config: { " << SecureShortDebugString(change_record->old_config()) << " }. "
<< "New config: { " << SecureShortDebugString(new_config) << " }";
}
}
return pending_->AddPendingOperation(round);
}
void RaftConsensus::NotifyCommitIndex(int64_t commit_index) {
TRACE_EVENT2("consensus", "RaftConsensus::NotifyCommitIndex",
"tablet", options_.tablet_id,
"commit_index", commit_index);
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
// We will process commit notifications while shutting down because a replica
// which has initiated a Prepare() / Replicate() may eventually commit even if
// its state has changed after the initial Append() / Update().
const State state = state_;
if (PREDICT_FALSE(state != kRunning && state != kStopping)) {
LOG_WITH_PREFIX_UNLOCKED(WARNING) << "Unable to update committed index: "
<< "Replica not in running state: "
<< State_Name(state);
return;
}
pending_->AdvanceCommittedIndex(commit_index);
if (cmeta_->active_role() == RaftPeerPB::LEADER) {
peer_manager_->SignalRequest(false);
}
}
void RaftConsensus::NotifyTermChange(int64_t term) {
TRACE_EVENT2("consensus", "RaftConsensus::NotifyTermChange",
"tablet", options_.tablet_id,
"term", term);
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
Status s = CheckRunningUnlocked();
if (PREDICT_FALSE(!s.ok())) {
LOG_WITH_PREFIX_UNLOCKED(WARNING) << "Unable to handle notification of new term "
<< "(" << term << "): " << s.ToString();
return;
}
WARN_NOT_OK(HandleTermAdvanceUnlocked(term), "Couldn't advance consensus term.");
}
void RaftConsensus::NotifyFailedFollower(const string& uuid,
int64_t term,
const string& reason) {
// Common info used in all of the log messages within this method.
string fail_msg = Substitute("Processing failure of peer $0 in term $1 ($2): ",
uuid, term, reason);
if (!FLAGS_evict_failed_followers) {
LOG(INFO) << LogPrefixThreadSafe() << fail_msg
<< "Eviction of failed followers is disabled. Doing nothing.";
return;
}
RaftConfigPB committed_config;
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
int64_t current_term = CurrentTermUnlocked();
if (current_term != term) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << fail_msg << "Notified about a follower failure in "
<< "previous term " << term << ", but a leader election "
<< "likely occurred since the failure was detected. "
<< "Doing nothing.";
return;
}
if (cmeta_->has_pending_config()) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << fail_msg << "There is already a config change operation "
<< "in progress. Unable to evict follower until it completes. "
<< "Doing nothing.";
return;
}
committed_config = cmeta_->CommittedConfig();
}
// Run config change on thread pool after dropping lock.
auto self = shared_from_this();
WARN_NOT_OK(raft_pool_token_->Submit(
[self, uuid, committed_config, reason]() {
self->TryRemoveFollowerTask(uuid, committed_config, reason);
}),
LogPrefixThreadSafe() + "Unable to start TryRemoveFollowerTask");
}
void RaftConsensus::NotifyPeerToPromote(const string& peer_uuid) {
// Run the config change on the raft thread pool.
auto self = shared_from_this();
WARN_NOT_OK(raft_pool_token_->Submit(
[self, peer_uuid]() { self->TryPromoteNonVoterTask(peer_uuid); }),
LogPrefixThreadSafe() + "Unable to start TryPromoteNonVoterTask");
}
void RaftConsensus::NotifyPeerToStartElection(const string& peer_uuid) {
const auto& log_prefix = LogPrefixThreadSafe();
LOG(INFO) << log_prefix << ": Instructing follower " << peer_uuid << " to start an election";
auto self = shared_from_this();
WARN_NOT_OK(raft_pool_token_->Submit(
[self, peer_uuid]() { self->TryStartElectionOnPeerTask(peer_uuid); }),
log_prefix + "Unable to start TryStartElectionOnPeerTask");
}
void RaftConsensus::NotifyPeerHealthChange() {
MarkDirty("Peer health change");
}
void RaftConsensus::TryRemoveFollowerTask(const string& uuid,
const RaftConfigPB& committed_config,
const string& reason) {
ChangeConfigRequestPB req;
req.set_tablet_id(options_.tablet_id);
req.mutable_server()->set_permanent_uuid(uuid);
req.set_type(REMOVE_PEER);
req.set_cas_config_opid_index(committed_config.opid_index());
LOG(INFO) << LogPrefixThreadSafe() << "Attempting to remove follower "
<< uuid << " from the Raft config. Reason: " << reason;
boost::optional<TabletServerErrorPB::Code> error_code;
WARN_NOT_OK(ChangeConfig(req, &DoNothingStatusCB, &error_code),
LogPrefixThreadSafe() + "Unable to remove follower " + uuid);
}
void RaftConsensus::TryPromoteNonVoterTask(const string& peer_uuid) {
string msg = Substitute("attempt to promote peer $0: ", peer_uuid);
int64_t current_committed_config_index;
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
if (cmeta_->has_pending_config()) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << msg << "there is already a config change operation "
<< "in progress. Unable to promote follower until it "
<< "completes. Doing nothing.";
return;
}
// Check if the peer is still part of the current committed config.
RaftConfigPB committed_config = cmeta_->CommittedConfig();
current_committed_config_index = committed_config.opid_index();
RaftPeerPB* peer_pb;
Status s = GetRaftConfigMember(&committed_config, peer_uuid, &peer_pb);
if (!s.ok()) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << msg << "can't find peer in the "
<< "current committed config: "
<< committed_config.ShortDebugString()
<< ". Doing nothing.";
return;
}
// Also check if the peer it still a NON_VOTER waiting for promotion.
if (peer_pb->member_type() != RaftPeerPB::NON_VOTER || !peer_pb->attrs().promote()) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << msg << "peer is either no longer a NON_VOTER "
<< "or not marked for promotion anymore. Current "
<< "config: " << committed_config.ShortDebugString()
<< ". Doing nothing.";
return;
}
}
ChangeConfigRequestPB req;
req.set_tablet_id(options_.tablet_id);
req.set_type(MODIFY_PEER);
req.mutable_server()->set_permanent_uuid(peer_uuid);
req.mutable_server()->set_member_type(RaftPeerPB::VOTER);
req.mutable_server()->mutable_attrs()->set_promote(false);
req.set_cas_config_opid_index(current_committed_config_index);
LOG(INFO) << LogPrefixThreadSafe() << "attempting to promote NON_VOTER "
<< peer_uuid << " to VOTER";
boost::optional<TabletServerErrorPB::Code> error_code;
WARN_NOT_OK(ChangeConfig(req, &DoNothingStatusCB, &error_code),
LogPrefixThreadSafe() + Substitute("Unable to promote non-voter $0", peer_uuid));
}
void RaftConsensus::TryStartElectionOnPeerTask(const string& peer_uuid) {
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
// Double-check that the peer is a voter in the active config.
if (!IsRaftConfigVoter(peer_uuid, cmeta_->ActiveConfig())) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Not signalling peer " << peer_uuid
<< "to start an election: it's not a voter "
<< "in the active config.";
return;
}
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Signalling peer " << peer_uuid
<< " to start an election";
WARN_NOT_OK(peer_manager_->StartElection(peer_uuid),
Substitute("unable to start election on peer $0", peer_uuid));
}
Status RaftConsensus::Update(const ConsensusRequestPB* request,
ConsensusResponsePB* response) {
++update_calls_for_tests_;
if (PREDICT_FALSE(FLAGS_follower_reject_update_consensus_requests)) {
return Status::IllegalState("Rejected: --follower_reject_update_consensus_requests "
"is set to true.");
}
response->set_responder_uuid(peer_uuid());
VLOG_WITH_PREFIX(2) << "Replica received request: " << SecureShortDebugString(*request);
// see var declaration
std::lock_guard<simple_spinlock> lock(update_lock_);
Status s = UpdateReplica(request, response);
if (PREDICT_FALSE(VLOG_IS_ON(1))) {
if (request->ops().empty()) {
VLOG_WITH_PREFIX(1)
<< Substitute("Replica replied to status only request. Replica: $0. Response: $1",
ToString(), SecureShortDebugString(*response));
}
}
return s;
}
// Helper function to check if the op is a no-op op.
static bool IsConsensusOnlyOperation(OperationType op_type) {
return op_type == NO_OP || op_type == CHANGE_CONFIG_OP;
}
Status RaftConsensus::StartFollowerOpUnlocked(const ReplicateRefPtr& msg) {
DCHECK(lock_.is_locked());
if (IsConsensusOnlyOperation(msg->get()->op_type())) {
return StartConsensusOnlyRoundUnlocked(msg);
}
if (PREDICT_FALSE(FLAGS_follower_fail_all_prepare)) {
return Status::IllegalState("Rejected: --follower_fail_all_prepare "
"is set to true.");
}
VLOG_WITH_PREFIX_UNLOCKED(1) << "Starting op: "
<< SecureShortDebugString(msg->get()->id());
scoped_refptr<ConsensusRound> round(new ConsensusRound(this, msg));
ConsensusRound* round_ptr = round.get();
RETURN_NOT_OK(round_handler_->StartFollowerOp(round));
return AddPendingOperationUnlocked(round_ptr);
}
bool RaftConsensus::IsSingleVoterConfig() const {
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
return cmeta_->CountVotersInConfig(COMMITTED_CONFIG) == 1 &&
cmeta_->IsVoterInConfig(peer_uuid(), COMMITTED_CONFIG);
}
string RaftConsensus::LeaderRequest::OpsRangeString() const {
string ret;
ret.reserve(100);
ret.push_back('[');
if (!messages.empty()) {
const OpId& first_op = (*messages.begin())->get()->id();
const OpId& last_op = (*messages.rbegin())->get()->id();
strings::SubstituteAndAppend(&ret, "$0.$1-$2.$3",
first_op.term(), first_op.index(),
last_op.term(), last_op.index());
}
ret.push_back(']');
return ret;
}
void RaftConsensus::DeduplicateLeaderRequestUnlocked(const ConsensusRequestPB* rpc_req,
LeaderRequest* deduplicated_req) {
DCHECK(lock_.is_locked());
// TODO(todd): use queue committed index?
int64_t last_committed_index = pending_->GetCommittedIndex();
// The leader's preceding id.
deduplicated_req->preceding_opid = &rpc_req->preceding_id();
int64_t dedup_up_to_index = queue_->GetLastOpIdInLog().index();
deduplicated_req->first_message_idx = -1;
// In this loop we discard duplicates and advance the leader's preceding id
// accordingly.
for (int i = 0; i < rpc_req->ops_size(); i++) {
const ReplicateMsg* leader_msg = &rpc_req->ops(i);
if (leader_msg->id().index() <= last_committed_index) {
VLOG_WITH_PREFIX_UNLOCKED(2) << "Skipping op id " << leader_msg->id()
<< " (already committed)";
deduplicated_req->preceding_opid = &leader_msg->id();
continue;
}
if (leader_msg->id().index() <= dedup_up_to_index) {
// If the index is uncommitted and below our match index, then it must be in the
// pendings set.
scoped_refptr<ConsensusRound> round =
pending_->GetPendingOpByIndexOrNull(leader_msg->id().index());
DCHECK(round) << "Could not find op with index " << leader_msg->id().index()
<< " in pending set. committed= " << last_committed_index
<< " dedup=" << dedup_up_to_index;
// If the OpIds match, i.e. if they have the same term and id, then this is just
// duplicate, we skip...
if (OpIdEquals(round->replicate_msg()->id(), leader_msg->id())) {
VLOG_WITH_PREFIX_UNLOCKED(2) << "Skipping op id " << leader_msg->id()
<< " (already replicated)";
deduplicated_req->preceding_opid = &leader_msg->id();
continue;
}
// ... otherwise we must adjust our match index, i.e. all messages from now on
// are "new"
dedup_up_to_index = leader_msg->id().index();
}
if (deduplicated_req->first_message_idx == - 1) {
deduplicated_req->first_message_idx = i;
}
deduplicated_req->messages.emplace_back(
make_scoped_refptr_replicate(new ReplicateMsg(*leader_msg)));
}
if (deduplicated_req->messages.size() != rpc_req->ops_size()) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Deduplicated request from leader. Original: "
<< rpc_req->preceding_id() << "->" << OpsRangeString(*rpc_req)
<< " Dedup: " << *deduplicated_req->preceding_opid << "->"
<< deduplicated_req->OpsRangeString();
}
}
Status RaftConsensus::HandleLeaderRequestTermUnlocked(const ConsensusRequestPB* request,
ConsensusResponsePB* response) {
DCHECK(lock_.is_locked());
// Do term checks first:
if (PREDICT_FALSE(request->caller_term() != CurrentTermUnlocked())) {
// If less, reject.
if (request->caller_term() < CurrentTermUnlocked()) {
string msg = Substitute("Rejecting Update request from peer $0 for earlier term $1. "
"Current term is $2. Ops: $3",
request->caller_uuid(),
request->caller_term(),
CurrentTermUnlocked(),
OpsRangeString(*request));
LOG_WITH_PREFIX_UNLOCKED(INFO) << msg;
FillConsensusResponseError(response,
ConsensusErrorPB::INVALID_TERM,
Status::IllegalState(msg));
return Status::OK();
}
RETURN_NOT_OK(HandleTermAdvanceUnlocked(request->caller_term()));
}
return Status::OK();
}
Status RaftConsensus::EnforceLogMatchingPropertyMatchesUnlocked(const LeaderRequest& req,
ConsensusResponsePB* response) {
DCHECK(lock_.is_locked());
bool term_mismatch;
if (pending_->IsOpCommittedOrPending(*req.preceding_opid, &term_mismatch)) {
return Status::OK();
}
string error_msg = Substitute(
"Log matching property violated."
" Preceding OpId in replica: $0. Preceding OpId from leader: $1. ($2 mismatch)",
SecureShortDebugString(queue_->GetLastOpIdInLog()),
SecureShortDebugString(*req.preceding_opid),
term_mismatch ? "term" : "index");
FillConsensusResponseError(response,
ConsensusErrorPB::PRECEDING_ENTRY_DIDNT_MATCH,
Status::IllegalState(error_msg));
// Adding a check to eliminate an unnecessary log message in the
// scenario where this is the first message from the Leader of a new tablet.
if (!OpIdEquals(MakeOpId(1,1), *req.preceding_opid)) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Refusing update from remote peer "
<< req.leader_uuid << ": " << error_msg;
}
// If the terms mismatch we abort down to the index before the leader's preceding,
// since we know that is the last opid that has a chance of not being overwritten.
// Aborting preemptively here avoids us reporting a last received index that is
// possibly higher than the leader's causing an avoidable cache miss on the leader's
// queue.
//
// TODO: this isn't just an optimization! if we comment this out, we get
// failures on raft_consensus-itest a couple percent of the time! Should investigate
// why this is actually critical to do here, as opposed to just on requests that
// append some ops.
if (term_mismatch) {
TruncateAndAbortOpsAfterUnlocked(req.preceding_opid->index() - 1);
}
return Status::OK();
}
void RaftConsensus::TruncateAndAbortOpsAfterUnlocked(int64_t truncate_after_index) {
DCHECK(lock_.is_locked());
pending_->AbortOpsAfter(truncate_after_index);
queue_->TruncateOpsAfter(truncate_after_index);
}
Status RaftConsensus::CheckLeaderRequestUnlocked(const ConsensusRequestPB* request,
ConsensusResponsePB* response,
LeaderRequest* deduped_req) {
DCHECK(lock_.is_locked());
if (PREDICT_FALSE(request->has_deprecated_committed_index() ||
!request->has_all_replicated_index())) {
return Status::InvalidArgument("Leader appears to be running an earlier version "
"of Kudu. Please shut down and upgrade all servers "
"before restarting.");
}
DeduplicateLeaderRequestUnlocked(request, deduped_req);
// This is an additional check for KUDU-639 that makes sure the message's index
// and term are in the right sequence in the request, after we've deduplicated
// them. We do this before we change any of the internal state.
//
// TODO move this to raft_consensus-state or whatever we transform that into.
// We should be able to do this check for each append, but right now the way
// we initialize raft_consensus-state is preventing us from doing so.
const OpId* prev = deduped_req->preceding_opid;
for (const ReplicateRefPtr& message : deduped_req->messages) {
Status s = PendingRounds::CheckOpInSequence(*prev, message->get()->id());
if (PREDICT_FALSE(!s.ok())) {
LOG(ERROR) << "Leader request contained out-of-sequence messages. Status: "
<< s.ToString() << ". Leader Request: " << SecureShortDebugString(*request);
return s;
}
prev = &message->get()->id();
}
RETURN_NOT_OK(HandleLeaderRequestTermUnlocked(request, response));
if (response->status().has_error()) {
return Status::OK();
}
RETURN_NOT_OK(EnforceLogMatchingPropertyMatchesUnlocked(*deduped_req, response));
if (response->status().has_error()) {
return Status::OK();
}
// If the first of the messages to apply is not in our log, either it follows the last
// received message or it replaces some in-flight.
if (!deduped_req->messages.empty()) {
bool term_mismatch;
CHECK(!pending_->IsOpCommittedOrPending(deduped_req->messages[0]->get()->id(), &term_mismatch));
// If the index is in our log but the terms are not the same abort down to the leader's
// preceding id.
if (term_mismatch) {
TruncateAndAbortOpsAfterUnlocked(deduped_req->preceding_opid->index());
}
}
// If all of the above logic was successful then we can consider this to be
// the effective leader of the configuration. If they are not currently marked as
// the leader locally, mark them as leader now.
const string& caller_uuid = request->caller_uuid();
if (PREDICT_FALSE(HasLeaderUnlocked() &&
GetLeaderUuidUnlocked() != caller_uuid)) {
LOG_WITH_PREFIX_UNLOCKED(FATAL) << "Unexpected new leader in same term! "
<< "Existing leader UUID: " << GetLeaderUuidUnlocked() << ", "
<< "new leader UUID: " << caller_uuid;
}
if (PREDICT_FALSE(!HasLeaderUnlocked())) {
SetLeaderUuidUnlocked(request->caller_uuid());
}
return Status::OK();
}
Status RaftConsensus::UpdateReplica(const ConsensusRequestPB* request,
ConsensusResponsePB* response) {
TRACE_EVENT2("consensus", "RaftConsensus::UpdateReplica",
"peer", peer_uuid(),
"tablet", options_.tablet_id);
Synchronizer log_synchronizer;
StatusCallback sync_status_cb = log_synchronizer.AsStatusCallback();
// The ordering of the following operations is crucial, read on for details.
//
// The main requirements explained in more detail below are:
//
// 1) We must enqueue the prepares before we write to our local log.
// 2) If we were able to enqueue a prepare then we must be able to log it.
// 3) If we fail to enqueue a prepare, we must not attempt to enqueue any
// later-indexed prepare or apply.
//
// See below for detailed rationale.
//
// The steps are:
//
// 0 - Split/Dedup
//
// We split the operations into replicates and commits and make sure that we don't
// don't do anything on operations we've already received in a previous call.
// This essentially makes this method idempotent.
//
// 1 - We mark as many pending ops as committed as we can.
//
// We may have some pending ops that, according to the leader, are now
// committed. We Apply them early, because:
// - Soon (step 2) we may reject the call due to excessive memory pressure. One
// way to relieve the pressure is by flushing the MRS, and applying these
// ops may unblock an in-flight Flush().
// - The Apply and subsequent Prepares (step 2) can take place concurrently.
//
// 2 - We enqueue the Prepare of the ops.
//
// The actual prepares are enqueued in order but happen asynchronously so we don't
// have decoding/acquiring locks on the critical path.
//
// We need to do this now for a number of reasons:
// - Prepares, by themselves, are inconsequential, i.e. they do not mutate the
// state machine so, were we to crash afterwards, having the prepares in-flight
// won't hurt.
// - Prepares depend on factors external to consensus (the op drivers and
// the TabletReplica) so if for some reason they cannot be enqueued we must know
// before we try write them to the WAL. Once enqueued, we assume that prepare will
// always succeed on a replica op (because the leader already prepared them
// successfully, and thus we know they are valid).
// - The prepares corresponding to every operation that was logged must be in-flight
// first. This because should we need to abort certain ops (say a new leader
// says they are not committed) we need to have those prepares in-flight so that
// the ops can be continued (in the abort path).
// - Failure to enqueue prepares is OK, we can continue and let the leader know that
// we only went so far. The leader will re-send the remaining messages.
// - Prepares represent new ops, and ops consume memory. Thus, if the
// overall memory pressure on the server is too high, we will reject the prepares.
//
// 3 - We enqueue the writes to the WAL.
//
// We enqueue writes to the WAL, but only the operations that were successfully
// enqueued for prepare (for the reasons introduced above). This means that even
// if a prepare fails to enqueue, if any of the previous prepares were successfully
// submitted they must be written to the WAL.
// If writing to the WAL fails, we're in an inconsistent state and we crash. In this
// case, no one will ever know of the ops we previously prepared so those are
// inconsequential.
//
// 4 - We mark the ops as committed.
//
// For each op which has been committed by the leader, we update the
// op state to reflect that. If the logging has already succeeded for that
// op, this will trigger the Apply phase. Otherwise, Apply will be triggered
// when the logging completes. In both cases the Apply phase executes asynchronously.
// This must, of course, happen after the prepares have been triggered as the same batch
// can both replicate/prepare and commit/apply an operation.
//
// Currently, if a prepare failed to enqueue we still trigger all applies for operations
// with an id lower than it (if we have them). This is important now as the leader will
// not re-send those commit messages. This will be moot when we move to the commit
// commitIndex way of doing things as we can simply ignore the applies as we know
// they will be triggered with the next successful batch.
//
// 5 - We wait for the writes to be durable.
//
// Before replying to the leader we wait for the writes to be durable. We then
// just update the last replicated watermark and respond.
//
// TODO - These failure scenarios need to be exercised in an unit
// test. Moreover we need to add more fault injection spots (well that
// and actually use the) for each of these steps.
// This will be done in a follow up patch.
TRACE("Updating replica for $0 ops", request->ops_size());
// The deduplicated request.
LeaderRequest deduped_req;
auto& messages = deduped_req.messages;
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
RETURN_NOT_OK(CheckRunningUnlocked());
if (!cmeta_->IsMemberInConfig(peer_uuid(), ACTIVE_CONFIG)) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Allowing update even though not a member of the config";
}
deduped_req.leader_uuid = request->caller_uuid();
RETURN_NOT_OK(CheckLeaderRequestUnlocked(request, response, &deduped_req));
if (response->status().has_error()) {
// We had an error, like an invalid term, we still fill the response.
FillConsensusResponseOKUnlocked(response);
return Status::OK();
}
// As soon as we decide to accept the message:
// * snooze the failure detector
// * prohibit voting for anyone for the minimum election timeout
// We are guaranteed to be acting as a FOLLOWER at this point by the above
// sanity check.
SnoozeFailureDetector();
WithholdVotes();
last_leader_communication_time_micros_ = GetMonoTimeMicros();
// Reset the 'failed_elections_since_stable_leader' metric now that we've
// accepted an update from the established leader. This is done in addition
// to the reset of the value in SetLeaderUuidUnlocked() because there is
// a potential race between resetting the failed elections count in
// SetLeaderUuidUnlocked() and incrementing after a failed election
// if another replica was elected leader in an election concurrent with
// the one called by this replica.
failed_elections_since_stable_leader_ = 0;
num_failed_elections_metric_->set_value(failed_elections_since_stable_leader_);
// We update the lag metrics here in addition to after appending to the queue so the
// metrics get updated even when the operation is rejected.
queue_->UpdateLastIndexAppendedToLeader(request->last_idx_appended_to_leader());
// 1 - Early commit pending (and committed) ops
// What should we commit?
// 1. As many pending ops as we can, except...
// 2. ...if we commit beyond the preceding index, we'd regress KUDU-639, and...
// 3. ...the leader's committed index is always our upper bound.
const int64_t early_apply_up_to = std::min({
pending_->GetLastPendingOpOpId().index(),
deduped_req.preceding_opid->index(),
request->committed_index()});
VLOG_WITH_PREFIX_UNLOCKED(1) << "Early marking committed up to " << early_apply_up_to
<< ", Last pending opid index: "
<< pending_->GetLastPendingOpOpId().index()
<< ", preceding opid index: "
<< deduped_req.preceding_opid->index()
<< ", requested index: " << request->committed_index();
TRACE("Early marking committed up to index $0", early_apply_up_to);
CHECK_OK(pending_->AdvanceCommittedIndex(early_apply_up_to));
// 2 - Enqueue the prepares
TRACE("Triggering prepare for $0 ops", messages.size());
if (PREDICT_TRUE(!messages.empty())) {
// This request contains at least one message, and is likely to increase
// our memory pressure.
double capacity_pct;
if (process_memory::SoftLimitExceeded(&capacity_pct)) {
if (follower_memory_pressure_rejections_) follower_memory_pressure_rejections_->Increment();
string msg = StringPrintf(
"Soft memory limit exceeded (at %.2f%% of capacity)",
capacity_pct);
if (capacity_pct >= FLAGS_memory_limit_warn_threshold_percentage) {
KLOG_EVERY_N_SECS(WARNING, 1) << "Rejecting consensus request: " << msg
<< THROTTLE_MSG;
} else {
KLOG_EVERY_N_SECS(INFO, 1) << "Rejecting consensus request: " << msg
<< THROTTLE_MSG;
}
return Status::ServiceUnavailable(msg);
}
}
Status prepare_status;
auto iter = messages.begin();
while (iter != messages.end()) {
prepare_status = StartFollowerOpUnlocked(*iter);
if (PREDICT_FALSE(!prepare_status.ok())) {
break;
}
// TODO(dralves) Without leader leases this shouldn't be allowed to fail.
// Once we have that functionality we'll have to revisit this.
CHECK_OK(time_manager_->MessageReceivedFromLeader(*(*iter)->get()));
++iter;
}
// If we stopped before reaching the end we failed to prepare some message(s) and need
// to perform cleanup, namely trimming deduped_req.messages to only contain the messages
// that were actually prepared, and deleting the other ones since we've taken ownership
// when we first deduped.
if (iter != messages.end()) {
LOG_WITH_PREFIX_UNLOCKED(WARNING) << Substitute(
"Could not prepare op '$0' and following $1 ops. "
"Status for this op: $2",
(*iter)->get()->id().ShortDebugString(),
std::distance(iter, messages.end()) - 1,
prepare_status.ToString());
iter = messages.erase(iter, messages.end());
// If this is empty, it means we couldn't prepare a single de-duped message. There is nothing
// else we can do. The leader will detect this and retry later.
if (messages.empty()) {
string msg = Substitute("Rejecting Update request from peer $0 for term $1. "
"Could not prepare a single op due to: $2",
request->caller_uuid(),
request->caller_term(),
prepare_status.ToString());
LOG_WITH_PREFIX_UNLOCKED(INFO) << msg;
FillConsensusResponseError(response, ConsensusErrorPB::CANNOT_PREPARE,
Status::IllegalState(msg));
FillConsensusResponseOKUnlocked(response);
return Status::OK();
}
}
// All ops that are going to be prepared were started, advance the safe timestamp.
// TODO(dralves) This is only correct because the queue only sets safe time when the request is
// an empty heartbeat. If we actually start setting this on a consensus request along with
// actual messages we need to be careful to ignore it if any of the messages fails to prepare.
if (request->has_safe_timestamp()) {
time_manager_->AdvanceSafeTime(Timestamp(request->safe_timestamp()));
}
OpId last_from_leader;
// 3 - Enqueue the writes.
// Now that we've triggered the prepares enqueue the operations to be written
// to the WAL.
if (PREDICT_TRUE(!messages.empty())) {
last_from_leader = messages.back()->get()->id();
// Trigger the log append asap, if fsync() is on this might take a while
// and we can't reply until this is done.
//
// Since we've prepared, we need to be able to append (or we risk trying to apply
// later something that wasn't logged). We crash if we can't.
CHECK_OK(queue_->AppendOperations(messages, sync_status_cb));
} else {
last_from_leader = *deduped_req.preceding_opid;
}
// 4 - Mark ops as committed
// Choose the last operation to be applied. This will either be 'committed_index', if
// no prepare enqueuing failed, or the minimum between 'committed_index' and the id of
// the last successfully enqueued prepare, if some prepare failed to enqueue.
int64_t apply_up_to;
if (last_from_leader.index() < request->committed_index()) {
// we should never apply anything later than what we received in this request
apply_up_to = last_from_leader.index();
VLOG_WITH_PREFIX_UNLOCKED(2) << "Received commit index "
<< request->committed_index() << " from the leader but only"
<< " marked up to " << apply_up_to << " as committed.";
} else {
apply_up_to = request->committed_index();
}
VLOG_WITH_PREFIX_UNLOCKED(1) << "Marking committed up to " << apply_up_to;
TRACE("Marking committed up to $0", apply_up_to);
CHECK_OK(pending_->AdvanceCommittedIndex(apply_up_to));
queue_->UpdateFollowerWatermarks(apply_up_to, request->all_replicated_index());
// If any messages failed to be started locally, then we already have removed them
// from 'deduped_req' at this point. So, 'last_from_leader' is the last one that
// we might apply.
last_received_cur_leader_ = last_from_leader;
// Fill the response with the current state. We will not mutate anymore state until
// we actually reply to the leader, we'll just wait for the messages to be durable.
FillConsensusResponseOKUnlocked(response);
}
// Release the lock while we wait for the log append to finish so that commits can go through.
// We'll re-acquire it before we update the state again.
// Update the last replicated op id
if (!messages.empty()) {
// 5 - We wait for the writes to be durable.
// Note that this is safe because dist consensus now only supports a single outstanding
// request at a time and this way we can allow commits to proceed while we wait.
TRACE("Waiting on the replicates to finish logging");
TRACE_EVENT0("consensus", "Wait for log");
Status s;
do {
s = log_synchronizer.WaitFor(
MonoDelta::FromMilliseconds(FLAGS_raft_heartbeat_interval_ms));
// If just waiting for our log append to finish lets snooze the timer.
// We don't want to fire leader election nor accept vote requests because
// we're still processing the Raft message from the leader,
// waiting on our own log.
if (s.IsTimedOut()) {
SnoozeFailureDetector();
WithholdVotes();
}
} while (s.IsTimedOut());
RETURN_NOT_OK(s);
TRACE("finished");
}
VLOG_WITH_PREFIX(2) << "Replica updated. " << ToString()
<< ". Request: " << SecureShortDebugString(*request);
TRACE("UpdateReplicas() finished");
return Status::OK();
}
void RaftConsensus::FillConsensusResponseOKUnlocked(ConsensusResponsePB* response) {
DCHECK(lock_.is_locked());
TRACE("Filling consensus response to leader.");
response->set_responder_term(CurrentTermUnlocked());
response->mutable_status()->mutable_last_received()->CopyFrom(
queue_->GetLastOpIdInLog());
response->mutable_status()->mutable_last_received_current_leader()->CopyFrom(
last_received_cur_leader_);
response->mutable_status()->set_last_committed_idx(
queue_->GetCommittedIndex());
if (PREDICT_TRUE(server_ctx_.quiescing) && server_ctx_.quiescing->load()) {
response->set_server_quiescing(true);
}
}
void RaftConsensus::FillConsensusResponseError(ConsensusResponsePB* response,
ConsensusErrorPB::Code error_code,
const Status& status) {
ConsensusErrorPB* error = response->mutable_status()->mutable_error();
error->set_code(error_code);
StatusToPB(status, error->mutable_status());
}
Status RaftConsensus::RequestVote(const VoteRequestPB* request,
TabletVotingState tablet_voting_state,
VoteResponsePB* response) {
TRACE_EVENT2("consensus", "RaftConsensus::RequestVote",
"peer", peer_uuid(),
"tablet", options_.tablet_id);
response->set_responder_uuid(peer_uuid());
// If we've heard recently from the leader, then we should ignore the request.
// It might be from a "disruptive" server. This could happen in a few cases:
//
// 1) Network partitions
// If the leader can talk to a majority of the nodes, but is partitioned from a
// bad node, the bad node's failure detector will trigger. If the bad node is
// able to reach other nodes in the cluster, it will continuously trigger elections.
//
// 2) An abandoned node
// It's possible that a node has fallen behind the log GC mark of the leader. In that
// case, the leader will stop sending it requests. Eventually, the the configuration
// will change to eject the abandoned node, but until that point, we don't want the
// abandoned follower to disturb the other nodes.
//
// 3) Other dynamic scenarios with a stale former leader
// This is a generalization of the case 1. It's possible that a stale former
// leader detects it's not a leader anymore at some point, but a majority
// of replicas has elected a new leader already.
//
// See also https://ramcloud.stanford.edu/~ongaro/thesis.pdf
// section 4.2.3.
if (PREDICT_TRUE(!request->ignore_live_leader()) &&
MonoTime::Now() < withhold_votes_until_) {
return RequestVoteRespondLeaderIsAlive(request, response);
}
// We must acquire the update lock in order to ensure that this vote action
// takes place between requests.
// Lock ordering: update_lock_ must be acquired before lock_.
std::unique_lock<simple_spinlock> update_guard(update_lock_, std::defer_lock);
if (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {
update_guard.try_lock();
} else {
// If failure detection is not enabled, then we can't just reject the vote,
// because there will be no automatic retry later. So, block for the lock.
update_guard.lock();
}
if (!update_guard.owns_lock()) {
// There is another vote or update concurrent with the vote. In that case, that
// other request is likely to reset the timer, and we'll end up just voting
// "NO" after waiting. To avoid starving RPC handlers and causing cascading
// timeouts, just vote a quick NO.
return RequestVoteRespondIsBusy(request, response);
}
// Acquire the replica state lock so we can read / modify the consensus state.
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
// Ensure our lifecycle state is compatible with voting.
// If RaftConsensus is running, we use the latest OpId from the WAL to vote.
// Otherwise, we must be voting while tombstoned.
OpId local_last_logged_opid;
const State state = state_;
switch (state) {
case kShutdown:
return Status::IllegalState("cannot vote while shut down");
case kRunning:
// Note: it is (theoretically) possible for 'tombstone_last_logged_opid'
// to be passed in and by the time we reach here the state is kRunning.
// That may occur when a vote request comes in at the end of a tablet
// copy and then tablet bootstrap completes quickly. In that case, we
// ignore the passed-in value and use the latest OpId from our queue.
local_last_logged_opid = queue_->GetLastOpIdInLog();
break;
default:
if (!tablet_voting_state.tombstone_last_logged_opid_) {
return Status::IllegalState("must be running to vote when last-logged opid is not known");
}
if (!FLAGS_raft_enable_tombstoned_voting) {
return Status::IllegalState("must be running to vote when tombstoned voting is disabled");
}
local_last_logged_opid = *(tablet_voting_state.tombstone_last_logged_opid_);
if (tablet_voting_state.data_state_ == tablet::TABLET_DATA_COPYING) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "voting while copying based on last-logged opid "
<< local_last_logged_opid;
} else if (tablet_voting_state.data_state_ == tablet::TABLET_DATA_TOMBSTONED) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "voting while tombstoned based on last-logged opid "
<< local_last_logged_opid;
}
break;
}
DCHECK(local_last_logged_opid.IsInitialized());
// If the node is not in the configuration, allow the vote (this is required by Raft)
// but log an informational message anyway.
if (!cmeta_->IsMemberInConfig(request->candidate_uuid(), ACTIVE_CONFIG)) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Handling vote request from an unknown peer "
<< request->candidate_uuid();
}
// Recheck the heard-from-the-leader condition. There is a slight chance
// that a heartbeat from the leader replica registers after the first check
// in the very beginning of this method and before lock_ is acquired. This
// extra check costs a little, but it helps in avoiding extra election rounds
// and disruptive transfers of the replica leadership.
if (PREDICT_TRUE(!request->ignore_live_leader()) &&
MonoTime::Now() < withhold_votes_until_) {
return RequestVoteRespondLeaderIsAlive(request, response);
}
// Candidate is running behind.
if (request->candidate_term() < CurrentTermUnlocked()) {
return RequestVoteRespondInvalidTerm(request, response);
}
// We already voted this term.
if (request->candidate_term() == CurrentTermUnlocked() &&
HasVotedCurrentTermUnlocked()) {
// Already voted for the same candidate in the current term.
if (GetVotedForCurrentTermUnlocked() == request->candidate_uuid()) {
return RequestVoteRespondVoteAlreadyGranted(request, response);
}
// Voted for someone else in current term.
return RequestVoteRespondAlreadyVotedForOther(request, response);
}
// Candidate must have last-logged OpId at least as large as our own to get
// our vote.
bool vote_yes = !OpIdLessThan(request->candidate_status().last_received(),
local_last_logged_opid);
// Record the term advancement if necessary. We don't do so in the case of
// pre-elections because it's possible that the node who called the pre-election
// has actually now successfully become leader of the prior term, in which case
// bumping our term here would disrupt it.
if (!request->is_pre_election() &&
request->candidate_term() > CurrentTermUnlocked()) {
// If we are going to vote for this peer, then we will flush the consensus metadata
// to disk below when we record the vote, and we can skip flushing the term advancement
// to disk here.
auto flush = vote_yes ? SKIP_FLUSH_TO_DISK : FLUSH_TO_DISK;
RETURN_NOT_OK_PREPEND(HandleTermAdvanceUnlocked(request->candidate_term(), flush),
Substitute("Could not step down in RequestVote. Current term: $0, candidate term: $1",
CurrentTermUnlocked(), request->candidate_term()));
}
if (!vote_yes) {
return RequestVoteRespondLastOpIdTooOld(local_last_logged_opid, request, response);
}
// Passed all our checks. Vote granted.
return RequestVoteRespondVoteGranted(request, response);
}
Status RaftConsensus::ChangeConfig(const ChangeConfigRequestPB& req,
StatusCallback client_cb,
boost::optional<TabletServerErrorPB::Code>* error_code) {
TRACE_EVENT2("consensus", "RaftConsensus::ChangeConfig",
"peer", peer_uuid(),
"tablet", options_.tablet_id);
BulkChangeConfigRequestPB bulk_req;
*bulk_req.mutable_tablet_id() = req.tablet_id();
if (req.has_dest_uuid()) {
*bulk_req.mutable_dest_uuid() = req.dest_uuid();
}
if (req.has_cas_config_opid_index()) {
bulk_req.set_cas_config_opid_index(req.cas_config_opid_index());
}
auto* change = bulk_req.add_config_changes();
if (req.has_type()) {
change->set_type(req.type());
}
if (req.has_server()) {
*change->mutable_peer() = req.server();
}
return BulkChangeConfig(bulk_req, std::move(client_cb), error_code);
}
Status RaftConsensus::BulkChangeConfig(const BulkChangeConfigRequestPB& req,
StatusCallback client_cb,
boost::optional<TabletServerErrorPB::Code>* error_code) {
TRACE_EVENT2("consensus", "RaftConsensus::BulkChangeConfig",
"peer", peer_uuid(),
"tablet", options_.tablet_id);
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
RETURN_NOT_OK(CheckRunningUnlocked());
RETURN_NOT_OK(CheckActiveLeaderUnlocked());
RETURN_NOT_OK(CheckNoConfigChangePendingUnlocked());
// We are required by Raft to reject config change operations until we have
// committed at least one operation in our current term as leader.
// See https://groups.google.com/forum/#!topic/raft-dev/t4xj6dJTP6E
if (!queue_->IsCommittedIndexInCurrentTerm()) {
return Status::IllegalState("Leader has not yet committed an operation in its own term");
}
const RaftConfigPB committed_config = cmeta_->CommittedConfig();
// Support atomic ChangeConfig requests.
if (req.has_cas_config_opid_index()) {
if (committed_config.opid_index() != req.cas_config_opid_index()) {
*error_code = TabletServerErrorPB::CAS_FAILED;
return Status::IllegalState(Substitute("Request specified cas_config_opid_index "
"of $0 but the committed config has opid_index "
"of $1",
req.cas_config_opid_index(),
committed_config.opid_index()));
}
}
// 'new_config' will be modified in-place and validated before being used
// as the new Raft configuration.
RaftConfigPB new_config = committed_config;
// Enforce the "one by one" config change rules, even with the bulk API.
// Keep track of total voters added, including non-voters promoted to
// voters, and removed, including voters demoted to non-voters.
int num_voters_modified = 0;
// A record of the peers being modified so that we can enforce only one
// change per peer per request.
unordered_set<string> peers_modified;
for (const auto& item : req.config_changes()) {
if (PREDICT_FALSE(!item.has_type())) {
*error_code = TabletServerErrorPB::INVALID_CONFIG;
return Status::InvalidArgument("Must specify 'type' argument",
SecureShortDebugString(req));
}
if (PREDICT_FALSE(!item.has_peer())) {
*error_code = TabletServerErrorPB::INVALID_CONFIG;
return Status::InvalidArgument("Must specify 'peer' argument",
SecureShortDebugString(req));
}
ChangeConfigType type = item.type();
const RaftPeerPB& peer = item.peer();
if (PREDICT_FALSE(!peer.has_permanent_uuid())) {
return Status::InvalidArgument("peer must have permanent_uuid specified",
SecureShortDebugString(req));
}
if (!InsertIfNotPresent(&peers_modified, peer.permanent_uuid())) {
return Status::InvalidArgument(
Substitute("only one change allowed per peer: peer $0 appears more "
"than once in the config change request",
peer.permanent_uuid()),
SecureShortDebugString(req));
}
const string& server_uuid = peer.permanent_uuid();
switch (type) {
case ADD_PEER:
// Ensure the peer we are adding is not already a member of the configuration.
if (IsRaftConfigMember(server_uuid, committed_config)) {
return Status::InvalidArgument(
Substitute("Server with UUID $0 is already a member of the config. RaftConfig: $1",
server_uuid, SecureShortDebugString(committed_config)));
}
if (!peer.has_member_type()) {
return Status::InvalidArgument("peer must have member_type specified",
SecureShortDebugString(req));
}
if (!peer.has_last_known_addr()) {
return Status::InvalidArgument("peer must have last_known_addr specified",
SecureShortDebugString(req));
}
if (peer.member_type() == RaftPeerPB::VOTER) {
num_voters_modified++;
}
*new_config.add_peers() = peer;
break;
case REMOVE_PEER:
if (server_uuid == peer_uuid()) {
return Status::InvalidArgument(
Substitute("Cannot remove peer $0 from the config because it is the leader. "
"Force another leader to be elected to remove this peer. "
"Consensus state: $1",
server_uuid,
SecureShortDebugString(cmeta_->ToConsensusStatePB())));
}
if (!RemoveFromRaftConfig(&new_config, server_uuid)) {
return Status::NotFound(
Substitute("Server with UUID $0 not a member of the config. RaftConfig: $1",
server_uuid, SecureShortDebugString(committed_config)));
}
if (IsRaftConfigVoter(server_uuid, committed_config)) {
num_voters_modified++;
}
break;
case MODIFY_PEER: {
RaftPeerPB* modified_peer;
RETURN_NOT_OK(GetRaftConfigMember(&new_config, server_uuid, &modified_peer));
const RaftPeerPB orig_peer(*modified_peer);
// Override 'member_type' and items within 'attrs' only if they are
// explicitly passed in the request. At least one field must be
// modified to be a valid request.
if (peer.has_member_type() && peer.member_type() != modified_peer->member_type()) {
if (modified_peer->member_type() == RaftPeerPB::VOTER ||
peer.member_type() == RaftPeerPB::VOTER) {
// This is a 'member_type' change involving a VOTER, i.e. a
// promotion or demotion.
num_voters_modified++;
}
// A leader must be forced to step down before demoting it.
if (server_uuid == peer_uuid()) {
return Status::InvalidArgument(
Substitute("Cannot modify member type of peer $0 because it is the leader. "
"Cause another leader to be elected to modify this peer. "
"Consensus state: $1",
server_uuid,
SecureShortDebugString(cmeta_->ToConsensusStatePB())));
}
modified_peer->set_member_type(peer.member_type());
}
if (peer.attrs().has_promote()) {
modified_peer->mutable_attrs()->set_promote(peer.attrs().promote());
}
if (peer.attrs().has_replace()) {
modified_peer->mutable_attrs()->set_replace(peer.attrs().replace());
}
// Ensure that MODIFY_PEER actually modified something.
if (MessageDifferencer::Equals(orig_peer, *modified_peer)) {
return Status::InvalidArgument("must modify a field when calling MODIFY_PEER");
}
break;
}
default:
return Status::NotSupported(Substitute(
"$0: unsupported type of configuration change",
ChangeConfigType_Name(type)));
}
}
// Don't allow no-op config changes to be committed.
if (MessageDifferencer::Equals(committed_config, new_config)) {
return Status::InvalidArgument("requested configuration change does not "
"actually modify the config",
SecureShortDebugString(req));
}
// Ensure this wasn't an illegal bulk change.
if (num_voters_modified > 1) {
return Status::InvalidArgument("it is not safe to modify the VOTER status "
"of more than one peer at a time",
SecureShortDebugString(req));
}
// We'll assign a new opid_index to this config change.
new_config.clear_opid_index();
RETURN_NOT_OK(ReplicateConfigChangeUnlocked(
committed_config, std::move(new_config),
[this, client_cb](const Status& s) {
this->MarkDirtyOnSuccess("Config change replication complete",
client_cb, s);
}));
} // Release lock before signaling request.
peer_manager_->SignalRequest();
return Status::OK();
}
Status RaftConsensus::UnsafeChangeConfig(
const UnsafeChangeConfigRequestPB& req,
boost::optional<tserver::TabletServerErrorPB::Code>* error_code) {
if (PREDICT_FALSE(!req.has_new_config())) {
*error_code = TabletServerErrorPB::INVALID_CONFIG;
return Status::InvalidArgument("Request must contain 'new_config' argument "
"to UnsafeChangeConfig()", SecureShortDebugString(req));
}
if (PREDICT_FALSE(!req.has_caller_id())) {
*error_code = TabletServerErrorPB::INVALID_CONFIG;
return Status::InvalidArgument("Must specify 'caller_id' argument to UnsafeChangeConfig()",
SecureShortDebugString(req));
}
// Grab the committed config and current term on this node.
int64_t current_term;
RaftConfigPB committed_config;
int64_t all_replicated_index;
int64_t last_committed_index;
OpId preceding_opid;
uint64_t msg_timestamp;
{
// Take the snapshot of the replica state and queue state so that
// we can stick them in the consensus update request later.
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
current_term = CurrentTermUnlocked();
committed_config = cmeta_->CommittedConfig();
if (cmeta_->has_pending_config()) {
LOG_WITH_PREFIX_UNLOCKED(WARNING)
<< "Replica has a pending config, but the new config "
<< "will be unsafely changed anyway. "
<< "Currently pending config on the node: "
<< SecureShortDebugString(cmeta_->PendingConfig());
}
all_replicated_index = queue_->GetAllReplicatedIndex();
last_committed_index = queue_->GetCommittedIndex();
preceding_opid = queue_->GetLastOpIdInLog();
msg_timestamp = time_manager_->GetSerialTimestamp().value();
}
// Validate that passed replica uuids are part of the committed config
// on this node. This allows a manual recovery tool to only have to specify
// the uuid of each replica in the new config without having to know the
// addresses of each server (since we can get the address information from
// the committed config). Additionally, only a subset of the committed config
// is required for typical cluster repair scenarios.
std::unordered_set<string> retained_peer_uuids;
const RaftConfigPB& config = req.new_config();
for (const RaftPeerPB& new_peer : config.peers()) {
const string& peer_uuid = new_peer.permanent_uuid();
retained_peer_uuids.insert(peer_uuid);
if (!IsRaftConfigMember(peer_uuid, committed_config)) {
*error_code = TabletServerErrorPB::INVALID_CONFIG;
return Status::InvalidArgument(Substitute("Peer with uuid $0 is not in the committed "
"config on this replica, rejecting the "
"unsafe config change request for tablet $1. "
"Committed config: $2",
peer_uuid, req.tablet_id(),
SecureShortDebugString(committed_config)));
}
}
RaftConfigPB new_config = committed_config;
for (const auto& peer : committed_config.peers()) {
const string& peer_uuid = peer.permanent_uuid();
if (!ContainsKey(retained_peer_uuids, peer_uuid)) {
CHECK(RemoveFromRaftConfig(&new_config, peer_uuid));
}
}
// Check that local peer is part of the new config and is a VOTER.
// Although it is valid for a local replica to not have itself
// in the committed config, it is rare and a replica without itself
// in the latest config is definitely not caught up with the latest leader's log.
if (!IsRaftConfigVoter(peer_uuid(), new_config)) {
*error_code = TabletServerErrorPB::INVALID_CONFIG;
return Status::InvalidArgument(Substitute("Local replica uuid $0 is not "
"a VOTER in the new config, "
"rejecting the unsafe config "
"change request for tablet $1. "
"Rejected config: $2" ,
peer_uuid(), req.tablet_id(),
SecureShortDebugString(new_config)));
}
new_config.set_unsafe_config_change(true);
int64_t replicate_opid_index = preceding_opid.index() + 1;
new_config.set_opid_index(replicate_opid_index);
// Sanity check the new config. 'type' is irrelevant here.
Status s = VerifyRaftConfig(new_config);
if (!s.ok()) {
*error_code = TabletServerErrorPB::INVALID_CONFIG;
return Status::InvalidArgument(Substitute("The resulting new config for tablet $0 "
"from passed parameters has failed raft "
"config sanity check: $1",
req.tablet_id(), s.ToString()));
}
// Prepare the consensus request as if the request is being generated
// from a different leader.
ConsensusRequestPB consensus_req;
consensus_req.set_caller_uuid(req.caller_id());
// Bumping up the term for the consensus request being generated.
// This makes this request appear to come from a new leader that
// the local replica doesn't know about yet. If the local replica
// happens to be the leader, this will cause it to step down.
const int64_t new_term = current_term + 1;
consensus_req.set_caller_term(new_term);
consensus_req.mutable_preceding_id()->CopyFrom(preceding_opid);
consensus_req.set_committed_index(last_committed_index);
consensus_req.set_all_replicated_index(all_replicated_index);
// Prepare the replicate msg to be replicated.
ReplicateMsg* replicate = consensus_req.add_ops();
ChangeConfigRecordPB* cc_req = replicate->mutable_change_config_record();
cc_req->set_tablet_id(req.tablet_id());
*cc_req->mutable_old_config() = committed_config;
*cc_req->mutable_new_config() = new_config;
OpId* id = replicate->mutable_id();
// Bumping up both the term and the opid_index from what's found in the log.
id->set_term(new_term);
id->set_index(replicate_opid_index);
replicate->set_op_type(CHANGE_CONFIG_OP);
replicate->set_timestamp(msg_timestamp);
VLOG_WITH_PREFIX(3) << "UnsafeChangeConfig: Generated consensus request: "
<< SecureShortDebugString(consensus_req);
LOG_WITH_PREFIX(WARNING)
<< "PROCEEDING WITH UNSAFE CONFIG CHANGE ON THIS SERVER, "
<< "COMMITTED CONFIG: " << SecureShortDebugString(committed_config)
<< "NEW CONFIG: " << SecureShortDebugString(new_config);
ConsensusResponsePB consensus_resp;
return Update(&consensus_req, &consensus_resp).AndThen([&consensus_resp]{
return consensus_resp.has_error()
? StatusFromPB(consensus_resp.error().status()) : Status::OK();
});
}
void RaftConsensus::Stop() {
TRACE_EVENT2("consensus", "RaftConsensus::Shutdown",
"peer", peer_uuid(),
"tablet", options_.tablet_id);
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
const State state = state_;
if (state == kStopping || state == kStopped || state == kShutdown) {
return;
}
// Transition to kStopping state.
SetStateUnlocked(kStopping);
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Raft consensus shutting down.";
}
// Close the peer manager.
if (peer_manager_) peer_manager_->Close();
// We must close the queue after we close the peers.
if (queue_) {
// If we were leader, decrement the number of leaders there are now.
if (queue_->IsInLeaderMode() && server_ctx_.num_leaders) {
server_ctx_.num_leaders->IncrementBy(-1);
}
queue_->Close();
}
{
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
if (pending_) {
CHECK_OK(pending_->CancelPendingOps());
}
SetStateUnlocked(kStopped);
// Clear leader status on Stop(), in case this replica was the leader. If
// we don't do this, the log messages still show this node as the leader.
// No need to sync it since it's not persistent state.
if (cmeta_) {
ClearLeaderUnlocked();
}
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Raft consensus is shut down!";
}
// If we were the leader, stop withholding votes.
auto max_time = MonoTime::Max();
withhold_votes_until_.compare_exchange_strong(max_time, MonoTime::Min());
// Shut down things that might acquire locks during destruction.
if (raft_pool_token_) raft_pool_token_->Shutdown();
if (failure_detector_) DisableFailureDetector();
}
void RaftConsensus::Shutdown() {
// Avoid taking locks if already shut down so we don't violate
// ThreadRestrictions assertions in the case where the RaftConsensus
// destructor runs on the reactor thread due to an election callback being
// the last outstanding reference.
if (shutdown_) {
return;
}
Stop();
{
LockGuard l(lock_);
SetStateUnlocked(kShutdown);
}
shutdown_ = true;
}
Status RaftConsensus::StartConsensusOnlyRoundUnlocked(const ReplicateRefPtr& msg) {
DCHECK(lock_.is_locked());
OperationType op_type = msg->get()->op_type();
CHECK(IsConsensusOnlyOperation(op_type))
<< "Expected a consensus-only op type, got " << OperationType_Name(op_type)
<< ": " << SecureShortDebugString(*msg->get());
VLOG_WITH_PREFIX_UNLOCKED(1) << "Starting consensus round: "
<< SecureShortDebugString(msg->get()->id());
auto client_cb = [this](const Status& s) {
this->MarkDirtyOnSuccess("Replicated consensus-only round",
&DoNothingStatusCB, s);
};
scoped_refptr<ConsensusRound> round(new ConsensusRound(this, msg));
auto* round_raw = round.get();
round->SetConsensusReplicatedCallback(
[this, round_raw, client_cb](const Status& s) {
this->NonTxRoundReplicationFinished(round_raw, client_cb, s);
});
return AddPendingOperationUnlocked(round);
}
Status RaftConsensus::AdvanceTermForTests(int64_t new_term) {
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
CHECK_OK(CheckRunningUnlocked());
return HandleTermAdvanceUnlocked(new_term);
}
string RaftConsensus::GetRequestVoteLogPrefixThreadSafe(const VoteRequestPB& request) const {
return Substitute("$0Leader $1election vote request",
LogPrefixThreadSafe(),
request.is_pre_election() ? "pre-" : "");
}
string RaftConsensus::GetRequestVoteLogPrefixUnlocked(const VoteRequestPB& request) const {
DCHECK(lock_.is_locked());
return Substitute("$0Leader $1election vote request",
LogPrefixUnlocked(),
request.is_pre_election() ? "pre-" : "");
}
void RaftConsensus::FillVoteResponseVoteGranted(VoteResponsePB* response) {
response->set_responder_term(CurrentTermUnlocked());
response->set_vote_granted(true);
}
void RaftConsensus::FillVoteResponseVoteDenied(
ConsensusErrorPB::Code error_code,
VoteResponsePB* response,
ResponderTermPolicy responder_term_policy) {
response->set_vote_granted(false);
response->mutable_consensus_error()->set_code(error_code);
if (responder_term_policy == ResponderTermPolicy::SET) {
response->set_responder_term(CurrentTermUnlocked());
}
}
Status RaftConsensus::RequestVoteRespondInvalidTerm(const VoteRequestPB* request,
VoteResponsePB* response) {
FillVoteResponseVoteDenied(ConsensusErrorPB::INVALID_TERM, response);
string msg = Substitute("$0: Denying vote to candidate $1 for earlier term $2. "
"Current term is $3.",
GetRequestVoteLogPrefixUnlocked(*request),
request->candidate_uuid(),
request->candidate_term(),
CurrentTermUnlocked());
LOG(INFO) << msg;
StatusToPB(Status::InvalidArgument(msg), response->mutable_consensus_error()->mutable_status());
return Status::OK();
}
Status RaftConsensus::RequestVoteRespondVoteAlreadyGranted(const VoteRequestPB* request,
VoteResponsePB* response) {
FillVoteResponseVoteGranted(response);
LOG(INFO) << Substitute("$0: Already granted yes vote for candidate $1 in term $2. "
"Re-sending same reply.",
GetRequestVoteLogPrefixUnlocked(*request),
request->candidate_uuid(),
request->candidate_term());
return Status::OK();
}
Status RaftConsensus::RequestVoteRespondAlreadyVotedForOther(const VoteRequestPB* request,
VoteResponsePB* response) {
FillVoteResponseVoteDenied(ConsensusErrorPB::ALREADY_VOTED, response);
string msg = Substitute("$0: Denying vote to candidate $1 in current term $2: "
"Already voted for candidate $3 in this term.",
GetRequestVoteLogPrefixUnlocked(*request),
request->candidate_uuid(),
CurrentTermUnlocked(),
GetVotedForCurrentTermUnlocked());
LOG(INFO) << msg;
StatusToPB(Status::InvalidArgument(msg), response->mutable_consensus_error()->mutable_status());
return Status::OK();
}
Status RaftConsensus::RequestVoteRespondLastOpIdTooOld(const OpId& local_last_logged_opid,
const VoteRequestPB* request,
VoteResponsePB* response) {
FillVoteResponseVoteDenied(ConsensusErrorPB::LAST_OPID_TOO_OLD, response);
string msg = Substitute("$0: Denying vote to candidate $1 for term $2 because "
"replica has last-logged OpId of $3, which is greater than that of the "
"candidate, which has last-logged OpId of $4.",
GetRequestVoteLogPrefixUnlocked(*request),
request->candidate_uuid(),
request->candidate_term(),
SecureShortDebugString(local_last_logged_opid),
SecureShortDebugString(request->candidate_status().last_received()));
LOG(INFO) << msg;
StatusToPB(Status::InvalidArgument(msg), response->mutable_consensus_error()->mutable_status());
return Status::OK();
}
Status RaftConsensus::RequestVoteRespondLeaderIsAlive(const VoteRequestPB* request,
VoteResponsePB* response) {
FillVoteResponseVoteDenied(ConsensusErrorPB::LEADER_IS_ALIVE, response,
ResponderTermPolicy::DO_NOT_SET);
auto msg = Substitute("$0: Denying vote to candidate $1 for term $2 because "
"replica is either leader or believes a valid leader to "
"be alive.",
GetRequestVoteLogPrefixThreadSafe(*request),
request->candidate_uuid(),
request->candidate_term());
VLOG(1) << msg;
StatusToPB(Status::InvalidArgument(msg),
response->mutable_consensus_error()->mutable_status());
return Status::OK();
}
Status RaftConsensus::RequestVoteRespondIsBusy(
const VoteRequestPB* request, VoteResponsePB* response) {
// Don't set the term in the response: the requestor doesn't need it
// to process the NO vote response in this case.
FillVoteResponseVoteDenied(ConsensusErrorPB::CONSENSUS_BUSY, response,
ResponderTermPolicy::DO_NOT_SET);
auto msg = Substitute("$0: Denying vote to candidate $1 for term $2 because "
"replica is already servicing an update from "
"a current leader or another vote",
GetRequestVoteLogPrefixThreadSafe(*request),
request->candidate_uuid(),
request->candidate_term());
VLOG(1) << msg;
StatusToPB(Status::ServiceUnavailable(msg),
response->mutable_consensus_error()->mutable_status());
return Status::OK();
}
Status RaftConsensus::RequestVoteRespondVoteGranted(const VoteRequestPB* request,
VoteResponsePB* response) {
// We know our vote will be "yes", so avoid triggering an election while we
// persist our vote to disk. We use an exponential backoff to avoid too much
// split-vote contention when nodes display high latencies.
MonoDelta backoff = LeaderElectionExpBackoffDeltaUnlocked();
SnoozeFailureDetector(string("vote granted"), backoff);
if (!request->is_pre_election()) {
// Persist our vote to disk.
RETURN_NOT_OK(SetVotedForCurrentTermUnlocked(request->candidate_uuid()));
}
FillVoteResponseVoteGranted(response);
// Give peer time to become leader. Snooze one more time after persisting our
// vote. When disk latency is high, this should help reduce churn.
SnoozeFailureDetector(/*reason_for_log=*/boost::none, backoff);
LOG(INFO) << Substitute("$0: Granting yes vote for candidate $1 in term $2.",
GetRequestVoteLogPrefixUnlocked(*request),
request->candidate_uuid(),
CurrentTermUnlocked());
return Status::OK();
}
RaftPeerPB::Role RaftConsensus::role() const {
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
return cmeta_->active_role();
}
RaftConsensus::RoleAndMemberType RaftConsensus::GetRoleAndMemberType() const {
ThreadRestrictions::AssertWaitAllowed();
auto member_type = RaftPeerPB::UNKNOWN_MEMBER_TYPE;
const auto& local_peer_uuid = peer_uuid();
LockGuard l(lock_);
for (const auto& peer : cmeta_->ActiveConfig().peers()) {
if (peer.permanent_uuid() == local_peer_uuid) {
member_type = peer.member_type();
break;
}
}
return std::make_pair(cmeta_->active_role(), member_type);
}
int64_t RaftConsensus::CurrentTerm() const {
LockGuard l(lock_);
return CurrentTermUnlocked();
}
void RaftConsensus::SetStateUnlocked(State new_state) {
DCHECK(lock_.is_locked());
switch (new_state) {
case kInitialized:
CHECK_EQ(kNew, state_);
break;
case kRunning:
CHECK_EQ(kInitialized, state_);
break;
case kStopping:
CHECK(state_ != kStopped && state_ != kShutdown) << "State = " << State_Name(state_);
break;
case kStopped:
CHECK_EQ(kStopping, state_);
break;
case kShutdown:
CHECK(state_ == kStopped || state_ == kShutdown) << "State = " << State_Name(state_);
break;
default:
LOG(FATAL) << "Disallowed transition to state = " << State_Name(new_state);
break;
}
state_ = new_state;
}
const char* RaftConsensus::State_Name(State state) {
switch (state) {
case kNew:
return "New";
case kInitialized:
return "Initialized";
case kRunning:
return "Running";
case kStopping:
return "Stopping";
case kStopped:
return "Stopped";
case kShutdown:
return "Shut down";
default:
LOG(DFATAL) << "Unknown State value: " << state;
return "Unknown";
}
}
MonoDelta RaftConsensus::MinimumElectionTimeout() {
int32_t failure_timeout = FLAGS_leader_failure_max_missed_heartbeat_periods *
FLAGS_raft_heartbeat_interval_ms;
return MonoDelta::FromMilliseconds(failure_timeout);
}
void RaftConsensus::SetLeaderUuidUnlocked(const string& uuid) {
DCHECK(lock_.is_locked());
failed_elections_since_stable_leader_ = 0;
num_failed_elections_metric_->set_value(failed_elections_since_stable_leader_);
cmeta_->set_leader_uuid(uuid);
MarkDirty(Substitute("New leader $0", uuid));
}
Status RaftConsensus::ReplicateConfigChangeUnlocked(
RaftConfigPB old_config,
RaftConfigPB new_config,
StatusCallback client_cb) {
DCHECK(lock_.is_locked());
auto cc_replicate = new ReplicateMsg();
cc_replicate->set_op_type(CHANGE_CONFIG_OP);
ChangeConfigRecordPB* cc_req = cc_replicate->mutable_change_config_record();
cc_req->set_tablet_id(options_.tablet_id);
*cc_req->mutable_old_config() = std::move(old_config);
*cc_req->mutable_new_config() = std::move(new_config);
CHECK_OK(time_manager_->AssignTimestamp(cc_replicate));
scoped_refptr<ConsensusRound> round(
new ConsensusRound(this, make_scoped_refptr(new RefCountedReplicate(cc_replicate))));
auto* round_raw = round.get();
round->SetConsensusReplicatedCallback(
[this, round_raw, client_cb](const Status& s) {
this->NonTxRoundReplicationFinished(round_raw, client_cb, s);
});
return AppendNewRoundToQueueUnlocked(round);
}
Status RaftConsensus::RefreshConsensusQueueAndPeersUnlocked() {
DCHECK(lock_.is_locked());
DCHECK_EQ(RaftPeerPB::LEADER, cmeta_->active_role());
const RaftConfigPB& active_config = cmeta_->ActiveConfig();
// Change the peers so that we're able to replicate messages remotely and
// locally. The peer manager must be closed before updating the active config
// in the queue -- when the queue is in LEADER mode, it checks that all
// registered peers are a part of the active config.
peer_manager_->Close();
// TODO(todd): should use queue committed index here? in that case do
// we need to pass it in at all?
queue_->SetLeaderMode(pending_->GetCommittedIndex(),
CurrentTermUnlocked(),
active_config);
return peer_manager_->UpdateRaftConfig(active_config);
}
const string& RaftConsensus::peer_uuid() const {
return local_peer_pb_.permanent_uuid();
}
const string& RaftConsensus::tablet_id() const {
return options_.tablet_id;
}
Status RaftConsensus::ConsensusState(ConsensusStatePB* cstate,
IncludeHealthReport report_health) const {
ThreadRestrictions::AssertWaitAllowed();
UniqueLock l(lock_);
if (state_ == kShutdown) {
return Status::IllegalState("Tablet replica is shutdown");
}
ConsensusStatePB cstate_tmp = cmeta_->ToConsensusStatePB();
// If we need to include the health report, merge it into the committed
// config iff we believe we are the current leader of the config.
if (report_health == INCLUDE_HEALTH_REPORT &&
cmeta_->active_role() == RaftPeerPB::LEADER) {
auto reports = queue_->ReportHealthOfPeers();
// We don't need to access the queue anymore, so drop the consensus lock.
l.unlock();
// Iterate through each peer in the committed config and attach the health
// report to it.
RaftConfigPB* committed_raft_config = cstate_tmp.mutable_committed_config();
for (int i = 0; i < committed_raft_config->peers_size(); i++) {
RaftPeerPB* peer = committed_raft_config->mutable_peers(i);
const HealthReportPB* report = FindOrNull(reports, peer->permanent_uuid());
if (!report) continue; // Only attach details if we know about the peer.
*peer->mutable_health_report() = *report;
}
}
*cstate = std::move(cstate_tmp);
return Status::OK();
}
RaftConfigPB RaftConsensus::CommittedConfig() const {
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
return cmeta_->CommittedConfig();
}
void RaftConsensus::DumpStatusHtml(std::ostream& out) const {
if (state_ != kRunning) {
out << "Tablet " << EscapeForHtmlToString(tablet_id())
<< " not running" << std::endl;
return;
}
const RaftPeerPB::Role role = cmeta_->GetRoleAndTerm().first;
out << "<h1>Raft Consensus State</h1>" << std::endl;
out << "<h2>State</h2>" << std::endl;
out << "<pre>" << EscapeForHtmlToString(ToString()) << "</pre>" << std::endl;
out << "<h2>Queue</h2>" << std::endl;
out << "<pre>" << EscapeForHtmlToString(queue_->ToString()) << "</pre>" << std::endl;
// Dump the queues on a leader.
if (role == RaftPeerPB::LEADER) {
out << "<h2>Queue overview</h2>" << std::endl;
out << "<pre>" << EscapeForHtmlToString(queue_->ToString()) << "</pre>" << std::endl;
out << "<hr/>" << std::endl;
out << "<h2>Queue details</h2>" << std::endl;
queue_->DumpToHtml(out);
}
}
void RaftConsensus::ElectionCallback(ElectionReason reason, const ElectionResult& result) {
// We're running on a reactor thread; service the callback on another thread.
//
// There's no need to reenable the failure detector; if this fails, it's a
// sign that RaftConsensus has stopped and we no longer need failure detection.
auto self = shared_from_this();
Status s = raft_pool_token_->Submit([=]() { self->DoElectionCallback(reason, result); });
if (!s.ok()) {
static const char* msg = "unable to run election callback";
CHECK(s.IsServiceUnavailable()) << LogPrefixThreadSafe() << msg;
WARN_NOT_OK(s, LogPrefixThreadSafe() + msg);
}
}
void RaftConsensus::DoElectionCallback(ElectionReason reason, const ElectionResult& result) {
const int64_t election_term = result.vote_request.candidate_term();
const bool was_pre_election = result.vote_request.is_pre_election();
const char* election_type = was_pre_election ? "pre-election" : "election";
ThreadRestrictions::AssertWaitAllowed();
UniqueLock l(lock_);
Status s = CheckRunningUnlocked();
if (PREDICT_FALSE(!s.ok())) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Received " << election_type << " callback for term "
<< election_term << " while not running: "
<< s.ToString();
return;
}
// If this election was not triggered by the failure detector, the fd may
// still be enabled and needs to be snoozed, both if we win and lose:
// - When we win because we're about to disable it and become leader.
// - When we lose or otherwise we can fall into a cycle, where everyone keeps
// triggering elections but no election ever completes because by the time they
// finish another one is triggered already.
//
// This is a no-op if the failure detector is disabled..
MonoDelta snooze_delta = LeaderElectionExpBackoffDeltaUnlocked();
SnoozeFailureDetector(string("election complete"), snooze_delta);
auto enable_fd = MakeScopedCleanup([&]() {
// The failure detector was disabled if it triggered this election. Reenable
// it when exiting this function.
if (reason == ELECTION_TIMEOUT_EXPIRED) {
EnableFailureDetector(snooze_delta);
}
});
if (result.decision == VOTE_DENIED) {
failed_elections_since_stable_leader_++;
num_failed_elections_metric_->set_value(failed_elections_since_stable_leader_);
// If we called an election and one of the voters had a higher term than we did,
// we should bump our term before we potentially try again. This is particularly
// important with pre-elections to avoid getting "stuck" in a case like:
// Peer A: has ops through 1.10, term = 2, voted in term 2 for peer C
// Peer B: has ops through 1.15, term = 1
// In this case, Peer B will reject peer A's pre-elections for term 3 because
// the local log is longer. Peer A will reject B's pre-elections for term 2
// because it already voted in term 2. The check below ensures that peer B
// will bump to term 2 when it gets the vote rejection, such that its
// next pre-election (for term 3) would succeed.
if (result.highest_voter_term > CurrentTermUnlocked()) {
HandleTermAdvanceUnlocked(result.highest_voter_term);
}
LOG_WITH_PREFIX_UNLOCKED(INFO)
<< "Leader " << election_type << " lost for term " << election_term
<< ". Reason: "
<< (!result.message.empty() ? result.message : "None given");
return;
}
// In a pre-election, we collected votes for the _next_ term.
// So, we need to adjust our expectations of what the current term should be.
int64_t election_started_in_term = election_term;
if (was_pre_election) {
election_started_in_term--;
}
if (election_started_in_term != CurrentTermUnlocked()) {
LOG_WITH_PREFIX_UNLOCKED(INFO)
<< "Leader " << election_type << " decision vote started in "
<< "defunct term " << election_started_in_term << ": won";
return;
}
if (!cmeta_->IsVoterInConfig(peer_uuid(), ACTIVE_CONFIG)) {
LOG_WITH_PREFIX_UNLOCKED(WARNING) << "Leader " << election_type
<< " decision while not in active config. "
<< "Result: Term " << election_term
<< ": won. RaftConfig: "
<< SecureShortDebugString(cmeta_->ActiveConfig());
return;
}
// At this point we're either already leader, we're going to become leader,
// or we're going to run a real election.
//
// In all cases, do not reenable the failure detector.
enable_fd.cancel();
if (cmeta_->active_role() == RaftPeerPB::LEADER) {
// If this was a pre-election, it's possible to see the following interleaving:
//
// 1. Term N (follower): send a real election for term N.
// 2. An externally-triggered election is started.
// 3. Term N (follower): send a pre-election for term N+1.
// 4. Election callback for real election from term N completes.
// Peer is now leader for term N.
// 5. Pre-election callback from term N+1 completes, even though
// we are currently a leader of term N.
// In this case, we should just ignore the pre-election, since we're
// happily the leader of the prior term.
if (was_pre_election) return;
LOG_WITH_PREFIX_UNLOCKED(DFATAL)
<< "Leader " << election_type << " callback while already leader! "
<< "Result: Term " << election_term << ": won";
return;
}
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Leader " << election_type << " won for term " << election_term;
if (was_pre_election) {
// We just won the pre-election. So, we need to call a real election.
l.unlock();
WARN_NOT_OK_EVERY_N_SECS(StartElection(NORMAL_ELECTION, reason),
"Couldn't start leader election after successful pre-election", 10);
} else {
// We won a real election. Convert role to LEADER.
SetLeaderUuidUnlocked(peer_uuid());
// TODO(todd): BecomeLeaderUnlocked() can fail due to state checks during shutdown.
// It races with the above state check.
// This could be a problem during tablet deletion.
CHECK_OK(BecomeLeaderUnlocked());
}
}
boost::optional<OpId> RaftConsensus::GetLastOpId(OpIdType type) {
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
return GetLastOpIdUnlocked(type);
}
boost::optional<OpId> RaftConsensus::GetLastOpIdUnlocked(OpIdType type) {
// Return early if this method is called on an instance of RaftConsensus that
// has not yet been started, failed during Init(), or failed during Start().
if (!queue_ || !pending_) return boost::none;
switch (type) {
case RECEIVED_OPID:
return queue_->GetLastOpIdInLog();
case COMMITTED_OPID:
return MakeOpId(pending_->GetTermWithLastCommittedOp(),
pending_->GetCommittedIndex());
default:
LOG(DFATAL) << LogPrefixUnlocked() << "Invalid OpIdType " << type;
return boost::none;
}
}
log::RetentionIndexes RaftConsensus::GetRetentionIndexes() {
// Grab the watermarks from the queue. It's OK to fetch these two watermarks
// separately -- the worst case is we see a relatively "out of date" watermark
// which just means we'll retain slightly more than necessary in this invocation
// of log GC.
return log::RetentionIndexes(queue_->GetCommittedIndex(), // for durability
queue_->GetAllReplicatedIndex()); // for peers
}
void RaftConsensus::MarkDirty(const string& reason) {
WARN_NOT_OK(raft_pool_token_->Submit([=]() { this->mark_dirty_clbk_(reason); }),
LogPrefixThreadSafe() + "Unable to run MarkDirty callback");
}
void RaftConsensus::MarkDirtyOnSuccess(const string& reason,
const StatusCallback& client_cb,
const Status& status) {
if (PREDICT_TRUE(status.ok())) {
MarkDirty(reason);
}
client_cb(status);
}
void RaftConsensus::NonTxRoundReplicationFinished(ConsensusRound* round,
const StatusCallback& client_cb,
const Status& status) {
// NOTE: lock_ is held here because this is triggered by
// PendingRounds::AbortOpsAfter() and AdvanceCommittedIndex().
DCHECK(lock_.is_locked());
OperationType op_type = round->replicate_msg()->op_type();
const string& op_type_str = OperationType_Name(op_type);
CHECK(IsConsensusOnlyOperation(op_type)) << "Unexpected op type: " << op_type_str;
if (op_type == CHANGE_CONFIG_OP) {
CompleteConfigChangeRoundUnlocked(round, status);
// Fall through to the generic handling.
}
// TODO(mpercy): May need some refactoring to unlock 'lock_' before invoking
// the client callback.
if (!status.ok()) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << op_type_str << " replication failed: "
<< status.ToString();
client_cb(status);
return;
}
VLOG_WITH_PREFIX_UNLOCKED(1) << "Committing " << op_type_str << " with op id "
<< round->id();
round_handler_->FinishConsensusOnlyRound(round);
CommitMsg commit_msg;
commit_msg.set_op_type(round->replicate_msg()->op_type());
*commit_msg.mutable_commited_op_id() = round->id();
CHECK_OK(log_->AsyncAppendCommit(
commit_msg, [](const Status& s) {
CrashIfNotOkStatusCB("Enqueued commit operation failed to write to WAL", s);
}));
client_cb(status);
}
void RaftConsensus::CompleteConfigChangeRoundUnlocked(ConsensusRound* round, const Status& status) {
DCHECK(lock_.is_locked());
const OpId& op_id = round->replicate_msg()->id();
if (!status.ok()) {
// If the config change being aborted is the current pending one, abort it.
if (cmeta_->has_pending_config() &&
cmeta_->GetConfigOpIdIndex(PENDING_CONFIG) == op_id.index()) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Aborting config change with OpId "
<< op_id << ": " << status.ToString();
cmeta_->clear_pending_config();
// Disable leader failure detection if transitioning from VOTER to
// NON_VOTER and vice versa.
UpdateFailureDetectorState();
} else {
LOG_WITH_PREFIX_UNLOCKED(INFO)
<< "Skipping abort of non-pending config change with OpId "
<< op_id << ": " << status.ToString();
}
// It's possible to abort a config change which isn't the pending one in the following
// sequence:
// - replicate a config change
// - it gets committed, so we write the new config to disk as the Committed configuration
// - we crash before the COMMIT message hits the WAL
// - we restart the server, and the config change is added as a pending round again,
// but isn't set as Pending because it's already committed.
// - we delete the tablet before committing it
// See KUDU-1735.
return;
}
// Commit the successful config change.
DCHECK(round->replicate_msg()->change_config_record().has_old_config());
DCHECK(round->replicate_msg()->change_config_record().has_new_config());
const RaftConfigPB& old_config = round->replicate_msg()->change_config_record().old_config();
const RaftConfigPB& new_config = round->replicate_msg()->change_config_record().new_config();
DCHECK(old_config.has_opid_index());
DCHECK(new_config.has_opid_index());
// Check if the pending Raft config has an OpId less than the committed
// config. If so, this is a replay at startup in which the COMMIT
// messages were delayed.
int64_t committed_config_opid_index = cmeta_->GetConfigOpIdIndex(COMMITTED_CONFIG);
if (new_config.opid_index() > committed_config_opid_index) {
LOG_WITH_PREFIX_UNLOCKED(INFO)
<< "Committing config change with OpId "
<< op_id << ": "
<< DiffRaftConfigs(old_config, new_config)
<< ". New config: { " << SecureShortDebugString(new_config) << " }";
CHECK_OK(SetCommittedConfigUnlocked(new_config));
} else {
LOG_WITH_PREFIX_UNLOCKED(INFO)
<< "Ignoring commit of config change with OpId "
<< op_id << " because the committed config has OpId index "
<< committed_config_opid_index << ". The config change we are ignoring is: "
<< "Old config: { " << SecureShortDebugString(old_config) << " }. "
<< "New config: { " << SecureShortDebugString(new_config) << " }";
}
}
void RaftConsensus::EnableFailureDetector(boost::optional<MonoDelta> delta) {
if (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {
failure_detector_->Start(std::move(delta));
}
}
void RaftConsensus::DisableFailureDetector() {
if (PREDICT_TRUE(FLAGS_enable_leader_failure_detection)) {
failure_detector_->Stop();
}
}
void RaftConsensus::UpdateFailureDetectorState(boost::optional<MonoDelta> delta) {
DCHECK(lock_.is_locked());
const auto& uuid = peer_uuid();
if (uuid != cmeta_->leader_uuid() &&
cmeta_->IsVoterInConfig(uuid, ACTIVE_CONFIG)) {
// A voter that is not the leader should run the failure detector.
EnableFailureDetector(std::move(delta));
} else {
// Otherwise, the local peer should not start leader elections
// (e.g. if it is the leader, a non-voter, a non-participant, etc).
DisableFailureDetector();
}
}
void RaftConsensus::SnoozeFailureDetector(boost::optional<string> reason_for_log,
boost::optional<MonoDelta> delta) {
if (PREDICT_TRUE(failure_detector_ && FLAGS_enable_leader_failure_detection)) {
if (reason_for_log) {
VLOG(1) << LogPrefixThreadSafe()
<< Substitute("Snoozing failure detection for $0 ($1)",
delta ? delta->ToString() : "election timeout",
*reason_for_log);
}
if (!delta) {
delta = MinimumElectionTimeout();
}
failure_detector_->Snooze(std::move(delta));
}
}
void RaftConsensus::WithholdVotes() {
MonoTime prev = withhold_votes_until_;
MonoTime next = MonoTime::Now() + MinimumElectionTimeout();
do {
if (prev == MonoTime::Max()) {
// Maximum withholding time already. It might be the case if replica
// has become a leader already.
break;
}
next = MonoTime::Now() + MinimumElectionTimeout();
} while (!withhold_votes_until_.compare_exchange_weak(prev, next));
}
MonoDelta RaftConsensus::LeaderElectionExpBackoffDeltaUnlocked() {
DCHECK(lock_.is_locked());
// Compute a backoff factor based on how many leader elections have
// failed since a stable leader was last seen.
double backoff_factor = pow(1.5, failed_elections_since_stable_leader_ + 1);
double min_timeout = MinimumElectionTimeout().ToMilliseconds();
double max_timeout = std::min<double>(
min_timeout * backoff_factor,
FLAGS_leader_failure_exp_backoff_max_delta_ms);
// Randomize the timeout between the minimum and the calculated value.
// We do this after the above capping to the max. Otherwise, after a
// churny period, we'd end up highly likely to backoff exactly the max
// amount.
double timeout = min_timeout + (max_timeout - min_timeout) * rng_.NextDoubleFraction();
DCHECK_GE(timeout, min_timeout);
return MonoDelta::FromMilliseconds(timeout);
}
Status RaftConsensus::HandleTermAdvanceUnlocked(ConsensusTerm new_term,
FlushToDisk flush) {
DCHECK(lock_.is_locked());
if (new_term <= CurrentTermUnlocked()) {
return Status::IllegalState(Substitute("Can't advance term to: $0 current term: $1 is higher.",
new_term, CurrentTermUnlocked()));
}
if (cmeta_->active_role() == RaftPeerPB::LEADER) {
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Stepping down as leader of term "
<< CurrentTermUnlocked();
RETURN_NOT_OK(BecomeReplicaUnlocked());
}
LOG_WITH_PREFIX_UNLOCKED(INFO) << "Advancing to term " << new_term;
RETURN_NOT_OK(SetCurrentTermUnlocked(new_term, flush));
if (term_metric_) {
term_metric_->set_value(new_term);
}
last_received_cur_leader_ = MinimumOpId();
return Status::OK();
}
Status RaftConsensus::CheckSafeToReplicateUnlocked(const ReplicateMsg& msg) const {
DCHECK(lock_.is_locked());
DCHECK(!msg.has_id()) << "Should not have an ID yet: " << SecureShortDebugString(msg);
RETURN_NOT_OK(CheckRunningUnlocked());
return CheckActiveLeaderUnlocked();
}
Status RaftConsensus::CheckRunningUnlocked() const {
DCHECK(lock_.is_locked());
if (PREDICT_FALSE(state_ != kRunning)) {
return Status::IllegalState("RaftConsensus is not running",
Substitute("State = $0", State_Name(state_)));
}
return Status::OK();
}
Status RaftConsensus::CheckActiveLeaderUnlocked() const {
DCHECK(lock_.is_locked());
RaftPeerPB::Role role = cmeta_->active_role();
switch (role) {
case RaftPeerPB::LEADER:
// Check for the consistency of the information in the consensus metadata
// and the state of the consensus queue.
DCHECK(queue_->IsInLeaderMode());
if (leader_transfer_in_progress_) {
return Status::ServiceUnavailable("leader transfer in progress");
}
return Status::OK();
default:
// Check for the consistency of the information in the consensus metadata
// and the state of the consensus queue.
DCHECK(!queue_->IsInLeaderMode());
return Status::IllegalState(Substitute("Replica $0 is not leader of this config. Role: $1. "
"Consensus state: $2",
peer_uuid(),
RaftPeerPB::Role_Name(role),
SecureShortDebugString(cmeta_->ToConsensusStatePB())));
}
}
Status RaftConsensus::CheckNoConfigChangePendingUnlocked() const {
DCHECK(lock_.is_locked());
if (cmeta_->has_pending_config()) {
return Status::IllegalState(
Substitute("RaftConfig change currently pending. Only one is allowed at a time.\n"
" Committed config: $0.\n Pending config: $1",
SecureShortDebugString(cmeta_->CommittedConfig()),
SecureShortDebugString(cmeta_->PendingConfig())));
}
return Status::OK();
}
Status RaftConsensus::SetPendingConfigUnlocked(const RaftConfigPB& new_config) {
DCHECK(lock_.is_locked());
RETURN_NOT_OK_PREPEND(VerifyRaftConfig(new_config),
"Invalid config to set as pending");
if (!new_config.unsafe_config_change()) {
CHECK(!cmeta_->has_pending_config())
<< "Attempt to set pending config while another is already pending! "
<< "Existing pending config: " << SecureShortDebugString(cmeta_->PendingConfig()) << "; "
<< "Attempted new pending config: " << SecureShortDebugString(new_config);
} else if (cmeta_->has_pending_config()) {
LOG_WITH_PREFIX_UNLOCKED(INFO)
<< "Allowing unsafe config change even though there is a pending config! "
<< "Existing pending config: " << SecureShortDebugString(cmeta_->PendingConfig()) << "; "
<< "New pending config: " << SecureShortDebugString(new_config);
}
cmeta_->set_pending_config(new_config);
UpdateFailureDetectorState();
return Status::OK();
}
Status RaftConsensus::SetCommittedConfigUnlocked(const RaftConfigPB& config_to_commit) {
TRACE_EVENT0("consensus", "RaftConsensus::SetCommittedConfigUnlocked");
DCHECK(lock_.is_locked());
DCHECK(config_to_commit.IsInitialized());
RETURN_NOT_OK_PREPEND(VerifyRaftConfig(config_to_commit),
"Invalid config to set as committed");
// Compare committed with pending configuration, ensure that they are the same.
// In the event of an unsafe config change triggered by an administrator,
// it is possible that the config being committed may not match the pending config
// because unsafe config change allows multiple pending configs to exist.
// Therefore we only need to validate that 'config_to_commit' matches the pending config
// if the pending config does not have its 'unsafe_config_change' flag set.
if (cmeta_->has_pending_config()) {
RaftConfigPB pending_config = cmeta_->PendingConfig();
if (!pending_config.unsafe_config_change()) {
// Quorums must be exactly equal, even w.r.t. peer ordering.
CHECK_EQ(pending_config.SerializeAsString(),
config_to_commit.SerializeAsString())
<< Substitute("New committed config must equal pending config, but does not. "
"Pending config: $0, committed config: $1",
SecureShortDebugString(pending_config),
SecureShortDebugString(config_to_commit));
}
}
cmeta_->set_committed_config(config_to_commit);
cmeta_->clear_pending_config();
CHECK_OK(cmeta_->Flush());
return Status::OK();
}
Status RaftConsensus::SetCurrentTermUnlocked(int64_t new_term,
FlushToDisk flush) {
TRACE_EVENT1("consensus", "RaftConsensus::SetCurrentTermUnlocked",
"term", new_term);
DCHECK(lock_.is_locked());
if (PREDICT_FALSE(new_term <= CurrentTermUnlocked())) {
return Status::IllegalState(
Substitute("Cannot change term to a term that is lower than or equal to the current one. "
"Current: $0, Proposed: $1", CurrentTermUnlocked(), new_term));
}
cmeta_->set_current_term(new_term);
cmeta_->clear_voted_for();
if (flush == FLUSH_TO_DISK) {
CHECK_OK(cmeta_->Flush());
}
ClearLeaderUnlocked();
return Status::OK();
}
int64_t RaftConsensus::CurrentTermUnlocked() const {
DCHECK(lock_.is_locked());
return cmeta_->current_term();
}
string RaftConsensus::GetLeaderUuidUnlocked() const {
DCHECK(lock_.is_locked());
return cmeta_->leader_uuid();
}
bool RaftConsensus::HasLeaderUnlocked() const {
DCHECK(lock_.is_locked());
return !GetLeaderUuidUnlocked().empty();
}
void RaftConsensus::ClearLeaderUnlocked() {
DCHECK(lock_.is_locked());
leader_is_ready_ = false;
cmeta_->set_leader_uuid("");
}
bool RaftConsensus::HasVotedCurrentTermUnlocked() const {
DCHECK(lock_.is_locked());
return cmeta_->has_voted_for();
}
Status RaftConsensus::SetVotedForCurrentTermUnlocked(const string& uuid) {
TRACE_EVENT1("consensus", "RaftConsensus::SetVotedForCurrentTermUnlocked",
"uuid", uuid);
DCHECK(lock_.is_locked());
cmeta_->set_voted_for(uuid);
CHECK_OK(cmeta_->Flush());
return Status::OK();
}
const string& RaftConsensus::GetVotedForCurrentTermUnlocked() const {
DCHECK(lock_.is_locked());
DCHECK(cmeta_->has_voted_for());
return cmeta_->voted_for();
}
const ConsensusOptions& RaftConsensus::GetOptions() const {
return options_;
}
string RaftConsensus::LogPrefix() const {
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
return LogPrefixUnlocked();
}
string RaftConsensus::LogPrefixUnlocked() const {
DCHECK(lock_.is_locked());
// 'cmeta_' may not be set if initialization failed.
string cmeta_info;
if (cmeta_) {
cmeta_info = Substitute(" [term $0 $1]",
cmeta_->current_term(),
RaftPeerPB::Role_Name(cmeta_->active_role()));
}
return Substitute("T $0 P $1$2: ", options_.tablet_id, peer_uuid(), cmeta_info);
}
string RaftConsensus::LogPrefixThreadSafe() const {
return Substitute("T $0 P $1: ",
options_.tablet_id,
peer_uuid());
}
string RaftConsensus::ToString() const {
ThreadRestrictions::AssertWaitAllowed();
LockGuard l(lock_);
return ToStringUnlocked();
}
string RaftConsensus::ToStringUnlocked() const {
DCHECK(lock_.is_locked());
return Substitute("Replica: $0, State: $1, Role: $2",
peer_uuid(), State_Name(state_), RaftPeerPB::Role_Name(cmeta_->active_role()));
}
int64_t RaftConsensus::MetadataOnDiskSize() const {
return cmeta_->on_disk_size();
}
ConsensusMetadata* RaftConsensus::consensus_metadata_for_tests() const {
return cmeta_.get();
}
int64_t RaftConsensus::GetMillisSinceLastLeaderHeartbeat() const {
return last_leader_communication_time_micros_ == 0 ?
0 : (GetMonoTimeMicros() - last_leader_communication_time_micros_) / 1000;
}
////////////////////////////////////////////////////////////////////////
// ConsensusBootstrapInfo
////////////////////////////////////////////////////////////////////////
ConsensusBootstrapInfo::ConsensusBootstrapInfo()
: last_id(MinimumOpId()),
last_committed_id(MinimumOpId()) {
}
ConsensusBootstrapInfo::~ConsensusBootstrapInfo() {
STLDeleteElements(&orphaned_replicates);
}
////////////////////////////////////////////////////////////////////////
// ConsensusRound
////////////////////////////////////////////////////////////////////////
ConsensusRound::ConsensusRound(RaftConsensus* consensus,
unique_ptr<ReplicateMsg> replicate_msg,
ConsensusReplicatedCallback replicated_cb)
: consensus_(consensus),
replicate_msg_(new RefCountedReplicate(replicate_msg.release())),
replicated_cb_(std::move(replicated_cb)),
bound_term_(-1) {}
ConsensusRound::ConsensusRound(RaftConsensus* consensus,
ReplicateRefPtr replicate_msg)
: consensus_(consensus),
replicate_msg_(std::move(replicate_msg)),
bound_term_(-1) {
DCHECK(replicate_msg_);
}
void ConsensusRound::NotifyReplicationFinished(const Status& status) {
if (PREDICT_FALSE(!replicated_cb_)) return;
replicated_cb_(status);
}
Status ConsensusRound::CheckBoundTerm(int64_t current_term) const {
if (PREDICT_FALSE(bound_term_ != -1 &&
bound_term_ != current_term)) {
return Status::Aborted(
strings::Substitute(
"Op submitted in term $0 cannot be replicated in term $1",
bound_term_, current_term));
}
return Status::OK();
}
} // namespace consensus
} // namespace kudu