You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by ad...@apache.org on 2017/07/31 21:45:45 UTC
kudu git commit: disk failure: error-handling macros in blocks
Repository: kudu
Updated Branches:
refs/heads/master 12efb85e1 -> a21a50f87
disk failure: error-handling macros in blocks
This patch adds macros to trigger error-handling code in the event of
disk failure during block IO. Block- and container-level functions that
call env or cache functions (e.g. File::Append, FileCache::DeleteFile)
that may result in disk failure can now run callbacks that, in the
future, will fail and shutdown tablets in the failed directory,
preventing further IO to the disk.
Disk failures are indicated by one of EIO, ENODEV, ENXIO, and EROFS,
although this list can be extended as needed in the future.
As failure-handling relies on the addition of a handful of other
features, the meat of the error-handling callback will come in a later
patch. This patch only serves to surround block IO calls with macros.
Failures elswhere (e.g. when reading instance or tablet metadata) will
be handled in separate patches.
This is part of a series of patches to handle disk failure. See
section 2.2 of:
https://docs.google.com/document/d/1zZk-vb_ETKUuePcZ9ZqoSK2oPvAAaEV1sjDXes8Pxgk/edit?usp=sharing
Change-Id: Ia03bfb711a1b022d7516f4adb37fe9fb28ec949c
Reviewed-on: http://gerrit.cloudera.org:8080/7030
Reviewed-by: Andrew Wong <aw...@cloudera.com>
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Tested-by: Adar Dembo <ad...@cloudera.com>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/a21a50f8
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/a21a50f8
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/a21a50f8
Branch: refs/heads/master
Commit: a21a50f878034767f42af58f1892dfc893d683f9
Parents: 12efb85
Author: Andrew Wong <aw...@cloudera.com>
Authored: Fri May 26 19:55:34 2017 -0700
Committer: Adar Dembo <ad...@cloudera.com>
Committed: Mon Jul 31 21:45:11 2017 +0000
----------------------------------------------------------------------
src/kudu/fs/block_manager.h | 9 +-
src/kudu/fs/error_manager.h | 38 ++++++
src/kudu/fs/file_block_manager.cc | 66 +++++++---
src/kudu/fs/file_block_manager.h | 2 +
src/kudu/fs/log_block_manager.cc | 224 ++++++++++++++++++++-------------
src/kudu/fs/log_block_manager.h | 16 +--
src/kudu/util/status.h | 14 +++
7 files changed, 257 insertions(+), 112 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/a21a50f8/src/kudu/fs/block_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/fs/block_manager.h b/src/kudu/fs/block_manager.h
index 7874879..c3bfc89 100644
--- a/src/kudu/fs/block_manager.h
+++ b/src/kudu/fs/block_manager.h
@@ -15,8 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-#ifndef KUDU_FS_BLOCK_MANAGER_H
-#define KUDU_FS_BLOCK_MANAGER_H
+#pragma once
#include <cstddef>
#include <cstdint>
@@ -43,6 +42,7 @@ namespace fs {
class BlockManager;
class DataDirManager;
+class FsErrorManager;
struct FsReport;
// The smallest unit of Kudu data that is backed by the local filesystem.
@@ -268,6 +268,9 @@ class BlockManager {
// Exposes the underlying DataDirManager.
virtual DataDirManager* dd_manager() = 0;
+
+ // Exposes the FsErrorManager used to handle fs errors.
+ virtual FsErrorManager* error_manager() = 0;
};
// Closes a group of blocks.
@@ -314,5 +317,3 @@ int64_t GetFileCacheCapacityForBlockManager(Env* env);
} // namespace fs
} // namespace kudu
-
-#endif
http://git-wip-us.apache.org/repos/asf/kudu/blob/a21a50f8/src/kudu/fs/error_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/fs/error_manager.h b/src/kudu/fs/error_manager.h
index fff702e..7724e4b 100644
--- a/src/kudu/fs/error_manager.h
+++ b/src/kudu/fs/error_manager.h
@@ -38,6 +38,39 @@ namespace fs {
typedef Callback<void(const std::string&)> ErrorNotificationCb;
static void DoNothingErrorNotification(const std::string& /* uuid */) {}
+// Evaluates the expression and handles it if it results in an error.
+// Returns if the status is an error.
+#define RETURN_NOT_OK_HANDLE_ERROR(status_expr) do { \
+ const Status& _s = (status_expr); \
+ if (PREDICT_TRUE(_s.ok())) { \
+ break; \
+ } \
+ HandleError(_s); \
+ return _s; \
+} while (0);
+
+// Evaluates the expression and runs 'err_handler' if it results in a disk
+// failure. Returns if the expression results in an error.
+#define RETURN_NOT_OK_HANDLE_DISK_FAILURE(status_expr, err_handler) do { \
+ const Status& _s = (status_expr); \
+ if (PREDICT_TRUE(_s.ok())) { \
+ break; \
+ } \
+ if (_s.IsDiskFailure()) { \
+ (err_handler); \
+ } \
+ return _s; \
+} while (0);
+
+// Evaluates the expression and runs 'err_handler' if it results in a disk
+// failure.
+#define HANDLE_DISK_FAILURE(status_expr, err_handler) do { \
+ const Status& _s = (status_expr); \
+ if (PREDICT_FALSE(_s.IsDiskFailure())) { \
+ (err_handler); \
+ } \
+} while (0);
+
// When certain operations fail, the side effects of the error can span
// multiple layers, many of which we prefer to keep separate. The FsErrorManager
// registers and runs error handlers without adding cross-layer dependencies.
@@ -71,6 +104,11 @@ class FsErrorManager {
notify_cb_.Run(uuid);
}
+ // Runs the error notification callback with the UUID of 'dir'.
+ void RunErrorNotificationCb(const DataDir* dir) const {
+ notify_cb_.Run(dir->instance()->metadata()->path_set().uuid());
+ }
+
private:
// Callback to be run when an error occurs.
ErrorNotificationCb notify_cb_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/a21a50f8/src/kudu/fs/file_block_manager.cc
----------------------------------------------------------------------
diff --git a/src/kudu/fs/file_block_manager.cc b/src/kudu/fs/file_block_manager.cc
index 20f7cb4..8744e6a 100644
--- a/src/kudu/fs/file_block_manager.cc
+++ b/src/kudu/fs/file_block_manager.cc
@@ -235,6 +235,8 @@ class FileWritableBlock : public WritableBlock {
virtual State state() const OVERRIDE;
+ void HandleError(const Status& s) const;
+
private:
enum SyncMode {
SYNC,
@@ -284,6 +286,11 @@ FileWritableBlock::~FileWritableBlock() {
}
}
+void FileWritableBlock::HandleError(const Status& s) const {
+ HANDLE_DISK_FAILURE(
+ s, block_manager_->error_manager()->RunErrorNotificationCb(location_.data_dir()));
+}
+
Status FileWritableBlock::Close() {
return Close(SYNC);
}
@@ -306,10 +313,9 @@ Status FileWritableBlock::Append(const Slice& data) {
}
Status FileWritableBlock::AppendV(const vector<Slice>& data) {
- DCHECK(state_ == CLEAN || state_ == DIRTY)
- << "Invalid state: " << state_;
- RETURN_NOT_OK(writer_->AppendV(data));
- RETURN_NOT_OK(location_.data_dir()->RefreshIsFull(
+ DCHECK(state_ == CLEAN || state_ == DIRTY) << "Invalid state: " << state_;
+ RETURN_NOT_OK_HANDLE_ERROR(writer_->AppendV(data));
+ RETURN_NOT_OK_HANDLE_ERROR(location_.data_dir()->RefreshIsFull(
DataDir::RefreshMode::ALWAYS));
state_ = DIRTY;
@@ -327,7 +333,7 @@ Status FileWritableBlock::FlushDataAsync() {
<< "Invalid state: " << state_;
if (state_ == DIRTY) {
VLOG(3) << "Flushing block " << id();
- RETURN_NOT_OK(writer_->Flush(WritableFile::FLUSH_ASYNC));
+ RETURN_NOT_OK_HANDLE_ERROR(writer_->Flush(WritableFile::FLUSH_ASYNC));
}
state_ = FLUSHING;
@@ -370,8 +376,12 @@ Status FileWritableBlock::Close(SyncMode mode) {
block_manager_->metrics_->total_bytes_written->IncrementBy(BytesAppended());
}
+ // Either Close() or Sync() could have run into an error.
+ HandleError(close);
+ HandleError(sync);
+
// Prefer the result of Close() to that of Sync().
- return !close.ok() ? close : sync;
+ return close.ok() ? close : sync;
}
////////////////////////////////////////////////////////////
@@ -385,7 +395,7 @@ Status FileWritableBlock::Close(SyncMode mode) {
// embed a FileBlockLocation, using the simpler BlockId instead.
class FileReadableBlock : public ReadableBlock {
public:
- FileReadableBlock(const FileBlockManager* block_manager, BlockId block_id,
+ FileReadableBlock(FileBlockManager* block_manager, BlockId block_id,
shared_ptr<RandomAccessFile> reader);
virtual ~FileReadableBlock();
@@ -402,9 +412,11 @@ class FileReadableBlock : public ReadableBlock {
virtual size_t memory_footprint() const OVERRIDE;
+ void HandleError(const Status& s) const;
+
private:
// Back pointer to the owning block manager.
- const FileBlockManager* block_manager_;
+ FileBlockManager* block_manager_;
// The block's identifier.
const BlockId block_id_;
@@ -419,7 +431,13 @@ class FileReadableBlock : public ReadableBlock {
DISALLOW_COPY_AND_ASSIGN(FileReadableBlock);
};
-FileReadableBlock::FileReadableBlock(const FileBlockManager* block_manager,
+void FileReadableBlock::HandleError(const Status& s) const {
+ const DataDir* dir = block_manager_->dd_manager()->FindDataDirByUuidIndex(
+ internal::FileBlockLocation::GetDataDirIdx(block_id_));
+ HANDLE_DISK_FAILURE(s, block_manager_->error_manager()->RunErrorNotificationCb(dir));
+}
+
+FileReadableBlock::FileReadableBlock(FileBlockManager* block_manager,
BlockId block_id,
shared_ptr<RandomAccessFile> reader)
: block_manager_(block_manager),
@@ -455,7 +473,8 @@ const BlockId& FileReadableBlock::id() const {
Status FileReadableBlock::Size(uint64_t* sz) const {
DCHECK(!closed_.Load());
- return reader_->Size(sz);
+ RETURN_NOT_OK_HANDLE_ERROR(reader_->Size(sz));
+ return Status::OK();
}
Status FileReadableBlock::Read(uint64_t offset, Slice* result) const {
@@ -466,7 +485,7 @@ Status FileReadableBlock::Read(uint64_t offset, Slice* result) const {
Status FileReadableBlock::ReadV(uint64_t offset, vector<Slice>* results) const {
DCHECK(!closed_.Load());
- RETURN_NOT_OK(reader_->ReadV(offset, results));
+ RETURN_NOT_OK_HANDLE_ERROR(reader_->ReadV(offset, results));
if (block_manager_->metrics_) {
// Calculate the read amount of data
@@ -512,7 +531,8 @@ Status FileBlockManager::SyncMetadata(const internal::FileBlockLocation& locatio
// Sync them.
if (FLAGS_enable_data_block_fsync) {
for (const string& s : to_sync) {
- RETURN_NOT_OK(env_->SyncDir(s));
+ RETURN_NOT_OK_HANDLE_DISK_FAILURE(env_->SyncDir(s),
+ error_manager_->RunErrorNotificationCb(location.data_dir()));
}
}
return Status::OK();
@@ -619,7 +639,12 @@ Status FileBlockManager::CreateBlock(const CreateBlockOptions& opts,
location = internal::FileBlockLocation::FromParts(dir, uuid_idx, id);
path = location.GetFullPath();
- RETURN_NOT_OK_PREPEND(location.CreateBlockDir(env_, &created_dirs), path);
+ s = location.CreateBlockDir(env_, &created_dirs);
+
+ // We could create a block in a different directory, but there's currently
+ // no point in doing so. On disk failure, the tablet specified by 'opts'
+ // will be shut down, so the returned block would not be used.
+ RETURN_NOT_OK_HANDLE_DISK_FAILURE(s, error_manager_->RunErrorNotificationCb(dir));
WritableFileOptions wr_opts;
wr_opts.mode = Env::CREATE_NON_EXISTING;
s = env_util::OpenFileForWrite(wr_opts, env_, path, &writer);
@@ -637,10 +662,19 @@ Status FileBlockManager::CreateBlock(const CreateBlockOptions& opts,
dirty_dirs_.insert(DirName(path));
}
block->reset(new internal::FileWritableBlock(this, location, writer));
+ } else {
+ HANDLE_DISK_FAILURE(s, error_manager_->RunErrorNotificationCb(dir));
+ return s;
}
- return s;
+ return Status::OK();
}
+#define RETURN_NOT_OK_FBM_DISK_FAILURE(status_expr) do { \
+ RETURN_NOT_OK_HANDLE_DISK_FAILURE((status_expr), \
+ error_manager_->RunErrorNotificationCb(dd_manager()->FindDataDirByUuidIndex( \
+ internal::FileBlockLocation::GetDataDirIdx(block_id)))); \
+} while (0);
+
Status FileBlockManager::OpenBlock(const BlockId& block_id,
unique_ptr<ReadableBlock>* block) {
string path;
@@ -652,7 +686,7 @@ Status FileBlockManager::OpenBlock(const BlockId& block_id,
VLOG(1) << "Opening block with id " << block_id.ToString() << " at " << path;
shared_ptr<RandomAccessFile> reader;
- RETURN_NOT_OK(file_cache_.OpenExistingFile(path, &reader));
+ RETURN_NOT_OK_FBM_DISK_FAILURE(file_cache_.OpenExistingFile(path, &reader));
block->reset(new internal::FileReadableBlock(this, block_id, reader));
return Status::OK();
}
@@ -665,7 +699,7 @@ Status FileBlockManager::DeleteBlock(const BlockId& block_id) {
return Status::NotFound(
Substitute("Block $0 not found", block_id.ToString()));
}
- RETURN_NOT_OK(file_cache_.DeleteFile(path));
+ RETURN_NOT_OK_FBM_DISK_FAILURE(file_cache_.DeleteFile(path));
// We don't bother fsyncing the parent directory as there's nothing to be
// gained by ensuring that the deletion is made durable. Even if we did
http://git-wip-us.apache.org/repos/asf/kudu/blob/a21a50f8/src/kudu/fs/file_block_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/fs/file_block_manager.h b/src/kudu/fs/file_block_manager.h
index ac5c735..ad3886e 100644
--- a/src/kudu/fs/file_block_manager.h
+++ b/src/kudu/fs/file_block_manager.h
@@ -94,6 +94,8 @@ class FileBlockManager : public BlockManager {
DataDirManager* dd_manager() override { return &dd_manager_; }
+ FsErrorManager* error_manager() override { return error_manager_; }
+
private:
friend class internal::FileBlockLocation;
friend class internal::FileReadableBlock;
http://git-wip-us.apache.org/repos/asf/kudu/blob/a21a50f8/src/kudu/fs/log_block_manager.cc
----------------------------------------------------------------------
diff --git a/src/kudu/fs/log_block_manager.cc b/src/kudu/fs/log_block_manager.cc
index 0e9def5..9977c3f 100644
--- a/src/kudu/fs/log_block_manager.cc
+++ b/src/kudu/fs/log_block_manager.cc
@@ -28,6 +28,7 @@
#include "kudu/fs/block_manager_metrics.h"
#include "kudu/fs/block_manager_util.h"
#include "kudu/fs/data_dirs.h"
+#include "kudu/fs/error_manager.h"
#include "kudu/fs/fs.pb.h"
#include "kudu/fs/fs_report.h"
#include "kudu/gutil/callback.h"
@@ -378,6 +379,9 @@ class LogBlockContainer {
// Produces a debug-friendly string representation of this container.
string ToString() const;
+ // Handles errors if the input status is not OK.
+ void HandleError(const Status& s) const;
+
// Simple accessors.
LogBlockManager* block_manager() const { return block_manager_; }
int64_t next_block_offset() const { return next_block_offset_; }
@@ -392,7 +396,6 @@ class LogBlockContainer {
}
const LogBlockManagerMetrics* metrics() const { return metrics_; }
const DataDir* data_dir() const { return data_dir_; }
- DataDir* mutable_data_dir() const { return data_dir_; }
const PathInstanceMetadataPB* instance() const { return data_dir_->instance()->metadata(); }
private:
@@ -487,6 +490,16 @@ LogBlockContainer::LogBlockContainer(
metrics_(block_manager->metrics()) {
}
+void LogBlockContainer::HandleError(const Status& s) const {
+ HANDLE_DISK_FAILURE(
+ s, block_manager()->error_manager()->RunErrorNotificationCb(data_dir_));
+}
+
+#define RETURN_NOT_OK_CONTAINER_DISK_FAILURE(status_expr) do { \
+ RETURN_NOT_OK_HANDLE_DISK_FAILURE((status_expr), \
+ block_manager->error_manager()->RunErrorNotificationCb(dir)); \
+} while (0);
+
Status LogBlockContainer::Create(LogBlockManager* block_manager,
DataDir* dir,
unique_ptr<LogBlockContainer>* container) {
@@ -530,15 +543,15 @@ Status LogBlockContainer::Create(LogBlockManager* block_manager,
metadata_writer.reset();
shared_ptr<RWFile> cached_metadata_writer;
- RETURN_NOT_OK(block_manager->file_cache_.OpenExistingFile(
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(block_manager->file_cache_.OpenExistingFile(
metadata_path, &cached_metadata_writer));
metadata_file.reset(new WritablePBContainerFile(
std::move(cached_metadata_writer)));
data_file.reset();
- RETURN_NOT_OK(block_manager->file_cache_.OpenExistingFile(
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(block_manager->file_cache_.OpenExistingFile(
data_path, &cached_data_file));
- RETURN_NOT_OK(metadata_file->CreateNew(BlockRecordPB()));
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(metadata_file->CreateNew(BlockRecordPB()));
container->reset(new LogBlockContainer(block_manager,
dir,
@@ -548,6 +561,8 @@ Status LogBlockContainer::Create(LogBlockManager* block_manager,
}
// Prefer metadata status (arbitrarily).
+ HANDLE_DISK_FAILURE(metadata_status, block_manager->error_manager()->RunErrorNotificationCb(dir));
+ HANDLE_DISK_FAILURE(data_status, block_manager->error_manager()->RunErrorNotificationCb(dir));
return !metadata_status.ok() ? metadata_status : data_status;
}
@@ -571,11 +586,13 @@ Status LogBlockContainer::Open(LogBlockManager* block_manager,
uint64_t data_size = 0;
Status s = env->GetFileSize(metadata_path, &metadata_size);
if (!s.IsNotFound()) {
- RETURN_NOT_OK_PREPEND(s, "unable to determine metadata file size");
+ s = s.CloneAndPrepend("unable to determine metadata file size");
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(s);
}
s = env->GetFileSize(data_path, &data_size);
if (!s.IsNotFound()) {
- RETURN_NOT_OK_PREPEND(s, "unable to determine data file size");
+ s = s.CloneAndPrepend("unable to determine data file size");
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(s);
}
if (metadata_size < pb_util::kPBContainerMinimumValidLength &&
@@ -587,19 +604,18 @@ Status LogBlockContainer::Open(LogBlockManager* block_manager,
// Open the existing metadata and data files for writing.
shared_ptr<RWFile> metadata_file;
- RETURN_NOT_OK(block_manager->file_cache_.OpenExistingFile(
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(block_manager->file_cache_.OpenExistingFile(
metadata_path, &metadata_file));
unique_ptr<WritablePBContainerFile> metadata_pb_writer;
- metadata_pb_writer.reset(new WritablePBContainerFile(
- std::move(metadata_file)));
- RETURN_NOT_OK(metadata_pb_writer->OpenExisting());
+ metadata_pb_writer.reset(new WritablePBContainerFile(std::move(metadata_file)));
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(metadata_pb_writer->OpenExisting());
shared_ptr<RWFile> data_file;
- RETURN_NOT_OK(block_manager->file_cache_.OpenExistingFile(
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(block_manager->file_cache_.OpenExistingFile(
data_path, &data_file));
uint64_t data_file_size;
- RETURN_NOT_OK(data_file->Size(&data_file_size));
+ RETURN_NOT_OK_CONTAINER_DISK_FAILURE(data_file->Size(&data_file_size));
// Create the in-memory container and populate it.
unique_ptr<LogBlockContainer> open_container(new LogBlockContainer(block_manager,
@@ -616,7 +632,7 @@ Status LogBlockContainer::TruncateDataToNextBlockOffset() {
if (full()) {
VLOG(2) << Substitute("Truncating container $0 to offset $1",
ToString(), next_block_offset_);
- RETURN_NOT_OK(data_file_->Truncate(next_block_offset_));
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->Truncate(next_block_offset_));
}
return Status::OK();
}
@@ -629,9 +645,10 @@ Status LogBlockContainer::ProcessRecords(
uint64_t* max_block_id) {
string metadata_path = metadata_file_->filename();
unique_ptr<RandomAccessFile> metadata_reader;
- RETURN_NOT_OK(block_manager()->env()->NewRandomAccessFile(metadata_path, &metadata_reader));
+ RETURN_NOT_OK_HANDLE_ERROR(block_manager()->env()->NewRandomAccessFile(
+ metadata_path, &metadata_reader));
ReadablePBContainerFile pb_reader(std::move(metadata_reader));
- RETURN_NOT_OK(pb_reader.Open());
+ RETURN_NOT_OK_HANDLE_ERROR(pb_reader.Open());
uint64_t data_file_size = 0;
Status read_status;
@@ -660,6 +677,8 @@ Status LogBlockContainer::ProcessRecords(
return Status::OK();
}
// If we've made it here, we've found (and are returning) an unrecoverable error.
+ // Handle any errors we can, e.g. disk failures.
+ HandleError(read_status);
return read_status;
}
@@ -692,7 +711,7 @@ Status LogBlockContainer::ProcessRecord(
// file size is expensive, so we only do it when the metadata indicates
// that additional data has been written to the file.
if (PREDICT_FALSE(record->offset() + record->length() > *data_file_size)) {
- RETURN_NOT_OK(data_file_->Size(data_file_size));
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->Size(data_file_size));
}
// If the record still extends beyond the end of the file, it is malformed.
@@ -799,7 +818,7 @@ Status LogBlockContainer::PunchHole(int64_t offset, int64_t length) {
if (length) {
// It's OK if we exceed the file's total size; the kernel will truncate
// our request.
- return data_file_->PunchHole(offset, length);
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->PunchHole(offset, length));
}
return Status::OK();
}
@@ -811,7 +830,7 @@ Status LogBlockContainer::WriteData(int64_t offset, const Slice& data) {
Status LogBlockContainer::WriteVData(int64_t offset, const vector<Slice>& data) {
DCHECK_GE(offset, next_block_offset_);
- RETURN_NOT_OK(data_file_->WriteV(offset, data));
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->WriteV(offset, data));
// This append may have changed the container size if:
// 1. It was large enough that it blew out the preallocated space.
@@ -821,62 +840,64 @@ Status LogBlockContainer::WriteVData(int64_t offset, const vector<Slice>& data)
return sum + curr.size();
});
if (offset + data_size > preallocated_offset_) {
- RETURN_NOT_OK(data_dir_->RefreshIsFull(DataDir::RefreshMode::ALWAYS));
+ RETURN_NOT_OK_HANDLE_ERROR(data_dir_->RefreshIsFull(DataDir::RefreshMode::ALWAYS));
}
return Status::OK();
}
Status LogBlockContainer::ReadData(int64_t offset, Slice* result) const {
DCHECK_GE(offset, 0);
-
- return data_file_->Read(offset, result);
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->Read(offset, result));
+ return Status::OK();
}
-
Status LogBlockContainer::ReadVData(int64_t offset, vector<Slice>* results) const {
DCHECK_GE(offset, 0);
-
- return data_file_->ReadV(offset, results);
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->ReadV(offset, results));
+ return Status::OK();
}
Status LogBlockContainer::AppendMetadata(const BlockRecordPB& pb) {
// Note: We don't check for sufficient disk space for metadata writes in
// order to allow for block deletion on full disks.
- return metadata_file_->Append(pb);
+ RETURN_NOT_OK_HANDLE_ERROR(metadata_file_->Append(pb));
+ return Status::OK();
}
Status LogBlockContainer::FlushData(int64_t offset, int64_t length) {
DCHECK_GE(offset, 0);
DCHECK_GE(length, 0);
- return data_file_->Flush(RWFile::FLUSH_ASYNC, offset, length);
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->Flush(RWFile::FLUSH_ASYNC, offset, length));
+ return Status::OK();
}
Status LogBlockContainer::FlushMetadata() {
- return metadata_file_->Flush();
+ RETURN_NOT_OK_HANDLE_ERROR(metadata_file_->Flush());
+ return Status::OK();
}
Status LogBlockContainer::SyncData() {
if (FLAGS_enable_data_block_fsync) {
- return data_file_->Sync();
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->Sync());
}
return Status::OK();
}
Status LogBlockContainer::SyncMetadata() {
if (FLAGS_enable_data_block_fsync) {
- return metadata_file_->Sync();
+ RETURN_NOT_OK_HANDLE_ERROR(metadata_file_->Sync());
}
return Status::OK();
}
Status LogBlockContainer::ReopenMetadataWriter() {
shared_ptr<RWFile> f;
- RETURN_NOT_OK(block_manager_->file_cache_.OpenExistingFile(
+ RETURN_NOT_OK_HANDLE_ERROR(block_manager_->file_cache_.OpenExistingFile(
metadata_file_->filename(), &f));
unique_ptr<WritablePBContainerFile> w;
w.reset(new WritablePBContainerFile(std::move(f)));
- RETURN_NOT_OK(w->OpenExisting());
+ RETURN_NOT_OK_HANDLE_ERROR(w->OpenExisting());
- RETURN_NOT_OK(metadata_file_->Close());
+ RETURN_NOT_OK_HANDLE_ERROR(metadata_file_->Close());
metadata_file_.swap(w);
return Status::OK();
}
@@ -895,8 +916,8 @@ Status LogBlockContainer::EnsurePreallocated(int64_t block_start_offset,
next_append_length > preallocated_offset_ - block_start_offset) {
int64_t off = std::max(preallocated_offset_, block_start_offset);
int64_t len = FLAGS_log_container_preallocate_bytes;
- RETURN_NOT_OK(data_file_->PreAllocate(off, len, RWFile::CHANGE_FILE_SIZE));
- RETURN_NOT_OK(data_dir_->RefreshIsFull(DataDir::RefreshMode::ALWAYS));
+ RETURN_NOT_OK_HANDLE_ERROR(data_file_->PreAllocate(off, len, RWFile::CHANGE_FILE_SIZE));
+ RETURN_NOT_OK_HANDLE_ERROR(data_dir_->RefreshIsFull(DataDir::RefreshMode::ALWAYS));
VLOG(2) << Substitute("Preallocated $0 bytes at offset $1 in container $2",
len, off, ToString());
@@ -1522,19 +1543,29 @@ Status LogBlockManager::Open(FsReport* report) {
for (const auto& dd : dd_manager_.data_dirs()) {
dd->WaitOnClosures();
}
+ if (dd_manager_.GetFailedDataDirs().size() == dd_manager_.data_dirs().size()) {
+ return Status::IOError("All data dirs failed to open", "", EIO);
+ }
- // Ensure that no open failed.
- for (const auto& s : statuses) {
- if (!s.ok()) {
+ // Ensure that no open failed without being handled.
+ //
+ // Currently only disk failures are handled. Reports from failed disks are
+ // unusable.
+ FsReport merged_report;
+ for (i = 0; i < statuses.size(); i++) {
+ const Status& s = statuses[i];
+ if (PREDICT_TRUE(s.ok())) {
+ merged_report.MergeFrom(reports[i]);
+ continue;
+ }
+ if (!s.IsDiskFailure()) {
return s;
}
+ LOG(ERROR) << Substitute("Not using report from $0: $1",
+ dd_manager_.data_dirs()[i]->dir(), s.ToString());
}
- // Merge the reports and either return or log the result.
- FsReport merged_report;
- for (const auto& r : reports) {
- merged_report.MergeFrom(r);
- }
+ // Either return or log the report.
if (report) {
*report = std::move(merged_report);
} else {
@@ -1686,10 +1717,13 @@ Status LogBlockManager::GetOrCreateContainer(const CreateBlockOptions& opts,
// All containers are in use; create a new one.
unique_ptr<LogBlockContainer> new_container;
- RETURN_NOT_OK_PREPEND(LogBlockContainer::Create(this,
- dir,
- &new_container),
- "Could not create new log block container at " + dir->dir());
+ Status s = LogBlockContainer::Create(this, dir, &new_container);
+
+ // We could create a container in a different directory, but there's
+ // currently no point in doing so. On disk failure, the tablet specified by
+ // 'opts' will be shut down, so the returned container would not be used.
+ HANDLE_DISK_FAILURE(s, error_manager_->RunErrorNotificationCb(dir));
+ RETURN_NOT_OK_PREPEND(s, "Could not create new log block container at " + dir->dir());
{
std::lock_guard<simple_spinlock> l(lock_);
dirty_dirs_.insert(dir->dir());
@@ -1731,6 +1765,7 @@ Status LogBlockManager::SyncContainer(const LogBlockContainer& container) {
// In the worst case (another block synced this container as we did),
// we'll sync it again needlessly.
if (!s.ok()) {
+ container.HandleError(s);
std::lock_guard<simple_spinlock> l(lock_);
dirty_dirs_.insert(container.data_dir()->dir());
}
@@ -1837,6 +1872,7 @@ void LogBlockManager::OpenDataDir(DataDir* dir,
vector<string> children;
Status s = env_->GetChildren(dir->dir(), &children);
if (!s.ok()) {
+ HANDLE_DISK_FAILURE(s, error_manager_->RunErrorNotificationCb(dir));
*result_status = s.CloneAndPrepend(Substitute(
"Could not list children of $0", dir->dir()));
return;
@@ -1989,6 +2025,7 @@ void LogBlockManager::OpenDataDir(DataDir* dir,
uint64_t reported_size;
s = env_->GetFileSizeOnDisk(data_filename, &reported_size);
if (!s.ok()) {
+ HANDLE_DISK_FAILURE(s, error_manager_->RunErrorNotificationCb(dir));
*result_status = s.CloneAndPrepend(Substitute(
"Could not get on-disk file size of container $0", container->ToString()));
return;
@@ -2067,6 +2104,18 @@ void LogBlockManager::OpenDataDir(DataDir* dir,
*result_status = Status::OK();
}
+#define RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(status_expr, msg) do { \
+ Status s_ = (status_expr); \
+ s_ = s_.CloneAndPrepend(msg); \
+ RETURN_NOT_OK_HANDLE_DISK_FAILURE(s_, error_manager_->RunErrorNotificationCb(dir)); \
+} while (0);
+
+#define WARN_NOT_OK_LBM_DISK_FAILURE(status_expr, msg) do { \
+ Status s_ = (status_expr); \
+ HANDLE_DISK_FAILURE(s_, error_manager_->RunErrorNotificationCb(dir)); \
+ WARN_NOT_OK(s_, msg); \
+} while (0);
+
Status LogBlockManager::Repair(
DataDir* dir,
FsReport* report,
@@ -2144,16 +2193,17 @@ Status LogBlockManager::Repair(
if (s.ok()) {
deleted_metadata_bytes += metadata_size;
} else {
- WARN_NOT_OK(s, "Could not get size of dead container metadata file " + metadata_file_name);
+ WARN_NOT_OK_LBM_DISK_FAILURE(s,
+ "Could not get size of dead container metadata file " + metadata_file_name);
}
- WARN_NOT_OK(file_cache_.DeleteFile(data_file_name),
+ WARN_NOT_OK_LBM_DISK_FAILURE(file_cache_.DeleteFile(data_file_name),
"Could not delete dead container data file " + data_file_name);
- WARN_NOT_OK(file_cache_.DeleteFile(metadata_file_name),
+ WARN_NOT_OK_LBM_DISK_FAILURE(file_cache_.DeleteFile(metadata_file_name),
"Could not delete dead container metadata file " + metadata_file_name);
}
if (!dead_containers.empty()) {
- WARN_NOT_OK(env_->SyncDir(dir->dir()), "Could not sync data directory");
+ WARN_NOT_OK_LBM_DISK_FAILURE(env_->SyncDir(dir->dir()), "Could not sync data directory");
LOG(INFO) << Substitute("Deleted $0 dead containers ($1 metadata bytes)",
dead_containers.size(), deleted_metadata_bytes);
}
@@ -2173,27 +2223,25 @@ Status LogBlockManager::Repair(
pr.repaired = true;
continue;
}
- RETURN_NOT_OK_PREPEND(
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(
env_->NewRWFile(opts,
StrCat(pr.container, kContainerMetadataFileSuffix),
&file),
"could not reopen container to truncate partial metadata record");
- RETURN_NOT_OK_PREPEND(file->Truncate(pr.offset),
- "could not truncate partial metadata record");
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(file->Truncate(pr.offset),
+ "could not truncate partial metadata record");
// Technically we've "repaired" the inconsistency if the truncation
// succeeded, even if the following logic fails.
pr.repaired = true;
- RETURN_NOT_OK_PREPEND(
- file->Close(),
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(file->Close(),
"could not close container after truncating partial metadata record");
// Reopen the PB writer so that it will refresh its metadata about the
// underlying file and resume appending to the new end of the file.
- RETURN_NOT_OK_PREPEND(
- container->ReopenMetadataWriter(),
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(container->ReopenMetadataWriter(),
"could not reopen container metadata file");
}
}
@@ -2207,12 +2255,12 @@ Status LogBlockManager::Repair(
Status s = env_->DeleteFile(
StrCat(ic.container, kContainerMetadataFileSuffix));
if (!s.ok() && !s.IsNotFound()) {
- WARN_NOT_OK(s, "could not delete incomplete container metadata file");
+ WARN_NOT_OK_LBM_DISK_FAILURE(s, "could not delete incomplete container metadata file");
}
s = env_->DeleteFile(StrCat(ic.container, kContainerDataFileSuffix));
if (!s.ok() && !s.IsNotFound()) {
- WARN_NOT_OK(s, "could not delete incomplete container data file");
+ WARN_NOT_OK_LBM_DISK_FAILURE(s, "could not delete incomplete container data file");
}
ic.repaired = true;
}
@@ -2268,7 +2316,7 @@ Status LogBlockManager::Repair(
// Rewrite this metadata file. Failures are non-fatal.
int64_t file_bytes_delta;
const auto& meta_path = StrCat(e.first, kContainerMetadataFileSuffix);
- Status s = RewriteMetadataFile(meta_path, e.second, &file_bytes_delta);
+ Status s = RewriteMetadataFile(*container, e.second, &file_bytes_delta);
if (!s.ok()) {
WARN_NOT_OK(s, "could not rewrite metadata file");
continue;
@@ -2287,13 +2335,17 @@ Status LogBlockManager::Repair(
// The data directory can be synchronized once for all of the new metadata files.
//
- // Failures are fatal: if a new metadata file doesn't durably exist in the
- // data directory, it would be unsafe to append new block records to it. This
- // is because after a crash the old metadata file may appear instead, and
- // that file lacks the newly appended block records.
+ // Non-disk failures are fatal: if a new metadata file doesn't durably exist
+ // in the data directory, it would be unsafe to append new block records to
+ // it. This is because after a crash the old metadata file may appear
+ // instead, and that file lacks the newly appended block records.
+ //
+ // TODO(awong): The below will only be true with persistent disk states.
+ // Disk failures do not suffer from this issue because, on the next startup,
+ // the entire directory will not be used.
if (metadata_files_compacted > 0) {
- RETURN_NOT_OK_PREPEND(env_->SyncDir(dir->dir()),
- "Could not sync data directory");
+ Status s = env_->SyncDir(dir->dir());
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(s, "Could not sync data directory");
LOG(INFO) << Substitute("Compacted $0 metadata files ($1 metadata bytes)",
metadata_files_compacted, metadata_bytes_delta);
}
@@ -2301,12 +2353,16 @@ Status LogBlockManager::Repair(
return Status::OK();
}
-Status LogBlockManager::RewriteMetadataFile(const string& metadata_file_name,
+Status LogBlockManager::RewriteMetadataFile(const LogBlockContainer& container,
const vector<BlockRecordPB>& records,
int64_t* file_bytes_delta) {
uint64_t old_metadata_size;
- RETURN_NOT_OK_PREPEND(env_->GetFileSize(metadata_file_name, &old_metadata_size),
- "could not get size of old metadata file");
+ const string metadata_file_name = StrCat(container.ToString(), kContainerMetadataFileSuffix);
+
+ // Get the container's data directory's UUID for error handling.
+ const string dir = container.data_dir()->dir();
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(env_->GetFileSize(metadata_file_name, &old_metadata_size),
+ "could not get size of old metadata file");
// Create a new container metadata file with only the live block records.
//
@@ -2316,26 +2372,26 @@ Status LogBlockManager::RewriteMetadataFile(const string& metadata_file_name,
string tmpl = metadata_file_name + kTmpInfix + ".XXXXXX";
unique_ptr<RWFile> tmp_file;
string tmp_file_name;
- RETURN_NOT_OK_PREPEND(env_->NewTempRWFile(RWFileOptions(), tmpl,
- &tmp_file_name, &tmp_file),
- "could not create temporary metadata file");
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(env_->NewTempRWFile(RWFileOptions(), tmpl,
+ &tmp_file_name, &tmp_file),
+ "could not create temporary metadata file");
env_util::ScopedFileDeleter tmp_deleter(env_, tmp_file_name);
WritablePBContainerFile pb_file(std::move(tmp_file));
- RETURN_NOT_OK_PREPEND(pb_file.CreateNew(BlockRecordPB()),
- "could not initialize temporary metadata file");
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(pb_file.CreateNew(BlockRecordPB()),
+ "could not initialize temporary metadata file");
for (const auto& r : records) {
- RETURN_NOT_OK_PREPEND(pb_file.Append(r),
- "could not append to temporary metadata file");
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(pb_file.Append(r),
+ "could not append to temporary metadata file");
}
- RETURN_NOT_OK_PREPEND(pb_file.Sync(),
- "could not sync temporary metadata file");
- RETURN_NOT_OK_PREPEND(pb_file.Close(),
- "could not close temporary metadata file");
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(pb_file.Sync(),
+ "could not sync temporary metadata file");
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(pb_file.Close(),
+ "could not close temporary metadata file");
uint64_t new_metadata_size;
- RETURN_NOT_OK_PREPEND(env_->GetFileSize(tmp_file_name, &new_metadata_size),
- "could not get file size of temporary metadata file");
- RETURN_NOT_OK_PREPEND(env_->RenameFile(tmp_file_name, metadata_file_name),
- "could not rename temporary metadata file");
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(env_->GetFileSize(tmp_file_name, &new_metadata_size),
+ "could not get file size of temporary metadata file");
+ RETURN_NOT_OK_LBM_DISK_FAILURE_PREPEND(env_->RenameFile(tmp_file_name, metadata_file_name),
+ "could not rename temporary metadata file");
// Evict the old path from the file cache, so that when we re-open the new
// metadata file for write, we don't accidentally get a cache hit on the
// old file descriptor pointing to the now-deleted old version.
http://git-wip-us.apache.org/repos/asf/kudu/blob/a21a50f8/src/kudu/fs/log_block_manager.h
----------------------------------------------------------------------
diff --git a/src/kudu/fs/log_block_manager.h b/src/kudu/fs/log_block_manager.h
index d485f99..7d31844 100644
--- a/src/kudu/fs/log_block_manager.h
+++ b/src/kudu/fs/log_block_manager.h
@@ -15,8 +15,7 @@
// specific language governing permissions and limitations
// under the License.
-#ifndef KUDU_FS_LOG_BLOCK_MANAGER_H
-#define KUDU_FS_LOG_BLOCK_MANAGER_H
+#pragma once
#include <deque>
#include <map>
@@ -188,7 +187,9 @@ class LogBlockManager : public BlockManager {
Status GetAllBlockIds(std::vector<BlockId>* block_ids) override;
- DataDirManager* dd_manager() override { return &dd_manager_; };
+ DataDirManager* dd_manager() override { return &dd_manager_; }
+
+ FsErrorManager* error_manager() override { return error_manager_; }
private:
FRIEND_TEST(LogBlockManagerTest, TestLookupBlockLimit);
@@ -308,15 +309,15 @@ class LogBlockManager : public BlockManager {
std::string,
std::vector<BlockRecordPB>> low_live_block_containers);
- // Rewrites a container metadata file with name 'metadata_file_name',
- // appending all entries in 'records'. The new metadata file is created as a
- // temporary file and renamed over the existing file after it is fully written.
+ // Rewrites a container metadata file, appending all entries in 'records'.
+ // The new metadata file is created as a temporary file and renamed over the
+ // existing file after it is fully written.
//
// On success, writes the difference in file sizes to 'file_bytes_delta'. On
// failure, an effort is made to delete the temporary file.
//
// Note: the new file is synced but its parent directory is not.
- Status RewriteMetadataFile(const std::string& metadata_file_name,
+ Status RewriteMetadataFile(const internal::LogBlockContainer& container,
const std::vector<BlockRecordPB>& records,
int64_t* file_bytes_delta);
@@ -427,4 +428,3 @@ class LogBlockManager : public BlockManager {
} // namespace fs
} // namespace kudu
-#endif
http://git-wip-us.apache.org/repos/asf/kudu/blob/a21a50f8/src/kudu/util/status.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/status.h b/src/kudu/util/status.h
index 5028392..ff65b57 100644
--- a/src/kudu/util/status.h
+++ b/src/kudu/util/status.h
@@ -13,7 +13,9 @@
#ifndef KUDU_UTIL_STATUS_H_
#define KUDU_UTIL_STATUS_H_
+#include <errno.h>
#include <stdint.h>
+
#include <string>
#ifdef KUDU_HEADERS_NO_STUBS
@@ -337,6 +339,18 @@ class KUDU_EXPORT Status {
/// @return @c true iff the status indicates end of file.
bool IsEndOfFile() const { return code() == kEndOfFile; }
+ /// @return @c true iff the status indicates a disk failure.
+ bool IsDiskFailure() const {
+ switch (posix_code()) {
+ case EIO:
+ case ENODEV:
+ case ENXIO:
+ case EROFS:
+ return true;
+ }
+ return false;
+ }
+
/// @return A string representation of this status suitable for printing.
/// Returns the string "OK" for success.
std::string ToString() const;