You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by mp...@apache.org on 2016/06/22 20:22:11 UTC
incubator-kudu git commit: consensus: Crash if we fail to append to
the WAL
Repository: incubator-kudu
Updated Branches:
refs/heads/master 4983a7982 -> c2d68da78
consensus: Crash if we fail to append to the WAL
Being unable to write to the WAL is a serious problem with complicated
consequences. Let's always crash when that happens.
This patch adds a fault injection test and a callback that simply
crashes on failure.
Change-Id: I9c5ffa6d60f46ce4d6c3da4c6e7878ab07d99c65
Reviewed-on: http://gerrit.cloudera.org:8080/3406
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/c2d68da7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/c2d68da7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/c2d68da7
Branch: refs/heads/master
Commit: c2d68da783315c79162870328da5761484eeabf8
Parents: 4983a79
Author: Mike Percy <mp...@apache.org>
Authored: Thu May 26 16:53:22 2016 -0700
Committer: Mike Percy <mp...@apache.org>
Committed: Wed Jun 22 20:19:37 2016 +0000
----------------------------------------------------------------------
src/kudu/consensus/consensus_queue.cc | 3 +-
src/kudu/consensus/log.cc | 11 +-
src/kudu/consensus/log_cache.cc | 2 +-
src/kudu/consensus/raft_consensus.cc | 9 +-
src/kudu/consensus/raft_consensus.h | 2 +-
.../integration-tests/raft_consensus-itest.cc | 101 +++++++++++++++++++
.../tablet/transactions/transaction_driver.cc | 4 +-
src/kudu/util/status_callback.cc | 11 +-
src/kudu/util/status_callback.h | 5 +
9 files changed, 139 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/consensus/consensus_queue.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/consensus_queue.cc b/src/kudu/consensus/consensus_queue.cc
index c622c01..a661996 100644
--- a/src/kudu/consensus/consensus_queue.cc
+++ b/src/kudu/consensus/consensus_queue.cc
@@ -239,7 +239,8 @@ void PeerMessageQueue::LocalPeerAppendFinished(const OpId& id,
}
Status PeerMessageQueue::AppendOperation(const ReplicateRefPtr& msg) {
- return AppendOperations({ msg }, Bind(DoNothingStatusCB));
+ return AppendOperations({ msg }, Bind(CrashIfNotOkStatusCB,
+ "Enqueued replicate operation failed to write to WAL"));
}
Status PeerMessageQueue::AppendOperations(const vector<ReplicateRefPtr>& msgs,
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/consensus/log.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/log.cc b/src/kudu/consensus/log.cc
index 62eca2b..3e10523 100644
--- a/src/kudu/consensus/log.cc
+++ b/src/kudu/consensus/log.cc
@@ -95,6 +95,12 @@ TAG_FLAG(log_inject_latency_ms_mean, unsafe);
TAG_FLAG(log_inject_latency_ms_stddev, unsafe);
TAG_FLAG(fault_crash_before_append_commit, unsafe);
+DEFINE_double(log_inject_io_error_on_append_fraction, 0.0,
+ "Fraction of the time when the log will fail to append and return an IOError. "
+ "(For testing only!)");
+TAG_FLAG(log_inject_io_error_on_append_fraction, unsafe);
+TAG_FLAG(log_inject_io_error_on_append_fraction, runtime);
+
// Validate that log_min_segments_to_retain >= 1
static bool ValidateLogsToRetain(const char* flagname, int value) {
if (value >= 1) {
@@ -186,7 +192,6 @@ void Log::AppendThread::RunThread() {
Status s = log_->DoAppend(entry_batch);
if (PREDICT_FALSE(!s.ok())) {
LOG(ERROR) << "Error appending to the log: " << s.ToString();
- DLOG(FATAL) << "Aborting: " << s.ToString();
entry_batch->set_failed_to_append();
// TODO If a single transaction fails to append, should we
// abort all subsequent transactions in this batch or allow
@@ -207,7 +212,6 @@ void Log::AppendThread::RunThread() {
}
if (PREDICT_FALSE(!s.ok())) {
LOG(ERROR) << "Error syncing log" << s.ToString();
- DLOG(FATAL) << "Aborting: " << s.ToString();
for (LogEntryBatch* entry_batch : entry_batches) {
if (!entry_batch->callback().is_null()) {
entry_batch->callback().Run(s);
@@ -473,6 +477,9 @@ Status Log::DoAppend(LogEntryBatch* entry_batch, bool caller_owns_operation) {
size_t num_entries = entry_batch->count();
DCHECK_GT(num_entries, 0) << "Cannot call DoAppend() with zero entries reserved";
+ MAYBE_RETURN_FAILURE(FLAGS_log_inject_io_error_on_append_fraction,
+ Status::IOError("Injected IOError in Log::DoAppend()"));
+
Slice entry_batch_data = entry_batch->data();
uint32_t entry_batch_bytes = entry_batch->total_size_bytes();
// If there is no data to write return OK.
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/consensus/log_cache.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/log_cache.cc b/src/kudu/consensus/log_cache.cc
index 643cf29..8aaa335 100644
--- a/src/kudu/consensus/log_cache.cc
+++ b/src/kudu/consensus/log_cache.cc
@@ -191,7 +191,7 @@ Status LogCache::AppendOperations(const vector<ReplicateRefPtr>& msgs,
callback));
l.lock();
if (!log_status.ok()) {
- LOG_WITH_PREFIX_UNLOCKED(WARNING) << "Couldn't append to log: " << log_status.ToString();
+ LOG_WITH_PREFIX_UNLOCKED(ERROR) << "Couldn't append to log: " << log_status.ToString();
tracker_->Release(mem_required);
return log_status;
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/consensus/raft_consensus.cc
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus.cc b/src/kudu/consensus/raft_consensus.cc
index c0bd303..ebe1938 100644
--- a/src/kudu/consensus/raft_consensus.cc
+++ b/src/kudu/consensus/raft_consensus.cc
@@ -562,6 +562,9 @@ Status RaftConsensus::AppendNewRoundToQueueUnlocked(const scoped_refptr<Consensu
// TODO Possibly evict a dangling peer from the configuration here.
// TODO count of number of ops failed due to consensus queue overflow.
+ } else if (PREDICT_FALSE(s.IsIOError())) {
+ // This likely came from the log.
+ LOG(FATAL) << "IO error appending to the queue: " << s.ToString();
}
RETURN_NOT_OK_PREPEND(s, "Unable to append operation to consensus queue");
state_->UpdateLastReceivedOpIdUnlocked(round->id());
@@ -1922,8 +1925,10 @@ void RaftConsensus::NonTxRoundReplicationFinished(ConsensusRound* round,
commit_msg->set_op_type(round->replicate_msg()->op_type());
*commit_msg->mutable_commited_op_id() = round->id();
- WARN_NOT_OK(log_->AsyncAppendCommit(std::move(commit_msg), Bind(&DoNothingStatusCB)),
- "Unable to append commit message");
+ CHECK_OK(log_->AsyncAppendCommit(std::move(commit_msg),
+ Bind(CrashIfNotOkStatusCB,
+ "Enqueued commit operation failed to write to WAL")));
+
client_cb.Run(status);
}
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/consensus/raft_consensus.h
----------------------------------------------------------------------
diff --git a/src/kudu/consensus/raft_consensus.h b/src/kudu/consensus/raft_consensus.h
index 584aa7a..b516405 100644
--- a/src/kudu/consensus/raft_consensus.h
+++ b/src/kudu/consensus/raft_consensus.h
@@ -176,7 +176,7 @@ class RaftConsensus : public Consensus,
const StatusCallback& client_cb,
const Status& status);
- // As a leader, append a new ConsensusRond to the queue.
+ // As a leader, append a new ConsensusRound to the queue.
// Only virtual and protected for mocking purposes.
virtual Status AppendNewRoundToQueueUnlocked(const scoped_refptr<ConsensusRound>& round);
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/integration-tests/raft_consensus-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/raft_consensus-itest.cc b/src/kudu/integration-tests/raft_consensus-itest.cc
index 712aaab..e24a789 100644
--- a/src/kudu/integration-tests/raft_consensus-itest.cc
+++ b/src/kudu/integration-tests/raft_consensus-itest.cc
@@ -2510,6 +2510,107 @@ TEST_F(RaftConsensusITest, TestCorruptReplicaMetadata) {
MonoDelta::FromSeconds(30)));
}
+// Test that an IOError when writing to the write-ahead log is a fatal error.
+// First, we test that failed replicates are fatal. Then, we test that failed
+// commits are fatal.
+TEST_F(RaftConsensusITest, TestLogIOErrorIsFatal) {
+ FLAGS_num_replicas = 3;
+ FLAGS_num_tablet_servers = 3;
+ vector<string> ts_flags, master_flags;
+ ts_flags.push_back("--enable_leader_failure_detection=false");
+ master_flags.push_back("--catalog_manager_wait_for_new_tablets_to_elect_leader=false");
+ NO_FATALS(BuildAndStart(ts_flags, master_flags));
+
+ vector<TServerDetails*> tservers;
+ AppendValuesFromMap(tablet_servers_, &tservers);
+ ASSERT_EQ(3, tservers.size());
+
+ // Test failed replicates.
+
+ // Elect server 2 as leader and wait for log index 1 to propagate to all servers.
+ ASSERT_OK(StartElection(tservers[2], tablet_id_, MonoDelta::FromSeconds(10)));
+ ASSERT_OK(WaitForServersToAgree(MonoDelta::FromSeconds(10), tablet_servers_, tablet_id_, 1));
+
+ // Inject an IOError the next time servers 1 and 2 write to their WAL.
+ // Then, cause server 0 to start and win a leader election.
+ // This will cause servers 0 and 1 to crash.
+ for (int i = 1; i <= 2; i++) {
+ ASSERT_OK(cluster_->SetFlag(cluster_->tablet_server(i),
+ "log_inject_io_error_on_append_fraction", "1.0"));
+ }
+ ASSERT_OK(StartElection(tservers[0], tablet_id_, MonoDelta::FromSeconds(10)));
+ for (int i = 1; i <= 2; i++) {
+ ASSERT_OK(cluster_->tablet_server(i)->WaitForCrash(MonoDelta::FromSeconds(10)));
+ }
+
+ // Now we know followers crash when they write to their log.
+ // Let's verify the same for the leader (server 0).
+ ASSERT_OK(cluster_->SetFlag(cluster_->tablet_server(0),
+ "log_inject_io_error_on_append_fraction", "1.0"));
+
+ // Attempt to write to the leader, but with a short timeout.
+ TestWorkload workload(cluster_.get());
+ workload.set_table_name(kTableId);
+ workload.set_timeout_allowed(true);
+ workload.set_write_timeout_millis(100);
+ workload.set_num_write_threads(1);
+ workload.set_write_batch_size(1);
+ workload.Setup();
+ workload.Start();
+
+ // Leader should crash as well.
+ ASSERT_OK(cluster_->tablet_server(0)->WaitForCrash(MonoDelta::FromSeconds(10)));
+ workload.StopAndJoin();
+
+ LOG(INFO) << "Everything crashed!";
+
+ // Test failed commits.
+
+ cluster_->Shutdown();
+ ASSERT_OK(cluster_->Restart());
+ NO_FATALS(WaitForTSAndReplicas());
+ tservers.clear();
+ AppendValuesFromMap(tablet_servers_, &tservers);
+ ASSERT_EQ(3, tservers.size());
+
+ // Elect server 0 as leader, wait until writes are going through.
+ ASSERT_OK(StartElection(tservers[0], tablet_id_, MonoDelta::FromSeconds(10)));
+ workload.Start();
+ int64_t prev_inserted = workload.rows_inserted();
+ while (workload.rows_inserted() == prev_inserted) {
+ SleepFor(MonoDelta::FromMilliseconds(10));
+ }
+ workload.StopAndJoin();
+ ASSERT_OK(WaitForServersToAgree(MonoDelta::FromSeconds(10), tablet_servers_, tablet_id_, 1));
+
+ // Now shutdown servers 1 and 2 so that writes cannot commit. Write to the
+ // leader, set flags so that commits crash the server, then bring the
+ // followers back up.
+ for (int i = 1; i <= 2; i++) {
+ cluster_->tablet_server(i)->Shutdown();
+ }
+
+ OpId prev_opid, cur_opid;
+ ASSERT_OK(GetLastOpIdForReplica(tablet_id_, tservers[0], consensus::RECEIVED_OPID,
+ MonoDelta::FromSeconds(10), &prev_opid));
+ VLOG(1) << "Previous OpId on server 0: " << OpIdToString(prev_opid);
+ workload.Start();
+ // Wait until we've got (uncommitted) entries into the leader's log.
+ do {
+ ASSERT_OK(GetLastOpIdForReplica(tablet_id_, tservers[0], consensus::RECEIVED_OPID,
+ MonoDelta::FromSeconds(10), &cur_opid));
+ VLOG(1) << "Current OpId on server 0: " << OpIdToString(cur_opid);
+ } while (consensus::OpIdEquals(prev_opid, cur_opid));
+ workload.StopAndJoin();
+ ASSERT_OK(cluster_->SetFlag(cluster_->tablet_server(0),
+ "log_inject_io_error_on_append_fraction", "1.0"));
+ for (int i = 1; i <= 2; i++) {
+ ASSERT_OK(cluster_->tablet_server(i)->Restart());
+ }
+ // Leader will crash.
+ ASSERT_OK(cluster_->tablet_server(0)->WaitForCrash(MonoDelta::FromSeconds(10)));
+}
+
} // namespace tserver
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/tablet/transactions/transaction_driver.cc
----------------------------------------------------------------------
diff --git a/src/kudu/tablet/transactions/transaction_driver.cc b/src/kudu/tablet/transactions/transaction_driver.cc
index 30ec10e..fd68ca6 100644
--- a/src/kudu/tablet/transactions/transaction_driver.cc
+++ b/src/kudu/tablet/transactions/transaction_driver.cc
@@ -380,7 +380,9 @@ void TransactionDriver::ApplyTask() {
{
TRACE_EVENT1("txn", "AsyncAppendCommit", "txn", this);
- CHECK_OK(log_->AsyncAppendCommit(std::move(commit_msg), Bind(DoNothingStatusCB)));
+ CHECK_OK(log_->AsyncAppendCommit(std::move(commit_msg),
+ Bind(CrashIfNotOkStatusCB,
+ "Enqueued commit operation failed to write to WAL")));
}
// If the client requested COMMIT_WAIT as the external consistency mode
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/util/status_callback.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/status_callback.cc b/src/kudu/util/status_callback.cc
index 0bb11a0..667bfec 100644
--- a/src/kudu/util/status_callback.cc
+++ b/src/kudu/util/status_callback.cc
@@ -15,13 +15,22 @@
// specific language governing permissions and limitations
// under the License.
-#include "kudu/util/status.h"
#include "kudu/util/status_callback.h"
+#include "kudu/util/status.h"
+
+using std::string;
+
namespace kudu {
void DoNothingStatusCB(const Status& status) {}
+void CrashIfNotOkStatusCB(const string& message, const Status& status) {
+ if (PREDICT_FALSE(!status.ok())) {
+ LOG(FATAL) << message << ": " << status.ToString();
+ }
+}
+
Status DoNothingStatusClosure() { return Status::OK(); }
} // end namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/c2d68da7/src/kudu/util/status_callback.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/status_callback.h b/src/kudu/util/status_callback.h
index 83b9d90..3a36b83 100644
--- a/src/kudu/util/status_callback.h
+++ b/src/kudu/util/status_callback.h
@@ -17,6 +17,8 @@
#ifndef KUDU_UTIL_STATUS_CALLBACK_H
#define KUDU_UTIL_STATUS_CALLBACK_H
+#include <string>
+
#include "kudu/gutil/callback_forward.h"
namespace kudu {
@@ -31,6 +33,9 @@ typedef Callback<void(const Status& status)> StatusCallback;
// is needed.
extern void DoNothingStatusCB(const Status& status);
+// A callback that crashes with a FATAL log message if the given Status is not OK.
+extern void CrashIfNotOkStatusCB(const std::string& message, const Status& status);
+
// A closure (callback without arguments) that returns a Status indicating
// whether it was successful or not.
typedef Callback<Status(void)> StatusClosure;