jabberwocky238/rocksdb
1
0
Fork 0
mirror of https://github.com/facebook/rocksdb.git synced 2026-07-07 14:47:40 +08:00
Code Issues Packages Projects Releases Wiki Activity

Persist fault injection logs and fail fast on expected-state trace writes (#14651)

Browse Source 
Summary:
- switch fault injection error recording from an in-memory ring buffer to per-run fixed-record binary logs under `TEST_TMPDIR/fault_injection_logs` (or `/tmp/fault_injection_logs`) so crash paths survive DB reopen cleanup
- keep the raw and decoded fault logs for external artifact collection/cleanup, and make `db_crashtest` print consistent blackbox/whitebox summaries after decoding
- make expected-state tracing fail fast on trace write failures and document offline trace inspection via `trace_analyzer`
- add coverage for binary log persistence/decoding/truncated-tail handling and keep info logs excluded from fault injection

Reviewed By: hx235

Differential Revision: D101973626

fbshipit-source-id: fdcb5b6370cf92a046e09b8d3391e80eecb66c23
This commit is contained in:
Xingbo Wang
2026-06-20 17:20:14 -07:00
committed by meta-codesync[bot]
parent 5bf78183db
commit 214869aacd
13 changed files with 1102 additions and 949 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db_stress_tool/db_stress_gflags.cc
-14
View File
@@ -706,20 +706,6 @@ DEFINE_string(
"only tracked when --sync_fault_injection is set. See --seed and "
"--nooverwritepercent for further requirements.");
DEFINE_bool(expected_state_trace_debug, true,
"If true, print debug logs while replaying expected-state trace "
"records during crash recovery verification.");
DEFINE_int64(
expected_state_trace_debug_key, -1,
"If non-negative, restrict expected-state trace debug logs to the "
"specified logical key where possible. Raw-key roundtrip mismatches for "
"that logical key are still logged.");
DEFINE_int32(expected_state_trace_debug_max_logs, 200,
"Maximum number of expected-state trace debug log lines to emit "
"per restore attempt.");
DEFINE_bool(verify_checksum, false,
"Verify checksum for every block read from storage");
db_stress_tool/db_stress_test_base.cc
+30 -18
View File
@@ -8,6 +8,7 @@
// found in the LICENSE file. See the AUTHORS file for names of contributors.
//
#include <cstdlib>
#include <ios>
#include <memory>
#include <mutex>
@@ -30,6 +31,7 @@
#include "db_stress_tool/db_stress_wide_merge_operator.h"
#include "file/file_util.h"
#include "options/options_parser.h"
#include "port/port.h"
#include "rocksdb/convenience.h"
#include "rocksdb/filter_policy.h"
#include "rocksdb/io_dispatcher.h"
@@ -258,6 +260,32 @@ bool NeedsFaultInjection() {
FLAGS_write_fault_one_in || FLAGS_sync_fault_injection;
}
std::string GetFaultInjectionLogBaseDir() {
if (!FLAGS_env_uri.empty() || !FLAGS_fs_uri.empty()) {
return "/tmp";
}
// db_stress reads environment variables during single-threaded startup,
// before worker threads are created.
// NOLINTNEXTLINE(concurrency-mt-unsafe)
const char* test_tmpdir = std::getenv("TEST_TMPDIR");
return test_tmpdir != nullptr && test_tmpdir[0] != '\0' ? test_tmpdir
: "/tmp";
}
std::string GetFaultInjectionLogPath(const std::string& db_label) {
const std::string log_dir =
GetFaultInjectionLogBaseDir() + "/fault_injection_logs";
Status s = Env::Default()->CreateDirIfMissing(log_dir);
if (!s.ok()) {
fprintf(stderr, "Failed to create directory %s: %s\n", log_dir.c_str(),
s.ToString().c_str());
exit(1);
}
return log_dir + "/fault_injection_" + std::to_string(port::GetProcessID()) +
"_" + std::to_string(Env::Default()->NowMicros()) + "_" + db_label +
".bin";
}
} // namespace
const std::string& StressTest::GetDbLabel() const { return db_label_; }
@@ -285,7 +313,8 @@ StressTest::StressTest(int db_index, const std::string& db_path,
std::make_shared<DbStressFSWrapper>(raw_env->GetFileSystem())),
db_fault_injection_fs_(
NeedsFaultInjection()
? std::make_shared<FaultInjectionTestFS>(db_stress_fs_)
? std::make_shared<FaultInjectionTestFS>(
db_stress_fs_, GetFaultInjectionLogPath(db_label_))
: nullptr),
db_env_(std::make_unique<CompositeEnvWrapper>(
raw_env, db_fault_injection_fs_
@@ -314,23 +343,6 @@ StressTest::StressTest(int db_index, const std::string& db_path,
// StressTest::Open() for open fault injection and in RunStressTestImpl()
// for proper fault injection setup.
db_fault_injection_fs_->SetFilesystemDirectWritable(true);
// Set the fault injection log file path so that PrintAll() writes to a
// file instead of stderr (signal-safe). PrintAll() opens this path with
// plain POSIX open(), not through raw_env, so the log path must stay on
// the local filesystem. Store it in the local test directory (TEST_TMPDIR
// via Env::Default()) outside the DB directory so it survives DB reopen
// and gets included in the sandcastle db.tar.gz artifact for post-failure
// analysis.
std::string log_dir;
if (!Env::Default()->GetTestDirectory(&log_dir).ok() || log_dir.empty()) {
log_dir = "/tmp";
}
std::string log_path = log_dir + "/fault_injection_" +
std::to_string(getpid()) + "_" +
std::to_string(Env::Default()->NowMicros()) + "_" +
GetDbLabel() + ".log";
db_fault_injection_fs_->SetInjectedErrorLogPath(log_path);
}
if (FLAGS_destroy_db_initially) {
db_stress_tool/db_stress_tool.cc
+7 -7
View File
@@ -38,7 +38,7 @@ static std::shared_ptr<ROCKSDB_NAMESPACE::Env> legacy_env_wrapper_guard;
// Raw pointers for signal-safe crash callback. Signal handlers can only
// access file-static/global variables; can't capture StressTest instances.
static std::vector<ROCKSDB_NAMESPACE::FaultInjectionTestFS*>
fault_fs_for_crash_report;
fault_fs_for_crash_flush;
int ValidateNumDbsFlags() {
if (FLAGS_num_dbs < 1) {
@@ -88,20 +88,20 @@ int DestroyAllDbs() {
void RegisterCrashCallbacks(
const std::vector<std::unique_ptr<StressTest>>& stress_tests, int num_dbs) {
fault_fs_for_crash_report.resize(num_dbs, nullptr);
fault_fs_for_crash_flush.resize(num_dbs, nullptr);
bool any_fault_fs = false;
for (int i = 0; i < num_dbs; i++) {
fault_fs_for_crash_report[i] =
fault_fs_for_crash_flush[i] =
stress_tests[i]->GetDbFaultInjectionFs().get();
if (fault_fs_for_crash_report[i]) {
if (fault_fs_for_crash_flush[i]) {
any_fault_fs = true;
}
}
if (any_fault_fs) {
port::RegisterCrashCallback([]() {
for (auto* fs : fault_fs_for_crash_report) {
for (auto* fs : fault_fs_for_crash_flush) {
if (fs) {
fs->PrintRecentInjectedErrors();
fs->FlushRecentInjectedErrors();
}
}
});
@@ -633,7 +633,7 @@ int db_stress_tool(int argc, char** argv) {
}
stress_tests[i]->CleanUp();
}
for (auto& fs : fault_fs_for_crash_report) {
for (auto& fs : fault_fs_for_crash_flush) {
fs = nullptr;
}
return all_passed ? 0 : 1;
db_stress_tool/expected_state.cc
+90 -446
View File
@@ -4,8 +4,8 @@
// (found in the LICENSE.Apache file in the root directory).
#include <atomic>
#include <iomanip>
#include <sstream>
#include <cstdio>
#include <cstdlib>
#ifdef GFLAGS
#include "db/wide/wide_column_serialization.h"
@@ -348,6 +348,42 @@ Status FileExpectedStateManager::Open() {
return s;
}
namespace {
class FatalExpectedStateTraceWriter : public TraceWriter {
public:
FatalExpectedStateTraceWriter(std::string trace_file_path,
std::unique_ptr<TraceWriter>&& target)
: trace_file_path_(std::move(trace_file_path)),
target_(std::move(target)) {
assert(target_ != nullptr);
}
Status Write(const Slice& data) override {
Status s = target_->Write(data);
if (!s.ok()) {
// Expected-state tracing is part of crash-recovery verification, not
// best-effort observability. Stop immediately before history diverges.
fprintf(stderr, "Fatal expected-state trace write failure for %s: %s\n",
trace_file_path_.c_str(), s.ToString().c_str());
fflush(stderr);
fflush(stdout);
std::_Exit(1);
}
return s;
}
Status Close() override { return target_->Close(); }
uint64_t GetFileSize() override { return target_->GetFileSize(); }
private:
const std::string trace_file_path_;
std::unique_ptr<TraceWriter> target_;
};
} // anonymous namespace
Status FileExpectedStateManager::SaveAtAndAfter(DB* db) {
SequenceNumber seqno = db->GetLatestSequenceNumber();
@@ -390,6 +426,10 @@ Status FileExpectedStateManager::SaveAtAndAfter(DB* db) {
soptions.writable_file_max_buffer_size = 0;
s = NewFileTraceWriter(Env::Default(), soptions, trace_file_path,
&trace_writer);
if (s.ok()) {
trace_writer.reset(new FatalExpectedStateTraceWriter(
trace_file_path, std::move(trace_writer)));
}
}
if (s.ok()) {
TraceOptions trace_opts;
@@ -427,67 +467,6 @@ bool FileExpectedStateManager::HasHistory() {
namespace {
std::string DescribeExpectedValue(const ExpectedValue& value) {
std::ostringstream oss;
oss << "{raw=0x" << std::hex << std::setw(8) << std::setfill('0')
<< value.Read() << std::dec << std::setfill(' ')
<< " value_base=" << value.GetValueBase()
<< " next_value_base=" << value.NextValueBase()
<< " del_counter=" << value.GetDelCounter()
<< " pending_write=" << value.PendingWrite()
<< " pending_delete=" << value.PendingDelete()
<< " deleted=" << value.IsDeleted() << "}";
return oss.str();
}
size_t CountTrailingXs(const std::string& key) {
size_t trailing_xs = 0;
while (trailing_xs < key.size() && key[key.size() - trailing_xs - 1] == 'x') {
++trailing_xs;
}
return trailing_xs;
}
struct TraceKeyDebugInfo {
std::string raw_key;
std::string raw_key_hex;
bool parse_ok = false;
uint64_t parsed_key_id = 0;
std::string roundtrip_key;
std::string roundtrip_key_hex;
bool roundtrip_matches_raw = false;
bool raw_matches_focus_key = false;
bool parsed_matches_focus_key = false;
bool roundtrip_matches_focus_key = false;
size_t trailing_bytes = 0;
size_t trailing_xs = 0;
bool MatchesFocusKey() const {
return raw_matches_focus_key || parsed_matches_focus_key ||
roundtrip_matches_focus_key;
}
};
std::string DescribeTraceKeyDebugInfo(const TraceKeyDebugInfo& info) {
std::ostringstream oss;
oss << "{raw_key=" << info.raw_key_hex << " size=" << info.raw_key.size()
<< " trailing_bytes=" << info.trailing_bytes
<< " trailing_xs=" << info.trailing_xs << " parse_ok=" << info.parse_ok;
if (info.parse_ok) {
oss << " parsed_key=" << info.parsed_key_id
<< " roundtrip_key=" << info.roundtrip_key_hex
<< " roundtrip_matches_raw=" << info.roundtrip_matches_raw;
}
if (FLAGS_expected_state_trace_debug_key >= 0) {
oss << " focus_key=" << FLAGS_expected_state_trace_debug_key
<< " raw_matches_focus_key=" << info.raw_matches_focus_key
<< " parsed_matches_focus_key=" << info.parsed_matches_focus_key
<< " roundtrip_matches_focus_key=" << info.roundtrip_matches_focus_key;
}
oss << "}";
return oss.str();
}
// An `ExpectedStateTraceRecordHandler` applies a configurable number of traced
// write operations to the configured expected state. It is used in
// `FileExpectedStateManager::Restore()` to sync the expected state with the
@@ -498,25 +477,12 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
ExpectedStateTraceRecordHandler(uint64_t max_write_ops, ExpectedState* state)
: max_write_ops_(max_write_ops),
state_(state),
debug_enabled_(FLAGS_expected_state_trace_debug),
debug_focus_key_(FLAGS_expected_state_trace_debug_key),
debug_focus_key_raw_(debug_focus_key_ >= 0 ? Key(debug_focus_key_)
: std::string()),
debug_max_logs_(static_cast<uint64_t>(
std::max(0, FLAGS_expected_state_trace_debug_max_logs))),
buffered_writes_(nullptr) {}
// True if we have already reached the limit on write operations to apply.
bool IsDone() const { return num_write_ops_ >= max_write_ops_; }
uint64_t NumWriteOps() const { return num_write_ops_; }
uint64_t NumKeyDecodeFailures() const { return key_decode_failures_; }
uint64_t NumKeyRoundtripMismatches() const {
return key_roundtrip_mismatches_;
}
uint64_t NumFocusKeyOpHits() const { return focus_key_op_hits_; }
uint64_t NumLogsEmitted() const { return emitted_debug_logs_; }
uint64_t NumLogsSuppressed() const { return suppressed_debug_logs_; }
bool Continue() override { return !IsDone(); }
@@ -556,37 +522,21 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
Slice key =
StripTimestampFromUserKey(key_with_ts, FLAGS_user_timestamp_size);
uint64_t key_id = 0;
TraceKeyDebugInfo key_info;
Status status =
ParseTracedKey(key, "unable to parse key", &key_id, &key_info);
if (status.ok()) {
Status status = ParseTracedKey(key, "unable to parse key", &key_id);
if (!status.ok()) {
return status;
}
const int64_t expected_key_id = static_cast<int64_t>(key_id);
const uint32_t value_base = GetValueBase(value);
bool should_buffer_write = !(buffered_writes_ == nullptr);
if (should_buffer_write) {
MaybeLogKeyOperation("PutCF", column_family_id, true /* buffered */,
key_info,
"value_base=" + std::to_string(value_base) +
" value_size=" + std::to_string(value.size()));
if (buffered_writes_ != nullptr) {
return WriteBatchInternal::Put(buffered_writes_.get(), column_family_id,
key, value);
}
const ExpectedValue before =
state_->Get(column_family_id, expected_key_id);
state_->SyncPut(column_family_id, expected_key_id, value_base);
const ExpectedValue after =
state_->Get(column_family_id, expected_key_id);
NoteWriteOpApplied();
MaybeLogKeyOperation("PutCF", column_family_id, false /* buffered */,
key_info,
"value_base=" + std::to_string(value_base) +
" value_size=" + std::to_string(value.size()),
&before, &after);
status = Status::OK();
}
return status;
return Status::OK();
}
Status TimedPutCF(uint32_t column_family_id, const Slice& key_with_ts,
@@ -594,40 +544,22 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
Slice key =
StripTimestampFromUserKey(key_with_ts, FLAGS_user_timestamp_size);
uint64_t key_id = 0;
TraceKeyDebugInfo key_info;
Status status =
ParseTracedKey(key, "unable to parse key", &key_id, &key_info);
if (status.ok()) {
Status status = ParseTracedKey(key, "unable to parse key", &key_id);
if (!status.ok()) {
return status;
}
const int64_t expected_key_id = static_cast<int64_t>(key_id);
const uint32_t value_base = GetValueBase(value);
bool should_buffer_write = !(buffered_writes_ == nullptr);
if (should_buffer_write) {
MaybeLogKeyOperation(
"TimedPutCF", column_family_id, true /* buffered */, key_info,
"value_base=" + std::to_string(value_base) +
" value_size=" + std::to_string(value.size()) +
" write_unix_time=" + std::to_string(write_unix_time));
if (buffered_writes_ != nullptr) {
return WriteBatchInternal::TimedPut(buffered_writes_.get(),
column_family_id, key, value,
write_unix_time);
}
const ExpectedValue before =
state_->Get(column_family_id, expected_key_id);
state_->SyncPut(column_family_id, expected_key_id, value_base);
const ExpectedValue after =
state_->Get(column_family_id, expected_key_id);
NoteWriteOpApplied();
MaybeLogKeyOperation(
"TimedPutCF", column_family_id, false /* buffered */, key_info,
"value_base=" + std::to_string(value_base) +
" value_size=" + std::to_string(value.size()) +
" write_unix_time=" + std::to_string(write_unix_time),
&before, &after);
status = Status::OK();
}
return status;
return Status::OK();
}
Status PutEntityCF(uint32_t column_family_id, const Slice& key_with_ts,
@@ -636,10 +568,10 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
StripTimestampFromUserKey(key_with_ts, FLAGS_user_timestamp_size);
uint64_t key_id = 0;
TraceKeyDebugInfo key_info;
Status status =
ParseTracedKey(key, "Unable to parse key", &key_id, &key_info);
if (status.ok()) {
Status status = ParseTracedKey(key, "Unable to parse key", &key_id);
if (!status.ok()) {
return status;
}
const int64_t expected_key_id = static_cast<int64_t>(key_id);
Slice entity_copy = entity;
@@ -655,33 +587,15 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
}
if (buffered_writes_) {
MaybeLogKeyOperation(
"PutEntityCF", column_family_id, true /* buffered */, key_info,
"entity_size=" + std::to_string(entity.size()) +
" num_columns=" + std::to_string(columns.size()));
return WriteBatchInternal::PutEntity(buffered_writes_.get(),
column_family_id, key, columns);
}
const uint32_t value_base =
GetValueBase(WideColumnsHelper::GetDefaultColumn(columns));
const ExpectedValue before =
state_->Get(column_family_id, expected_key_id);
state_->SyncPut(column_family_id, expected_key_id, value_base);
const ExpectedValue after =
state_->Get(column_family_id, expected_key_id);
NoteWriteOpApplied();
MaybeLogKeyOperation(
"PutEntityCF", column_family_id, false /* buffered */, key_info,
"entity_size=" + std::to_string(entity.size()) +
" num_columns=" + std::to_string(columns.size()) +
" default_value_base=" + std::to_string(value_base),
&before, &after);
status = Status::OK();
}
return status;
return Status::OK();
}
Status DeleteCF(uint32_t column_family_id,
@@ -689,31 +603,20 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
Slice key =
StripTimestampFromUserKey(key_with_ts, FLAGS_user_timestamp_size);
uint64_t key_id = 0;
TraceKeyDebugInfo key_info;
Status status =
ParseTracedKey(key, "unable to parse key", &key_id, &key_info);
if (status.ok()) {
Status status = ParseTracedKey(key, "unable to parse key", &key_id);
if (!status.ok()) {
return status;
}
const int64_t expected_key_id = static_cast<int64_t>(key_id);
bool should_buffer_write = !(buffered_writes_ == nullptr);
if (should_buffer_write) {
MaybeLogKeyOperation("DeleteCF", column_family_id, true /* buffered */,
key_info, "");
if (buffered_writes_ != nullptr) {
return WriteBatchInternal::Delete(buffered_writes_.get(),
column_family_id, key);
}
const ExpectedValue before =
state_->Get(column_family_id, expected_key_id);
state_->SyncDelete(column_family_id, expected_key_id);
const ExpectedValue after =
state_->Get(column_family_id, expected_key_id);
NoteWriteOpApplied();
MaybeLogKeyOperation("DeleteCF", column_family_id, false /* buffered */,
key_info, "", &before, &after);
status = Status::OK();
}
return status;
return Status::OK();
}
Status SingleDeleteCF(uint32_t column_family_id,
@@ -742,56 +645,25 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
StripTimestampFromUserKey(end_key_with_ts, FLAGS_user_timestamp_size);
uint64_t begin_key_id = 0;
uint64_t end_key_id = 0;
TraceKeyDebugInfo begin_info;
TraceKeyDebugInfo end_info;
Status status = ParseTracedKey(begin_key, "unable to parse begin key",
&begin_key_id, &begin_info);
Status status =
ParseTracedKey(begin_key, "unable to parse begin key", &begin_key_id);
if (status.ok()) {
status = ParseTracedKey(end_key, "unable to parse end key", &end_key_id,
&end_info);
status = ParseTracedKey(end_key, "unable to parse end key", &end_key_id);
}
if (status.ok()) {
bool should_buffer_write = !(buffered_writes_ == nullptr);
if (should_buffer_write) {
const uint64_t affected_keys =
end_key_id > begin_key_id ? end_key_id - begin_key_id : 0;
MaybeLogRangeOperation(
"DeleteRangeCF", column_family_id, true /* buffered */, begin_info,
end_info,
"affected_keys=" + std::to_string(affected_keys) +
" inverted_range=" +
std::to_string(end_key_id < begin_key_id ? 1 : 0));
if (!status.ok()) {
return status;
}
if (buffered_writes_ != nullptr) {
return WriteBatchInternal::DeleteRange(
buffered_writes_.get(), column_family_id, begin_key, end_key);
}
const bool focus_in_range = FocusKeyInRange(begin_key_id, end_key_id);
const uint64_t affected_keys =
end_key_id > begin_key_id ? end_key_id - begin_key_id : 0;
ExpectedValue focus_before;
ExpectedValue focus_after;
if (focus_in_range) {
focus_before = state_->Get(column_family_id, debug_focus_key_);
}
state_->SyncDeleteRange(column_family_id,
static_cast<int64_t>(begin_key_id),
static_cast<int64_t>(end_key_id));
if (focus_in_range) {
focus_after = state_->Get(column_family_id, debug_focus_key_);
}
NoteWriteOpApplied();
MaybeLogRangeOperation(
"DeleteRangeCF", column_family_id, false /* buffered */, begin_info,
end_info,
"affected_keys=" + std::to_string(affected_keys) +
" inverted_range=" +
std::to_string(end_key_id < begin_key_id ? 1 : 0) +
" focus_in_range=" + std::to_string(focus_in_range ? 1 : 0),
focus_in_range ? &focus_before : nullptr,
focus_in_range ? &focus_after : nullptr);
status = Status::OK();
}
return status;
return Status::OK();
}
Status MergeCF(uint32_t column_family_id, const Slice& key_with_ts,
@@ -813,17 +685,13 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
Slice key =
StripTimestampFromUserKey(key_with_ts, FLAGS_user_timestamp_size);
uint64_t key_id = 0;
TraceKeyDebugInfo key_info;
Status status =
ParseTracedKey(key, "unable to parse key", &key_id, &key_info);
if (status.ok()) {
Status status = ParseTracedKey(key, "unable to parse key", &key_id);
if (!status.ok()) {
return status;
}
const int64_t expected_key_id = static_cast<int64_t>(key_id);
bool should_buffer_write = !(buffered_writes_ == nullptr);
if (should_buffer_write) {
MaybeLogKeyOperation("PutBlobIndexCF", column_family_id,
true /* buffered */, key_info,
"blob_index_size=" + std::to_string(value.size()));
if (buffered_writes_ != nullptr) {
return WriteBatchInternal::PutBlobIndex(buffered_writes_.get(),
column_family_id, key, value);
}
@@ -832,21 +700,11 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
// than the original value bytes. For expected-state replay we only need
// the logical effect of "another put to this key", and db_stress values
// advance deterministically by one value_base per committed write.
const ExpectedValue before =
state_->Get(column_family_id, expected_key_id);
const uint32_t value_base = before.NextValueBase();
const uint32_t value_base =
state_->Get(column_family_id, expected_key_id).NextValueBase();
state_->SyncPut(column_family_id, expected_key_id, value_base);
const ExpectedValue after =
state_->Get(column_family_id, expected_key_id);
NoteWriteOpApplied();
MaybeLogKeyOperation(
"PutBlobIndexCF", column_family_id, false /* buffered */, key_info,
"blob_index_size=" + std::to_string(value.size()) +
" derived_value_base=" + std::to_string(value_base),
&before, &after);
status = Status::OK();
}
return status;
return Status::OK();
}
Status MarkBeginPrepare(bool = false) override {
@@ -891,162 +749,15 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
}
private:
bool HasFocusKey() const { return debug_focus_key_ >= 0; }
bool FocusKeyInRange(uint64_t begin_key_id, uint64_t end_key_id) const {
return HasFocusKey() &&
begin_key_id <= static_cast<uint64_t>(debug_focus_key_) &&
static_cast<uint64_t>(debug_focus_key_) < end_key_id;
}
void MaybeNoteFocusKeyHit(bool hit) {
if (hit) {
++focus_key_op_hits_;
}
}
void MaybeEmitDebugLog(const std::string& line) {
if (!debug_enabled_) {
return;
}
if (emitted_debug_logs_ >= debug_max_logs_) {
++suppressed_debug_logs_;
return;
}
++emitted_debug_logs_;
fprintf(stdout, "[expected_state_trace_debug] %s\n", line.c_str());
fflush(stdout);
}
TraceKeyDebugInfo BuildTraceKeyDebugInfo(const std::string& raw_key,
bool parse_ok,
uint64_t parsed_key_id) {
TraceKeyDebugInfo info;
if (!debug_enabled_) {
return info;
}
info.raw_key = raw_key;
info.raw_key_hex = Slice(raw_key).ToString(/* hex */ true);
info.parse_ok = parse_ok;
info.trailing_bytes = raw_key.size() % sizeof(uint64_t);
info.trailing_xs = CountTrailingXs(raw_key);
info.raw_matches_focus_key =
HasFocusKey() && raw_key == debug_focus_key_raw_;
if (!parse_ok) {
++key_decode_failures_;
return info;
}
info.parsed_key_id = parsed_key_id;
info.roundtrip_key = Key(static_cast<int64_t>(parsed_key_id));
info.roundtrip_key_hex = Slice(info.roundtrip_key).ToString(/* hex */ true);
info.roundtrip_matches_raw = raw_key == info.roundtrip_key;
info.parsed_matches_focus_key =
HasFocusKey() &&
parsed_key_id == static_cast<uint64_t>(debug_focus_key_);
info.roundtrip_matches_focus_key =
HasFocusKey() && info.roundtrip_key == debug_focus_key_raw_;
if (!info.roundtrip_matches_raw) {
++key_roundtrip_mismatches_;
}
return info;
}
Status ParseTracedKey(const Slice& key, const char* error_msg,
uint64_t* key_id, TraceKeyDebugInfo* debug_info) {
uint64_t* key_id) {
const std::string raw_key = key.ToString();
const bool parse_ok = GetIntVal(raw_key, key_id);
if (debug_enabled_) {
*debug_info =
BuildTraceKeyDebugInfo(raw_key, parse_ok, parse_ok ? *key_id : 0);
if (!parse_ok && (!HasFocusKey() || debug_info->MatchesFocusKey())) {
std::ostringstream oss;
oss << "parse_failure error=\"" << error_msg << "\" "
<< DescribeTraceKeyDebugInfo(*debug_info);
MaybeEmitDebugLog(oss.str());
}
}
if (!parse_ok) {
if (!GetIntVal(raw_key, key_id)) {
return Status::Corruption(error_msg, raw_key);
}
return Status::OK();
}
bool ShouldLogKeyOperation(const TraceKeyDebugInfo& info) {
if (!debug_enabled_) {
return false;
}
const bool focus_hit = info.MatchesFocusKey();
MaybeNoteFocusKeyHit(focus_hit);
return !HasFocusKey() || focus_hit;
}
bool ShouldLogRangeOperation(const TraceKeyDebugInfo& begin_info,
const TraceKeyDebugInfo& end_info) {
if (!debug_enabled_) {
return false;
}
const bool focus_hit =
begin_info.MatchesFocusKey() || end_info.MatchesFocusKey() ||
(begin_info.parse_ok && end_info.parse_ok &&
FocusKeyInRange(begin_info.parsed_key_id, end_info.parsed_key_id));
MaybeNoteFocusKeyHit(focus_hit);
return !HasFocusKey() || focus_hit;
}
void MaybeLogKeyOperation(const char* op, uint32_t column_family_id,
bool buffered, const TraceKeyDebugInfo& key_info,
const std::string& details,
const ExpectedValue* before = nullptr,
const ExpectedValue* after = nullptr) {
if (!ShouldLogKeyOperation(key_info)) {
return;
}
std::ostringstream oss;
oss << op << " cf=" << column_family_id << " buffered=" << buffered << " "
<< DescribeTraceKeyDebugInfo(key_info);
if (!details.empty()) {
oss << " " << details;
}
if (before != nullptr) {
oss << " before=" << DescribeExpectedValue(*before);
}
if (after != nullptr) {
oss << " after=" << DescribeExpectedValue(*after);
}
MaybeEmitDebugLog(oss.str());
}
void MaybeLogRangeOperation(const char* op, uint32_t column_family_id,
bool buffered,
const TraceKeyDebugInfo& begin_info,
const TraceKeyDebugInfo& end_info,
const std::string& details,
const ExpectedValue* focus_before = nullptr,
const ExpectedValue* focus_after = nullptr) {
if (!ShouldLogRangeOperation(begin_info, end_info)) {
return;
}
std::ostringstream oss;
oss << op << " cf=" << column_family_id << " buffered=" << buffered
<< " begin=" << DescribeTraceKeyDebugInfo(begin_info)
<< " end=" << DescribeTraceKeyDebugInfo(end_info);
if (!details.empty()) {
oss << " " << details;
}
if (focus_before != nullptr) {
oss << " focus_before=" << DescribeExpectedValue(*focus_before);
}
if (focus_after != nullptr) {
oss << " focus_after=" << DescribeExpectedValue(*focus_after);
}
MaybeEmitDebugLog(oss.str());
}
void NoteWriteOpApplied() {
++num_write_ops_;
assert(num_write_ops_ <= max_write_ops_);
@@ -1055,15 +766,6 @@ class ExpectedStateTraceRecordHandler : public TraceRecord::Handler,
uint64_t num_write_ops_ = 0;
uint64_t max_write_ops_;
ExpectedState* state_;
bool debug_enabled_;
int64_t debug_focus_key_;
std::string debug_focus_key_raw_;
uint64_t debug_max_logs_;
uint64_t key_decode_failures_ = 0;
uint64_t key_roundtrip_mismatches_ = 0;
uint64_t focus_key_op_hits_ = 0;
uint64_t emitted_debug_logs_ = 0;
uint64_t suppressed_debug_logs_ = 0;
std::unordered_map<std::string, std::unique_ptr<WriteBatch>>
xid_to_buffered_writes_;
std::unique_ptr<WriteBatch> buffered_writes_;
@@ -1077,7 +779,6 @@ Status FileExpectedStateManager::Restore(DB* db) {
if (seqno < saved_seqno_) {
return Status::Corruption("DB is older than any restorable expected state");
}
const bool trace_debug = FLAGS_expected_state_trace_debug;
const uint64_t replay_write_ops = seqno - saved_seqno_;
std::string state_filename =
@@ -1093,24 +794,6 @@ Status FileExpectedStateManager::Restore(DB* db) {
std::to_string(saved_seqno_) + kTraceFilenameSuffix;
std::string trace_file_path = GetPathForFilename(trace_filename);
if (trace_debug) {
std::string focus_key_hex = "<unset>";
if (FLAGS_expected_state_trace_debug_key >= 0) {
focus_key_hex = Slice(Key(FLAGS_expected_state_trace_debug_key))
.ToString(/* hex */ true);
}
fprintf(stdout,
"[expected_state_trace_debug] restore_begin saved_seqno=%" PRIu64
" db_seqno=%" PRIu64 " replay_write_ops=%" PRIu64
" state_path=%s trace_path=%s focus_key=%" PRIi64
" focus_key_hex=%s max_logs=%d\n",
static_cast<uint64_t>(saved_seqno_), static_cast<uint64_t>(seqno),
replay_write_ops, state_file_path.c_str(), trace_file_path.c_str(),
FLAGS_expected_state_trace_debug_key, focus_key_hex.c_str(),
FLAGS_expected_state_trace_debug_max_logs);
fflush(stdout);
}
std::unique_ptr<TraceReader> trace_reader;
Status s = NewFileTraceReader(Env::Default(), EnvOptions(), trace_file_path,
&trace_reader);
@@ -1118,13 +801,6 @@ Status FileExpectedStateManager::Restore(DB* db) {
std::string persisted_seqno_file_path = GetPathForFilename(
kPersistedSeqnoBasename + kPersistedSeqnoFilenameSuffix);
uint64_t replayed_write_ops = 0;
uint64_t key_decode_failures = 0;
uint64_t key_roundtrip_mismatches = 0;
uint64_t focus_key_op_hits = 0;
uint64_t logs_emitted = 0;
uint64_t logs_suppressed = 0;
if (s.ok()) {
// We are going to replay on top of "`seqno`.state" to create a new
// "LATEST.state". Start off by creating a tempfile so we can later make the
@@ -1145,8 +821,8 @@ Status FileExpectedStateManager::Restore(DB* db) {
s = state->Open(false /* create */);
}
if (s.ok()) {
handler.reset(new ExpectedStateTraceRecordHandler(seqno - saved_seqno_,
state.get()));
handler.reset(
new ExpectedStateTraceRecordHandler(replay_write_ops, state.get()));
// TODO(ajkr): An API limitation requires we provide `handles` although
// they will be unused since we only use the replayer for reading records.
// Just give a default CFH for now to satisfy the requirement.
@@ -1161,15 +837,9 @@ Status FileExpectedStateManager::Restore(DB* db) {
std::unique_ptr<TraceRecord> record;
s = replayer->Next(&record);
if (!s.ok()) {
if (trace_debug) {
fprintf(stdout,
"[expected_state_trace_debug] restore_replay_next status=%s "
"handler_done=%d\n",
s.ToString().c_str(),
handler != nullptr && handler->IsDone());
fflush(stdout);
}
if (s.IsCorruption() && handler->IsDone()) {
const bool handler_done = handler != nullptr && handler->IsDone();
const bool tolerated_tail_corruption = s.IsCorruption() && handler_done;
if (tolerated_tail_corruption) {
// There could be a corruption reading the tail record of the trace
// due to `db_stress` crashing while writing it. It shouldn't matter
// as long as we already found all the write ops we need to catch up
@@ -1190,29 +860,8 @@ Status FileExpectedStateManager::Restore(DB* db) {
s = Status::Corruption(
"Trace ended before replaying all expected write ops",
std::to_string(handler->NumWriteOps()) + " < " +
std::to_string(seqno - saved_seqno_));
std::to_string(replay_write_ops));
}
if (handler) {
replayed_write_ops = handler->NumWriteOps();
key_decode_failures = handler->NumKeyDecodeFailures();
key_roundtrip_mismatches = handler->NumKeyRoundtripMismatches();
focus_key_op_hits = handler->NumFocusKeyOpHits();
logs_emitted = handler->NumLogsEmitted();
logs_suppressed = handler->NumLogsSuppressed();
}
}
if (trace_debug) {
fprintf(stdout,
"[expected_state_trace_debug] restore_replay_summary status=%s "
"replayed_write_ops=%" PRIu64 "/%" PRIu64
" key_decode_failures=%" PRIu64 " key_roundtrip_mismatches=%" PRIu64
" focus_key_op_hits=%" PRIu64 " logs_emitted=%" PRIu64
" logs_suppressed=%" PRIu64 "\n",
s.ToString().c_str(), replayed_write_ops, replay_write_ops,
key_decode_failures, key_roundtrip_mismatches, focus_key_op_hits,
logs_emitted, logs_suppressed);
fflush(stdout);
}
if (s.ok()) {
@@ -1260,11 +909,6 @@ Status FileExpectedStateManager::Restore(DB* db) {
saved_seqno_ = kMaxSequenceNumber;
}
}
if (trace_debug) {
fprintf(stdout, "[expected_state_trace_debug] restore_end status=%s\n",
s.ToString().c_str());
fflush(stdout);
}
return s;
}
docs/components/stress_test/expected_state_trace.md
+42 -22
View File
@@ -399,35 +399,55 @@ loop still keeps reading trace records until `Next()` returns EOF, footer, or
corruption, at which point restore decides whether the trace prefix it already
consumed was sufficient.
## Debugging support
## Offline trace inspection
Three flags control replay debugging:
`<N>.trace` uses RocksDB's generic binary query-trace format, so there is
already an offline printer for it: `trace_analyzer`.
- `--expected_state_trace_debug`
- `--expected_state_trace_debug_key`
- `--expected_state_trace_debug_max_logs`
Before adding any expected-state-specific debug logging, use this tool to dump
the trace to a readable text file. This is the easiest path for both humans
and agents to inspect replay inputs.
When enabled, restore prints lines prefixed with
`[expected_state_trace_debug]`, including:
Build it with:
- restore begin/end markers
- `Next()` failures such as EOF or corruption
- per-key or per-range replay details
- parse failures and key roundtrip mismatches
- a replay summary with counters
```bash
make -j128 trace_analyzer
```
Useful counters in the summary include:
Create an output directory first, then run:
- `replayed_write_ops`
- `key_decode_failures`
- `key_roundtrip_mismatches`
- `focus_key_op_hits`
- `logs_emitted`
- `logs_suppressed`
```bash
mkdir -p /tmp/trace_dump
./trace_analyzer \
-trace_path=/path/to/<N>.trace \
-output_dir=/tmp/trace_dump \
-output_prefix=<N> \
-convert_to_human_readable_trace \
-try_process_corrupted_trace \
-no_print
```
`--expected_state_trace_debug_key=<k>` narrows logging to a particular logical
key where possible. This is useful when the trace is large and only one key's
history matters.
This writes:
- `/tmp/trace_dump/<N>-human_readable_trace.txt`
The line format is:
- normal record: `<hex_key> type_id cf_id value_size timestamp_us`
- range delete: `<begin_hex> <end_hex> type_id cf_id 0 timestamp_us`
Useful flags:
- `-no_key` omits the hex key columns to reduce output size
- `-try_process_corrupted_trace` is recommended for `db_stress` crash traces,
since they can legitimately have a truncated or corrupt tail record
Two important caveats:
- `trace_analyzer` expects `-output_dir` to already exist
- expected-state replay only needs the write prefix of the trace, but the file
format itself is the generic RocksDB trace format rather than an
expected-state-specific one
## Crash-safety rules encoded in file deletion order
env/io_posix.cc
Vendored
+9 -1
View File
@@ -133,7 +133,8 @@ bool PosixWrite(int fd, const char* buf, size_t nbyte) {
return true;
}
bool PosixPositionedWrite(int fd, const char* buf, size_t nbyte, off_t offset) {
bool PosixPositionedWriteInternal(int fd, const char* buf, size_t nbyte,
off_t offset) {
const size_t kLimit1Gb = 1UL << 30;
const char* src = buf;
@@ -149,6 +150,9 @@ bool PosixPositionedWrite(int fd, const char* buf, size_t nbyte, off_t offset) {
}
return false;
}
if (done == 0) {
return false;
}
left -= done;
offset += done;
src += done;
@@ -202,6 +206,10 @@ bool IsSyncFileRangeSupported(int fd) {
} // anonymous namespace
bool PosixPositionedWrite(int fd, const char* buf, size_t nbyte, off_t offset) {
return PosixPositionedWriteInternal(fd, buf, nbyte, offset);
}
/*
* PosixSequentialFile
*/
env/io_posix.h
Vendored
+5
View File
@@ -27,7 +27,9 @@
#define IORING_SETUP_DEFER_TASKRUN (1U << 13)
#endif
#endif
#if !defined(OS_WIN)
#include <unistd.h>
#endif
#include <atomic>
#include <functional>
@@ -66,6 +68,9 @@ std::string IOErrorMsg(const std::string& context,
// file_name can be left empty if it is not unkown.
IOStatus IOError(const std::string& context, const std::string& file_name,
int err_number);
#if !defined(OS_WIN)
bool PosixPositionedWrite(int fd, const char* buf, size_t nbyte, off_t offset);
#endif
// SyncPoint payload used by deterministic TSAN regression tests to observe
// which virtual address range a freshly created mapping occupies.
tools/db_crashtest.py
+40 -43
View File
@@ -14,6 +14,12 @@ import sys
import tempfile
import time
_TOOLS_DIR = os.path.dirname(__file__)
if _TOOLS_DIR and _TOOLS_DIR not in sys.path:
sys.path.insert(0, _TOOLS_DIR)
import fault_injection_log_parser
per_iteration_random_seed_override = 0
remain_argv = None
is_remote_db = False
@@ -37,6 +43,7 @@ _TSAN_OPTIONS_ENV_VAR = "TSAN_OPTIONS"
_TSAN_SUPPRESSIONS_FILE = os.path.abspath(
os.path.join(os.path.dirname(__file__), "tsan_suppressions.txt")
)
_FAULT_INJECTION_LOG_DIR_NAME = "fault_injection_logs"
def get_random_seed(override):
@@ -2051,52 +2058,41 @@ def cleanup_after_success(db_arg, num_dbs=1):
sys.exit(2)
def print_and_cleanup_fault_injection_log(pid):
def _fault_injection_log_dir():
if is_remote_db:
base_dir = "/tmp"
else:
base_dir = os.environ.get(_TEST_DIR_ENV_VAR) or "/tmp"
return os.path.join(
base_dir,
_FAULT_INJECTION_LOG_DIR_NAME,
)
def print_fault_injection_log(pid):
# Fault injection logs are stored in TEST_TMPDIR (or /tmp) to survive
# DB reopen cleanup, and to be included in sandcastle's db.tar.gz artifact.
# Filter by pid to only print the log from the current run.
max_tail_entries = 32
log_dir = os.environ.get(_TEST_DIR_ENV_VAR) or "/tmp"
pattern = os.path.join(log_dir, "fault_injection_%d_*.log" % pid)
for log in glob.glob(pattern):
print("=== Fault injection log: %s ===" % log)
# DB reopen cleanup, and to be included in crash-test artifacts.
# Filter by pid to only print the log from the current run. Raw and decoded
# logs are intentionally left behind for external artifact collection and
# cleanup.
log_dir = _fault_injection_log_dir()
raw_pattern = os.path.join(log_dir, "fault_injection_%d_*.bin" % pid)
for raw_log in sorted(glob.glob(raw_pattern)):
decoded_log = raw_log + ".txt"
try:
with open(log) as f:
lines = f.readlines()
# Log format: header line(s), entry lines, footer line.
# The footer starts with "=== End of".
# Print header and footer always, truncate entries in the middle.
header = []
footer = []
entries = []
for line in lines:
stripped = line.strip()
if stripped.startswith("=== End of"):
footer.append(line)
elif stripped.startswith("===") or stripped == "(none)":
header.append(line)
else:
entries.append(line)
total_entries = len(entries)
print("".join(header), end="")
if total_entries <= max_tail_entries:
print("".join(entries), end="")
print("".join(footer), end="")
else:
skipped = total_entries - max_tail_entries
print(
"... (%d entries omitted, showing last %d. "
"Full log: %s)\n" % (skipped, max_tail_entries, log),
end="",
entry_count = fault_injection_log_parser.decode_fault_injection_log(
raw_log, decoded_log
)
print("".join(entries[-max_tail_entries:]), end="")
print(
"=== Showed %d of %d injected error entries ===\n"
% (max_tail_entries, total_entries),
end="",
"Fault injection log saved: raw=%s decoded=%s entries=%d"
% (raw_log, decoded_log, entry_count)
)
except (OSError, ValueError) as exc:
print(
"WARNING: failed to decode fault injection log %s: %s\n"
% (raw_log, exc)
)
except OSError:
pass
# This script runs and kills db_stress multiple times. It checks consistency
@@ -2130,7 +2126,7 @@ def blackbox_crash_main(args, unknown_args):
hit_timeout, retcode, outs, errs, pid = execute_cmd(cmd, cmd_params["interval"])
print_and_cleanup_fault_injection_log(pid)
print_fault_injection_log(pid)
outs, errs = strip_expected_sigterm_stderr(outs, errs, hit_timeout)
# Reset destroy_db_initially after each run (it may have been set by
@@ -2159,7 +2155,7 @@ def blackbox_crash_main(args, unknown_args):
cmd, cmd_params["verify_timeout"], True
)
print_and_cleanup_fault_injection_log(pid)
print_fault_injection_log(pid)
# For the final run
print_run_output_and_exit_on_error(args, finalized_params, outs, errs)
@@ -2307,6 +2303,7 @@ def whitebox_crash_main(args, unknown_args):
hit_timeout, retncode, stdoutdata, stderrdata, pid = execute_cmd(
cmd, exit_time - time.time() + 900
)
print_fault_injection_log(pid)
# Reset destroy_db_initially after each run (it may have been set by
# command line for first run, or set for various reasons for a step)
tools/db_crashtest_test.py
+107
View File
@@ -6,14 +6,20 @@
# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
import importlib.util
import io
import os
import shutil
import struct
import sys
import tempfile
import unittest
from contextlib import redirect_stdout
_DB_CRASHTEST_PATH = os.path.join(os.path.dirname(__file__), "db_crashtest.py")
_FAULT_INJECTION_LOG_PARSER_PATH = os.path.join(
os.path.dirname(__file__), "fault_injection_log_parser.py"
)
_TEST_DIR_ENV_VAR = "TEST_TMPDIR"
_TEST_EXPECTED_DIR_ENV_VAR = "TEST_TMPDIR_EXPECTED"
_TSAN_OPTIONS_ENV_VAR = "TSAN_OPTIONS"
@@ -33,6 +39,15 @@ def load_db_crashtest_module():
return module
def load_fault_injection_log_parser_module():
spec = importlib.util.spec_from_file_location(
"fault_injection_log_parser_under_test", _FAULT_INJECTION_LOG_PARSER_PATH
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class DBCrashTestTest(unittest.TestCase):
def setUp(self):
self.test_tmpdir = tempfile.mkdtemp(prefix="db_crashtest_test_")
@@ -66,6 +81,9 @@ class DBCrashTestTest(unittest.TestCase):
def load_db_crashtest(self):
return load_db_crashtest_module()
def load_fault_injection_log_parser(self):
return load_fault_injection_log_parser_module()
def build_params(self, base_params, overrides=None):
params = dict(base_params)
params["db"] = self.test_tmpdir
@@ -352,6 +370,95 @@ class DBCrashTestTest(unittest.TestCase):
diagnostics,
)
# Goal: verify db_crashtest decodes the headerless streaming binary fault
# injection log format used on crash paths while keeping stdout concise.
# The test writes a raw .bin file with two complete entries plus a
# truncated tail, then checks the decoded text artifact and the summary
# line printed to stdout.
def test_print_fault_injection_log_decodes_streaming_raw_trace(self):
db_crashtest = self.load_db_crashtest()
fault_parser = self.load_fault_injection_log_parser()
pid = 5151
log_dir = os.path.join(self.test_tmpdir, "fault_injection_logs")
os.makedirs(log_dir)
raw_log = os.path.join(log_dir, f"fault_injection_{pid}_1.bin")
decoded_log = raw_log + ".txt"
entry0 = fault_parser.ENTRY_STRUCT.pack(
123456789,
17,
7,
4,
0,
0,
b"abcd".ljust(48, b"\0"),
fault_parser.DETAIL_KIND_OFFSET_SIZE_AND_HEAD,
4,
1,
0,
b"Append\0".ljust(32, b"\0"),
b"/tmp/000001.log\0".ljust(72, b"\0"),
b"injected write error\0".ljust(56, b"\0"),
)
entry1 = fault_parser.ENTRY_STRUCT.pack(
123456790,
23,
0,
6,
0,
0,
b"/tmp/b".ljust(48, b"\0"),
fault_parser.DETAIL_KIND_TWO_FILES,
6,
0,
1,
b"Rename\0".ljust(32, b"\0"),
b"/tmp/a\0".ljust(72, b"\0"),
b"injected metadata read error\0".ljust(56, b"\0"),
)
with open(raw_log, "wb") as f:
f.write(entry0)
f.write(entry1)
f.write(b"tail")
stdout = io.StringIO()
with redirect_stdout(stdout):
db_crashtest.print_fault_injection_log(pid)
self.assertTrue(os.path.exists(decoded_log))
with open(decoded_log) as f:
decoded_text = f.read()
self.assertIn(
'Append("/tmp/000001.log", offset=7, size=4, head=[61 62 63 64])',
decoded_text,
)
self.assertIn("IO error: injected write error [retryable]", decoded_text)
self.assertIn(
'Rename("/tmp/a", "/tmp/b") -> IO error: injected metadata read error [data_loss]',
decoded_text,
)
self.assertIn("max=unbounded", decoded_text)
self.assertIn(
"Fault injection log saved: raw=%s decoded=%s entries=2"
% (raw_log, decoded_log),
stdout.getvalue(),
)
# Goal: verify fault-injection logs do not follow a remote TEST_TMPDIR.
# The test marks the DB as remote, points TEST_TMPDIR at a remote-looking
# path, and checks db_crashtest uses the same local staging root as
# db_stress.
def test_fault_injection_log_dir_uses_local_tmp_for_remote_db(self):
db_crashtest = self.load_db_crashtest()
os.environ[_TEST_DIR_ENV_VAR] = "/dev_test/rocksdb_crash_test/job123"
db_crashtest.is_remote_db = True
self.assertEqual(
os.path.join("/tmp", "fault_injection_logs"),
db_crashtest._fault_injection_log_dir(),
)
if __name__ == "__main__":
unittest.main()
tools/fault_injection_log_parser.py
+256
View File
@@ -0,0 +1,256 @@
#!/usr/bin/env python3
# Copyright (c) Meta Platforms, Inc. and affiliates.
# This source code is licensed under both the GPLv2 (found in the COPYING file
# in the root directory) and the Apache 2.0 License (found in the
# LICENSE.Apache file in the root directory).
import argparse
import struct
TRACE_FILE_MAGIC = b"FINJLOG1"
LEGACY_TRACE_FILE_VERSION = 1
RING_TRACE_FILE_VERSION = 2
TRACE_FILE_VERSION = 3
DETAIL_KIND_NONE = 0
DETAIL_KIND_TWO_FILES = 1
DETAIL_KIND_SIZE_AND_HEAD = 2
DETAIL_KIND_OFFSET_SIZE_AND_HEAD = 3
DETAIL_KIND_OFFSET_AND_SIZE = 4
DETAIL_KIND_SIZE = 5
DETAIL_KIND_COUNT = 6
DETAIL_KIND_REQ_OFFSET_AND_SIZE = 7
HEADER_STRUCT = struct.Struct("<8sQIIIIII")
LEGACY_ENTRY_STRUCT = struct.Struct("<QQ256s")
ENTRY_STRUCT = struct.Struct("<QQQQII48sBBBB32s72s56s4x")
def _read_exact(infile, size):
data = infile.read(size)
if len(data) != size:
raise ValueError("truncated injected error log")
return data
def _decode_c_string(raw):
return raw.split(b"\0", 1)[0].decode("utf-8", "replace")
def _format_hex_head(payload, total_size):
head = " ".join(f"{byte:02x}" for byte in payload)
if total_size > len(payload):
return f"{head} ..." if head else "..."
return head
def _format_detail(detail_kind, offset, size, count, req_idx, payload):
if detail_kind == DETAIL_KIND_NONE:
return ""
if detail_kind == DETAIL_KIND_TWO_FILES:
suffix = "..." if size > len(payload) else ""
return f"\"{payload.decode('utf-8', 'replace')}{suffix}\""
if detail_kind == DETAIL_KIND_SIZE_AND_HEAD:
return f"size={size}, head=[{_format_hex_head(payload, size)}]"
if detail_kind == DETAIL_KIND_OFFSET_SIZE_AND_HEAD:
return (
f"offset={offset}, size={size}, "
f"head=[{_format_hex_head(payload, size)}]"
)
if detail_kind == DETAIL_KIND_OFFSET_AND_SIZE:
return f"offset={offset}, size={size}"
if detail_kind == DETAIL_KIND_SIZE:
return f"size={size}"
if detail_kind == DETAIL_KIND_COUNT:
return f"num_reqs={count}"
if detail_kind == DETAIL_KIND_REQ_OFFSET_AND_SIZE:
return f"req[{req_idx}], offset={offset}, size={size}"
return f"detail_kind={detail_kind}"
def _decode_legacy_entries(infile, outfile, dumped_entries):
printed = 0
for _ in range(dumped_entries):
timestamp_us, thread_id, context = LEGACY_ENTRY_STRUCT.unpack(
_read_exact(infile, LEGACY_ENTRY_STRUCT.size)
)
if timestamp_us == 0:
continue
secs = timestamp_us // 1000000
usecs = timestamp_us % 1000000
context_str = _decode_c_string(context)
outfile.write(f"[{secs}.{usecs:06d}] thread={thread_id}: {context_str}\n")
printed += 1
return printed
def _decode_entry_bytes(entry_bytes, outfile):
(
timestamp_us,
thread_id,
offset,
size,
count,
req_idx,
detail_payload,
detail_kind,
payload_size,
retryable,
data_loss,
op_name,
file_name,
status_message,
) = ENTRY_STRUCT.unpack(entry_bytes)
if timestamp_us == 0:
return 0
if payload_size > len(detail_payload):
raise ValueError(f"invalid payload size in trace entry: {payload_size}")
secs = timestamp_us // 1000000
usecs = timestamp_us % 1000000
op_name = _decode_c_string(op_name)
file_name = _decode_c_string(file_name)
status_message = _decode_c_string(status_message)
payload = detail_payload[:payload_size]
detail = _format_detail(detail_kind, offset, size, count, req_idx, payload)
line = f"[{secs}.{usecs:06d}] thread={thread_id}: {op_name}(\"{file_name}\""
if detail:
line += f", {detail}"
line += f") -> IO error: {status_message}"
flags = []
if retryable:
flags.append("retryable")
if data_loss:
flags.append("data_loss")
if flags:
line += " [" + ",".join(flags) + "]"
outfile.write(line + "\n")
return 1
def _decode_ring_entries(infile, outfile, dumped_entries):
printed = 0
for _ in range(dumped_entries):
printed += _decode_entry_bytes(_read_exact(infile, ENTRY_STRUCT.size), outfile)
return printed
def _decode_stream_entries(infile, outfile):
data = infile.read()
full_entries = len(data) // ENTRY_STRUCT.size
printed = 0
for index in range(full_entries):
start = index * ENTRY_STRUCT.size
end = start + ENTRY_STRUCT.size
try:
printed += _decode_entry_bytes(data[start:end], outfile)
except ValueError:
if index + 1 == full_entries:
break
raise
return printed, full_entries
def decode_fault_injection_log(raw_path, output_path=None):
if output_path is None:
output_path = raw_path + ".txt"
with open(raw_path, "rb") as infile, open(output_path, "w") as outfile:
outfile.write(
"=== Recently Injected Fault Injection Errors (most recent last) ===\n"
)
header = infile.read(HEADER_STRUCT.size)
footer_total_entries = 0
footer_max_entries = 0
printed = 0
if len(header) >= len(TRACE_FILE_MAGIC) and header.startswith(TRACE_FILE_MAGIC):
if len(header) != HEADER_STRUCT.size:
raise ValueError("truncated injected error log header")
(
magic,
total_entries,
version,
header_size,
entry_size,
max_entries,
dumped_entries,
reserved,
) = HEADER_STRUCT.unpack(header)
if magic != TRACE_FILE_MAGIC:
raise ValueError(f"unexpected trace magic: {magic!r}")
if version not in (
LEGACY_TRACE_FILE_VERSION,
RING_TRACE_FILE_VERSION,
TRACE_FILE_VERSION,
):
raise ValueError(f"unsupported trace version: {version}")
if header_size != HEADER_STRUCT.size:
raise ValueError(
f"unexpected trace header size: {header_size} != {HEADER_STRUCT.size}"
)
footer_total_entries = total_entries
footer_max_entries = max_entries
if version == LEGACY_TRACE_FILE_VERSION:
if entry_size != LEGACY_ENTRY_STRUCT.size:
raise ValueError(
"unexpected legacy trace entry size: "
f"{entry_size} != {LEGACY_ENTRY_STRUCT.size}"
)
if reserved != 256:
raise ValueError(f"unexpected legacy max message len: {reserved}")
printed = _decode_legacy_entries(infile, outfile, dumped_entries)
elif version == RING_TRACE_FILE_VERSION:
if entry_size != ENTRY_STRUCT.size:
raise ValueError(
f"unexpected trace entry size: {entry_size} != {ENTRY_STRUCT.size}"
)
printed = _decode_ring_entries(infile, outfile, dumped_entries)
else:
if entry_size != ENTRY_STRUCT.size:
raise ValueError(
f"unexpected trace entry size: {entry_size} != {ENTRY_STRUCT.size}"
)
printed, full_entries = _decode_stream_entries(infile, outfile)
if footer_total_entries == 0:
footer_total_entries = full_entries
else:
infile.seek(0)
printed, footer_total_entries = _decode_stream_entries(infile, outfile)
if printed == 0:
outfile.write("(none)\n")
if footer_max_entries == 0:
outfile.write(
"=== End of injected error log (%d entries, total=%d, max=unbounded) ===\n"
% (printed, footer_total_entries)
)
else:
outfile.write(
"=== End of injected error log (%d entries, total=%d, max=%d) ===\n"
% (printed, footer_total_entries, footer_max_entries)
)
return printed
def _main():
parser = argparse.ArgumentParser(
description="Decode raw fault injection logs emitted by db_stress."
)
parser.add_argument("raw_log", help="Path to the raw .bin fault injection log")
parser.add_argument(
"--output",
help="Path for decoded text output. Defaults to <raw_log>.txt",
)
args = parser.parse_args()
output_path = args.output if args.output is not None else args.raw_log + ".txt"
decode_fault_injection_log(args.raw_log, output_path)
print(output_path)
if __name__ == "__main__":
_main()
utilities/fault_injection_fs.cc
+124 -39
View File
@@ -17,11 +17,13 @@
#include "utilities/fault_injection_fs.h"
#include <algorithm>
#include <chrono>
#include <cstdio>
#include <functional>
#include <utility>
#include "env/composite_env_wrapper.h"
#include "env/io_posix.h"
#include "monitoring/thread_status_util.h"
#include "port/lang.h"
#include "port/stack_trace.h"
@@ -42,6 +44,55 @@ const std::string kNewFileNoOverwrite;
namespace {
#ifndef OS_WIN
uint64_t NowMicros() {
auto now = std::chrono::system_clock::now();
return std::chrono::duration_cast<std::chrono::microseconds>(
now.time_since_epoch())
.count();
}
// Preserve the suffix so long artifact paths still retain the basename.
void CopyStringSuffix(const Slice& src, char* dst, size_t dst_len) {
if (dst_len == 0) {
return;
}
const size_t copied = std::min(src.size(), dst_len - 1);
if (copied > 0) {
std::memcpy(dst, src.data() + src.size() - copied, copied);
}
dst[copied] = '\0';
}
uint8_t DetailPayloadSize(const InjectedErrorLog::TaggedEntryDetail& detail) {
switch (detail.kind) {
case InjectedErrorLog::kDetailTwoFiles:
case InjectedErrorLog::kDetailSizeAndHead:
case InjectedErrorLog::kDetailOffsetSizeAndHead:
return static_cast<uint8_t>(std::min<uint64_t>(
detail.entry.size, InjectedErrorLog::kMaxDetailPayloadLen));
default:
return 0;
}
}
#endif
int OpenInjectedErrorLogFile(const std::string& path) {
#ifndef OS_WIN
if (path.empty()) {
return -1;
}
int flags = O_WRONLY | O_CREAT | O_TRUNC;
#ifdef O_CLOEXEC
flags |= O_CLOEXEC;
#endif
return open(path.c_str(), flags, 0644);
#else
(void)path;
return -1;
#endif
}
bool TryParseInfoLogFileName(const std::string& file_name, uint64_t* number,
FileType* type) {
size_t prefix_len = std::string::npos;
@@ -69,6 +120,62 @@ bool TryParseInfoLogFileName(const std::string& file_name, uint64_t* number,
} // namespace
InjectedErrorLog::InjectedErrorLog(const std::string& path)
: next_write_offset_(0), log_fd_(OpenInjectedErrorLogFile(path)) {}
InjectedErrorLog::~InjectedErrorLog() {
#ifndef OS_WIN
if (log_fd_ >= 0) {
Flush();
close(log_fd_);
}
#endif
}
void InjectedErrorLog::Record(const Slice& op_name, const Slice& file_name,
const TaggedEntryDetail& detail,
const IOStatus& status) {
#ifndef OS_WIN
if (log_fd_ < 0) {
return;
}
Entry entry{};
entry.timestamp_us = NowMicros();
entry.thread_id = Env::Default()->GetThreadID();
entry.detail = detail.entry;
entry.detail_kind = detail.kind;
entry.detail_payload_size = DetailPayloadSize(detail);
entry.retryable = status.GetRetryable() ? 1 : 0;
entry.data_loss = status.GetDataLoss() ? 1 : 0;
CopyStringSuffix(op_name, entry.op_name, sizeof(entry.op_name));
CopyStringSuffix(file_name, entry.file_name, sizeof(entry.file_name));
const char* status_message = status.getState();
CopyStringSuffix(status_message == nullptr ? Slice() : Slice(status_message),
entry.status_message, sizeof(entry.status_message));
const uint64_t offset =
next_write_offset_.fetch_add(sizeof(entry), std::memory_order_relaxed);
PosixPositionedWrite(log_fd_, reinterpret_cast<const char*>(&entry),
sizeof(entry), static_cast<off_t>(offset));
#else
(void)op_name;
(void)file_name;
(void)detail;
(void)status;
#endif
}
void InjectedErrorLog::Flush() const {
#ifndef OS_WIN
if (log_fd_ < 0) {
return;
}
while (fsync(log_fd_) != 0 && errno == EINTR) {
}
#endif
}
// Assume a filename, and not a directory name like "/foo/bar/"
std::string TestFSGetDirName(const std::string filename) {
size_t found = filename.find_last_of("/\\");
@@ -1619,9 +1726,10 @@ void FaultInjectionTestFS::UntrackFile(const std::string& f) {
IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalReadError(
const IOOptions& io_options, const char* op_name,
const std::string& file_name, std::function<std::string()> detail_fn,
ErrorOperation op, Slice* result, bool direct_io, char* scratch,
bool need_count_increase, bool* fault_injected) {
const std::string& file_name,
const InjectedErrorLog::TaggedEntryDetail& detail, ErrorOperation op,
Slice* result, bool direct_io, char* scratch, bool need_count_increase,
bool* fault_injected) {
bool dummy_bool;
bool& ret_fault_injected = fault_injected ? *fault_injected : dummy_bool;
ret_fault_injected = false;
@@ -1636,7 +1744,7 @@ IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalReadError(
IOStatus ret;
if (ctx->rand.OneIn(ctx->one_in)) {
if (ctx->count == 0) {
ctx->message = "";
ctx->message.clear();
}
if (need_count_increase) {
ctx->count++;
@@ -1646,12 +1754,9 @@ IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalReadError(
}
ctx->callstack = port::SaveStack(&ctx->frames);
std::stringstream msg;
msg << FaultInjectionTestFS::kInjected << " ";
if (op != ErrorOperation::kMultiReadSingleReq) {
// Likely non-per read status code for MultiRead
msg << "read error";
ctx->message = msg.str();
ctx->message = kInjectedReadError;
ret_fault_injected = true;
ret = IOStatus::IOError(ctx->message);
} else if (Random::GetTLSInstance()->OneIn(8)) {
@@ -1659,8 +1764,7 @@ IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalReadError(
// For a small chance, set the failure to status but turn the
// result to be empty, which is supposed to be caught for a check.
*result = Slice();
msg << "empty result";
ctx->message = msg.str();
ctx->message = kInjectedEmptyResult;
ret_fault_injected = true;
ret = IOStatus::IOError(ctx->message);
} else if (!direct_io && Random::GetTLSInstance()->OneIn(7) &&
@@ -1678,13 +1782,11 @@ IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalReadError(
// It would work for CRC. Not 100% sure for xxhash and will adjust
// if it is not the case.
const_cast<char*>(result->data())[result->size() - 1]++;
msg << "corrupt last byte";
ctx->message = msg.str();
ctx->message = kInjectedCorruptLastByte;
ret_fault_injected = true;
ret = IOStatus::IOError(ctx->message);
} else {
msg << "error result multiget single";
ctx->message = msg.str();
ctx->message = kInjectedErrorResultMultiGetSingle;
ret_fault_injected = true;
ret = IOStatus::IOError(ctx->message);
}
@@ -1693,15 +1795,7 @@ IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalReadError(
ret.SetRetryable(ctx->retryable);
ret.SetDataLoss(ctx->has_data_loss);
if (!ret.ok()) {
std::string detail = detail_fn ? detail_fn() : "";
if (detail.empty()) {
injected_error_log_.Record("%s(\"%.128s\") -> %s", op_name,
file_name.c_str(), ret.ToString().c_str());
} else {
injected_error_log_.Record("%s(\"%.128s\", %s) -> %s", op_name,
file_name.c_str(), detail.c_str(),
ret.ToString().c_str());
}
injected_error_log_.Record(op_name, file_name, detail, ret);
}
return ret;
}
@@ -1719,13 +1813,14 @@ bool FaultInjectionTestFS::TryParseFileName(const std::string& file_name,
IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalError(
FaultInjectionIOType type, const IOOptions& io_options, const char* op_name,
const std::string& file_name, std::function<std::string()> detail_fn,
ErrorOperation op, Slice* result, bool direct_io, char* scratch,
bool need_count_increase, bool* fault_injected) {
const std::string& file_name,
const InjectedErrorLog::TaggedEntryDetail& detail, ErrorOperation op,
Slice* result, bool direct_io, char* scratch, bool need_count_increase,
bool* fault_injected) {
if (type == FaultInjectionIOType::kRead) {
return MaybeInjectThreadLocalReadError(
io_options, op_name, file_name, std::move(detail_fn), op, result,
direct_io, scratch, need_count_increase, fault_injected);
io_options, op_name, file_name, detail, op, result, direct_io, scratch,
need_count_increase, fault_injected);
}
ErrorContext* ctx = GetErrorContextFromFaultInjectionIOType(type);
@@ -1746,17 +1841,7 @@ IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalError(
ret = IOStatus::IOError(ctx->message);
ret.SetRetryable(ctx->retryable);
ret.SetDataLoss(ctx->has_data_loss);
{
std::string detail = detail_fn ? detail_fn() : "";
if (detail.empty()) {
injected_error_log_.Record("%s(\"%.128s\") -> %s", op_name,
file_name.c_str(), ret.ToString().c_str());
} else {
injected_error_log_.Record("%s(\"%.128s\", %s) -> %s", op_name,
file_name.c_str(), detail.c_str(),
ret.ToString().c_str());
}
}
injected_error_log_.Record(op_name, file_name, detail, ret);
if (type == FaultInjectionIOType::kWrite) {
TEST_SYNC_POINT(
"FaultInjectionTestFS::InjectMetadataWriteError:Injected");
utilities/fault_injection_fs.h
+170 -237
View File
@@ -18,25 +18,20 @@
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdarg>
#include <cstring>
#include <functional>
#include <limits>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <thread>
#ifndef OS_WIN
#include <fcntl.h>
#include <limits.h>
#include <unistd.h>
#endif
// PATH_MAX may not be defined on all platforms
#ifndef PATH_MAX
#define PATH_MAX 4096
#endif
#include "file/filename.h"
#include "port/lang.h"
#include "rocksdb/file_system.h"
@@ -46,64 +41,65 @@
namespace ROCKSDB_NAMESPACE {
// A fixed-size circular buffer that records recently injected errors.
// Thread-safe for concurrent writes. Designed to be safe to read from a
// signal handler (PrintAll uses only fprintf to stderr).
// A binary log that records injected errors directly to a file.
// Thread-safe for concurrent writes. The log file can be flushed on exit or
// from a signal handler so records survive clean exits and crash paths.
class InjectedErrorLog {
public:
static constexpr size_t kMaxEntries = 1000;
static constexpr size_t kMaxMessageLen = 256;
static constexpr size_t kMaxOpNameLen = 32;
static constexpr size_t kMaxFileNameLen = 72;
static constexpr size_t kMaxStatusMessageLen = 56;
static constexpr size_t kMaxDetailPayloadLen = 48;
static constexpr uint8_t kDetailNone = 0;
static constexpr uint8_t kDetailTwoFiles = 1;
static constexpr uint8_t kDetailSizeAndHead = 2;
static constexpr uint8_t kDetailOffsetSizeAndHead = 3;
static constexpr uint8_t kDetailOffsetAndSize = 4;
static constexpr uint8_t kDetailSize = 5;
static constexpr uint8_t kDetailCount = 6;
static constexpr uint8_t kDetailReqOffsetAndSize = 7;
struct EntryDetail {
uint64_t offset = 0;
uint64_t size = 0;
uint32_t count = 0;
uint32_t req_idx = 0;
char payload[kMaxDetailPayloadLen] = {};
};
static_assert(sizeof(EntryDetail) == 72,
"Injected error log detail size must stay stable");
struct TaggedEntryDetail {
uint8_t kind = kDetailNone;
EntryDetail entry;
};
struct Entry {
uint64_t timestamp_us;
uint64_t thread_id;
char context[kMaxMessageLen];
EntryDetail detail;
uint8_t detail_kind;
uint8_t detail_payload_size;
uint8_t retryable;
uint8_t data_loss;
char op_name[kMaxOpNameLen];
char file_name[kMaxFileNameLen];
char status_message[kMaxStatusMessageLen];
};
InjectedErrorLog() : head_(0), entries_{} { log_path_[0] = '\0'; }
static_assert(sizeof(Entry) == 256,
"Injected error log entry size must stay stable");
static_assert(port::kLittleEndian,
"Injected error log requires little-endian platforms");
// Set the file path for PrintAll() output. Must be called before any
// signal handler invocation (not async-signal-safe itself due to string
// copy, but called once at setup time). If not set, PrintAll() falls
// back to writing to stderr.
void SetLogFilePath(const std::string& path) {
size_t len = std::min(path.size(), sizeof(log_path_) - 1);
memcpy(log_path_, path.data(), len);
log_path_[len] = '\0';
}
explicit InjectedErrorLog(const std::string& path);
~InjectedErrorLog();
InjectedErrorLog(const InjectedErrorLog&) = delete;
InjectedErrorLog& operator=(const InjectedErrorLog&) = delete;
TSAN_SUPPRESSION void Record(const char* fmt, ...)
#if defined(__GNUC__) || defined(__clang__)
__attribute__((format(printf, 2, 3)))
#endif
{
size_t idx = head_.fetch_add(1, std::memory_order_relaxed) % kMaxEntries;
Entry& e = entries_[idx];
e.thread_id = std::hash<std::thread::id>{}(std::this_thread::get_id());
auto now = std::chrono::system_clock::now();
e.timestamp_us = std::chrono::duration_cast<std::chrono::microseconds>(
now.time_since_epoch())
.count();
// Format into a local buffer first, then copy into the shared entry.
// This avoids calling the TSAN-intercepted vsnprintf directly on shared
// memory. We use a byte-by-byte loop instead of memcpy because
// TSAN_SUPPRESSION (no_sanitize("thread")) only suppresses
// compiler-inserted instrumentation -- it does NOT suppress TSAN's
// runtime interceptors for libc functions like memcpy, vsnprintf, and
// snprintf. Plain store instructions are always suppressed regardless
// of optimization level. The volatile source pointer prevents the
// compiler from recognizing this as a memcpy idiom and replacing it
// with a memcpy call.
char local_buf[kMaxMessageLen];
va_list args;
va_start(args, fmt);
vsnprintf(local_buf, kMaxMessageLen, fmt, args);
va_end(args);
const volatile char* src = local_buf;
for (size_t i = 0; i < kMaxMessageLen; i++) {
e.context[i] = src[i];
}
}
void Record(const Slice& op_name, const Slice& file_name,
const TaggedEntryDetail& detail, const IOStatus& status);
// Format the first few bytes of a buffer as hex for logging.
// Returns a string like "ab cd ef 01 02 ..."
@@ -111,171 +107,109 @@ class InjectedErrorLog {
size_t max_bytes = 8) {
std::string result;
size_t n = std::min(size, max_bytes);
char buf[4];
static const char kHexDigits[] = "0123456789abcdef";
result.reserve(n * 3 + ((size > max_bytes) ? 4 : 0));
for (size_t i = 0; i < n; i++) {
snprintf(buf, sizeof(buf), "%02x ", (unsigned char)data[i]);
result += buf;
if (i > 0) {
result.push_back(' ');
}
uint8_t byte = static_cast<uint8_t>(data[i]);
result.push_back(kHexDigits[byte >> 4]);
result.push_back(kHexDigits[byte & 0x0f]);
}
if (size > max_bytes) {
result.append(" ...");
}
if (size > max_bytes) result += "...";
if (!result.empty() && result.back() == ' ') result.pop_back();
return result;
}
// Print all recorded entries to a log file (or stderr as fallback).
// Async-signal-safe: uses only open/write/close/snprintf (no fprintf,
// no malloc). Safe to call from a signal handler.
//
// Note: entries may be read while being written by another thread.
// This is a benign race -- at worst, one entry may appear garbled.
// We accept this trade-off to keep PrintAll() free of locks and safe
// for use in signal handlers.
TSAN_SUPPRESSION void PrintAll() const {
#ifndef OS_WIN
int fd = -1;
if (log_path_[0] != '\0') {
fd = open(log_path_, O_WRONLY | O_CREAT | O_TRUNC, 0644);
}
// Fall back to stdout if open failed or no path was set.
// We avoid stderr because db_crashtest.py treats any stderr output
// as a test failure.
if (fd < 0) {
fd = STDOUT_FILENO;
}
auto write_str = [fd](const char* buf, int len) {
if (len > 0) {
// Ignore return value in signal handler -- nothing we can do
auto unused __attribute__((unused)) = write(fd, buf, len);
}
};
char buf[512];
int len = snprintf(buf, sizeof(buf),
"\n=== Recently Injected Fault Injection Errors "
"(most recent last) ===\n");
write_str(buf, len);
size_t total = head_.load(std::memory_order_relaxed);
if (total == 0) {
len = snprintf(buf, sizeof(buf), "(none)\n");
write_str(buf, len);
if (fd != STDOUT_FILENO) close(fd);
return;
}
size_t count = std::min(total, kMaxEntries);
size_t start = (total >= kMaxEntries) ? (total % kMaxEntries) : 0;
for (size_t i = 0; i < count; i++) {
size_t idx = (start + i) % kMaxEntries;
// Copy entry fields to locals to avoid passing shared memory through
// TSAN-intercepted snprintf. See comment in Record() for why we use a
// volatile pointer to prevent loop-to-memcpy optimization.
const Entry& e = entries_[idx];
uint64_t local_ts = e.timestamp_us;
uint64_t local_tid = e.thread_id;
char local_ctx[kMaxMessageLen];
const volatile char* ctx_src = e.context;
for (size_t j = 0; j < kMaxMessageLen; j++) {
local_ctx[j] = ctx_src[j];
}
if (local_ts == 0) continue;
uint64_t secs = local_ts / 1000000;
uint64_t usecs = local_ts % 1000000;
len = snprintf(buf, sizeof(buf), "[%llu.%06llu] thread=%llu: %s\n",
(unsigned long long)secs, (unsigned long long)usecs,
(unsigned long long)local_tid, local_ctx);
write_str(buf, len);
}
len = snprintf(buf, sizeof(buf),
"=== End of injected error log (%zu entries) ===\n", count);
write_str(buf, len);
if (fd != STDOUT_FILENO) close(fd);
#else
// On Windows, crash callbacks via signal handlers are not used,
// so PrintAll() is a no-op.
#endif
}
// Flush the already-appended log file so crash/termination paths preserve
// the most recent records.
void Flush() const;
private:
std::atomic<size_t> head_;
Entry entries_[kMaxEntries];
char log_path_[PATH_MAX];
std::atomic<uint64_t> next_write_offset_;
const int log_fd_;
};
class TestFSWritableFile;
class FaultInjectionTestFS;
// Deferred detail builders for injected error logging.
// These return lambdas that are only evaluated when a fault is actually
// injected, avoiding string formatting overhead on the common (no-fault) path.
// Captured references are safe because the lambda is called synchronously
// within MaybeInjectThreadLocalError before the caller returns.
// Fixed-size detail builders for injected error logging.
// These avoid hot-path string formatting while keeping the detail payload owned
// and reusable by the on-disk entry layout.
namespace fault_injection_detail {
inline std::function<std::string()> NoDetail() { return {}; }
using TaggedEntryDetail = InjectedErrorLog::TaggedEntryDetail;
inline std::function<std::string()> TwoFiles(const std::string& /*f1*/,
inline void CopyPayloadBytes(const Slice& src, char* dst, size_t dst_len) {
const size_t copied = std::min(src.size(), dst_len);
if (copied > 0) {
std::memcpy(dst, src.data(), copied);
}
}
inline TaggedEntryDetail NoDetail() { return TaggedEntryDetail(); }
inline TaggedEntryDetail TwoFiles(const std::string& /*f1*/,
const std::string& f2) {
return [&f2]() -> std::string {
char buf[160];
snprintf(buf, sizeof(buf), "\"%.128s\"", f2.c_str());
return std::string(buf);
};
TaggedEntryDetail detail;
detail.kind = InjectedErrorLog::kDetailTwoFiles;
detail.entry.size = static_cast<uint64_t>(f2.size());
CopyPayloadBytes(Slice(f2), detail.entry.payload,
sizeof(detail.entry.payload));
return detail;
}
inline std::function<std::string()> SizeAndHead(const Slice& data) {
return [data]() -> std::string {
char buf[128];
snprintf(buf, sizeof(buf), "size=%zu, head=[%s]", data.size(),
InjectedErrorLog::HexHead(data.data(), data.size()).c_str());
return std::string(buf);
};
inline TaggedEntryDetail SizeAndHead(const Slice& data) {
TaggedEntryDetail detail;
detail.kind = InjectedErrorLog::kDetailSizeAndHead;
detail.entry.size = static_cast<uint64_t>(data.size());
CopyPayloadBytes(data, detail.entry.payload, sizeof(detail.entry.payload));
return detail;
}
inline std::function<std::string()> OffsetSizeAndHead(uint64_t offset,
const Slice& data) {
return [offset, data]() -> std::string {
char buf[160];
snprintf(buf, sizeof(buf), "offset=%llu, size=%zu, head=[%s]",
(unsigned long long)offset, data.size(),
InjectedErrorLog::HexHead(data.data(), data.size()).c_str());
return std::string(buf);
};
inline TaggedEntryDetail OffsetSizeAndHead(uint64_t offset, const Slice& data) {
TaggedEntryDetail detail;
detail.kind = InjectedErrorLog::kDetailOffsetSizeAndHead;
detail.entry.offset = offset;
detail.entry.size = static_cast<uint64_t>(data.size());
CopyPayloadBytes(data, detail.entry.payload, sizeof(detail.entry.payload));
return detail;
}
inline std::function<std::string()> OffsetAndSize(uint64_t offset, size_t n) {
return [offset, n]() -> std::string {
char buf[64];
snprintf(buf, sizeof(buf), "offset=%llu, size=%zu",
(unsigned long long)offset, n);
return std::string(buf);
};
inline TaggedEntryDetail OffsetAndSize(uint64_t offset, size_t n) {
TaggedEntryDetail detail;
detail.kind = InjectedErrorLog::kDetailOffsetAndSize;
detail.entry.offset = offset;
detail.entry.size = static_cast<uint64_t>(n);
return detail;
}
inline std::function<std::string()> Size(uint64_t size) {
return [size]() -> std::string {
char buf[32];
snprintf(buf, sizeof(buf), "size=%llu", (unsigned long long)size);
return std::string(buf);
};
inline TaggedEntryDetail Size(uint64_t size) {
TaggedEntryDetail detail;
detail.kind = InjectedErrorLog::kDetailSize;
detail.entry.size = size;
return detail;
}
inline std::function<std::string()> Count(size_t count) {
return [count]() -> std::string {
char buf[32];
snprintf(buf, sizeof(buf), "num_reqs=%zu", count);
return std::string(buf);
};
inline TaggedEntryDetail Count(size_t count) {
assert(count <= std::numeric_limits<uint32_t>::max());
TaggedEntryDetail detail;
detail.kind = InjectedErrorLog::kDetailCount;
detail.entry.count = static_cast<uint32_t>(count);
return detail;
}
inline std::function<std::string()> ReqOffsetAndSize(size_t req_idx,
uint64_t offset,
inline TaggedEntryDetail ReqOffsetAndSize(size_t req_idx, uint64_t offset,
size_t n) {
return [req_idx, offset, n]() -> std::string {
char buf[96];
snprintf(buf, sizeof(buf), "req[%zu], offset=%llu, size=%zu", req_idx,
(unsigned long long)offset, n);
return std::string(buf);
};
assert(req_idx <= std::numeric_limits<uint32_t>::max());
TaggedEntryDetail detail;
detail.kind = InjectedErrorLog::kDetailReqOffsetAndSize;
detail.entry.req_idx = static_cast<uint32_t>(req_idx);
detail.entry.offset = offset;
detail.entry.size = static_cast<uint64_t>(n);
return detail;
}
} // namespace fault_injection_detail
@@ -473,7 +407,9 @@ class TestFSDirectory : public FSDirectory {
class FaultInjectionTestFS : public FileSystemWrapper {
public:
explicit FaultInjectionTestFS(const std::shared_ptr<FileSystem>& base)
explicit FaultInjectionTestFS(
const std::shared_ptr<FileSystem>& base,
const std::string& injected_error_log_path = std::string())
: FileSystemWrapper(base),
filesystem_active_(true),
filesystem_writable_(false),
@@ -487,7 +423,8 @@ class FaultInjectionTestFS : public FileSystemWrapper {
injected_thread_local_metadata_write_error_(
DeleteThreadLocalErrorContext),
ingest_data_corruption_before_write_(false),
checksum_handoff_func_type_(kCRC32c) {}
checksum_handoff_func_type_(kCRC32c),
injected_error_log_(injected_error_log_path) {}
virtual ~FaultInjectionTestFS() override { fs_error_.PermitUncheckedError(); }
static const char* kClassName() { return "FaultInjectionTestFS"; }
@@ -831,8 +768,10 @@ class FaultInjectionTestFS : public FileSystemWrapper {
IOStatus MaybeInjectThreadLocalError(
FaultInjectionIOType type, const IOOptions& io_options,
const char* op_name, const std::string& file_name,
std::function<std::string()> detail_fn = {}, ErrorOperation op = kUnknown,
Slice* slice = nullptr, bool direct_io = false, char* scratch = nullptr,
const InjectedErrorLog::TaggedEntryDetail& detail =
InjectedErrorLog::TaggedEntryDetail(),
ErrorOperation op = kUnknown, Slice* slice = nullptr,
bool direct_io = false, char* scratch = nullptr,
bool need_count_increase = false, bool* fault_injected = nullptr);
int GetAndResetInjectedThreadLocalErrorCount(FaultInjectionIOType type) {
@@ -901,21 +840,9 @@ class FaultInjectionTestFS : public FileSystemWrapper {
void ReadUnsynced(const std::string& fname, uint64_t offset, size_t n,
Slice* result, char* scratch, int64_t* pos_at_last_sync);
// Access the injected error log for printing on crash or test failure.
InjectedErrorLog& GetInjectedErrorLog() { return injected_error_log_; }
const InjectedErrorLog& GetInjectedErrorLog() const {
return injected_error_log_;
}
// Print recently injected errors to stderr. Call this on test failure
// to see what errors were injected leading up to the failure.
void PrintRecentInjectedErrors() const { injected_error_log_.PrintAll(); }
// Set the file path where PrintAll() will write its output.
// Must be called before any signal handler invocation.
void SetInjectedErrorLogPath(const std::string& path) {
injected_error_log_.SetLogFilePath(path);
}
// Flush recently injected errors to the configured binary log file and sync
// it to storage.
void FlushRecentInjectedErrors() const { injected_error_log_.Flush(); }
inline static const std::string kInjected = "injected";
@@ -927,6 +854,20 @@ class FaultInjectionTestFS : public FileSystemWrapper {
const char* op_name,
bool allow_missing_file);
inline static const std::string kInjectedReadError = "injected read error";
inline static const std::string kInjectedEmptyResult =
"injected empty result";
inline static const std::string kInjectedCorruptLastByte =
"injected corrupt last byte";
inline static const std::string kInjectedErrorResultMultiGetSingle =
"injected error result multiget single";
inline static const std::string kInjectedWriteError = "injected write error";
inline static const std::string kInjectedWriteErrorFailedToWriteToWAL =
"injected write error failed to write to WAL";
inline static const std::string kInjectedMetadataReadError =
"injected metadata read error";
inline static const std::string kInjectedMetadataWriteError =
"injected metadata write error";
port::Mutex mutex_;
std::map<std::string, FSFileState> db_file_state_;
std::map<std::string, FileOpenContract> open_managed_files_;
@@ -1000,9 +941,10 @@ class FaultInjectionTestFS : public FileSystemWrapper {
// because some fault is inected with IOStatus to be OK.
IOStatus MaybeInjectThreadLocalReadError(
const IOOptions& io_options, const char* op_name,
const std::string& file_name, std::function<std::string()> detail_fn,
ErrorOperation op, Slice* slice, bool direct_io, char* scratch,
bool need_count_increase, bool* fault_injected);
const std::string& file_name,
const InjectedErrorLog::TaggedEntryDetail& detail, ErrorOperation op,
Slice* slice, bool direct_io, char* scratch, bool need_count_increase,
bool* fault_injected);
bool ShouldExcludeFromFaultInjection(const std::string& file_name,
FaultInjectionIOType type) {
@@ -1091,37 +1033,28 @@ class FaultInjectionTestFS : public FileSystemWrapper {
std::string GetErrorMessage(FaultInjectionIOType type,
const std::string& file_name, ErrorOperation op) {
std::ostringstream msg;
msg << kInjected << " ";
switch (type) {
case FaultInjectionIOType::kRead:
msg << "read error";
break;
case FaultInjectionIOType::kWrite:
msg << "write error";
break;
case FaultInjectionIOType::kMetadataRead:
msg << "metadata read error";
break;
case FaultInjectionIOType::kMetadataWrite:
msg << "metadata write error";
break;
default:
assert(false);
break;
}
if (type == FaultInjectionIOType::kWrite &&
(op == ErrorOperation::kOpen || op == ErrorOperation::kAppend ||
op == ErrorOperation::kPositionedAppend)) {
return kInjectedReadError;
case FaultInjectionIOType::kWrite: {
if (op == ErrorOperation::kOpen || op == ErrorOperation::kAppend ||
op == ErrorOperation::kPositionedAppend) {
FileType file_type = kTempFile;
uint64_t ignore = 0;
if (TryParseFileName(file_name, &ignore, &file_type) &&
file_type == FileType::kWalFile) {
msg << " " << kFailedToWriteToWAL;
return kInjectedWriteErrorFailedToWriteToWAL;
}
}
return msg.str();
return kInjectedWriteError;
}
case FaultInjectionIOType::kMetadataRead:
return kInjectedMetadataReadError;
case FaultInjectionIOType::kMetadataWrite:
return kInjectedMetadataWriteError;
}
assert(false);
return kInjectedReadError;
}
};
utilities/fault_injection_fs_test.cc
+138 -38
View File
@@ -5,6 +5,8 @@
#include "utilities/fault_injection_fs.h"
#include <atomic>
#include <cstring>
#include <thread>
#include <vector>
@@ -12,7 +14,34 @@
namespace ROCKSDB_NAMESPACE {
class InjectedErrorLogTest : public testing::Test {};
class InjectedErrorLogTest : public testing::Test {
protected:
static std::string DecodeCString(const char* data, size_t len) {
size_t actual_len = 0;
while (actual_len < len && data[actual_len] != '\0') {
++actual_len;
}
return std::string(data, actual_len);
}
std::string ReadRawLog(const std::string& path) {
std::string raw;
Status s = ReadFileToString(Env::Default(), path, &raw);
EXPECT_OK(s);
return raw;
}
InjectedErrorLog::Entry DecodeEntry(const std::string& raw, size_t index) {
InjectedErrorLog::Entry entry{};
size_t offset = index * sizeof(InjectedErrorLog::Entry);
EXPECT_GE(raw.size(), offset + sizeof(entry));
if (raw.size() >= offset + sizeof(entry)) {
std::memcpy(&entry, raw.data() + offset, sizeof(entry));
}
return entry;
}
};
class FaultInjectionTestFSTest : public testing::Test {};
namespace {
@@ -46,53 +75,91 @@ class CloseCountingWritableFile : public FSWritableFileOwnerWrapper {
} // namespace
// Test basic Record and PrintAll functionality.
TEST_F(InjectedErrorLogTest, BasicRecordAndPrint) {
InjectedErrorLog log;
log.SetLogFilePath("/dev/null");
// Record some entries.
log.Record("op=Get key=0x%08x status=%s", 0x12345678, "OK");
log.Record("op=Put key=0x%08x value_size=%d", 0xABCDEF00, 100);
log.Record("op=Delete key=0x%08x", 0x00000001);
// PrintAll should not crash.
log.PrintAll();
// Goal: verify a single structured record is persisted to the binary log with
// the expected fixed-width fields. The test writes one entry, flushes the
// file, and decodes the raw bytes back into an entry struct.
TEST_F(InjectedErrorLogTest, BasicRecordAndFlush) {
std::string path = test::PerThreadDBPath("injected_error_log_basic.bin");
{
InjectedErrorLog log(path);
IOStatus status = IOStatus::IOError("injected write error");
status.SetRetryable(false);
status.SetDataLoss(true);
log.Record("Append", "/tmp/000001.log",
fault_injection_detail::OffsetSizeAndHead(7, Slice("abcd", 4)),
status);
log.Flush();
}
// Test that the circular buffer wraps correctly.
TEST_F(InjectedErrorLogTest, CircularBufferWrap) {
InjectedErrorLog log;
log.SetLogFilePath("/dev/null");
// Fill beyond kMaxEntries to trigger wraparound.
for (size_t i = 0; i < InjectedErrorLog::kMaxEntries + 100; i++) {
log.Record("entry=%zu", i);
std::string raw = ReadRawLog(path);
ASSERT_EQ(raw.size(), sizeof(InjectedErrorLog::Entry));
auto entry = DecodeEntry(raw, 0);
EXPECT_NE(entry.timestamp_us, 0U);
EXPECT_EQ(entry.detail.offset, 7U);
EXPECT_EQ(entry.detail.size, 4U);
EXPECT_EQ(entry.detail_kind, InjectedErrorLog::kDetailOffsetSizeAndHead);
EXPECT_EQ(entry.detail_payload_size, 4U);
EXPECT_EQ(entry.retryable, 0U);
EXPECT_EQ(entry.data_loss, 1U);
EXPECT_EQ(DecodeCString(entry.op_name, sizeof(entry.op_name)), "Append");
EXPECT_EQ(DecodeCString(entry.file_name, sizeof(entry.file_name)),
"/tmp/000001.log");
EXPECT_EQ(DecodeCString(entry.status_message, sizeof(entry.status_message)),
"injected write error");
EXPECT_EQ(std::string(entry.detail.payload, entry.detail.payload + 4),
"abcd");
}
// PrintAll should handle the wrapped buffer without crashing.
log.PrintAll();
// Goal: verify the file-backed logger keeps all records instead of truncating
// to the old in-memory ring size. The test writes more than 1,000 entries and
// checks that both the first and last ones are still present in the file.
TEST_F(InjectedErrorLogTest, DirectLogKeepsAllEntries) {
std::string path =
test::PerThreadDBPath("injected_error_log_all_entries.bin");
{
InjectedErrorLog log(path);
IOStatus status = IOStatus::IOError("injected write error");
constexpr size_t kNumEntries = 1100;
for (size_t i = 0; i < kNumEntries; ++i) {
std::string file_name = "file" + std::to_string(i);
log.Record("Append", file_name, fault_injection_detail::NoDetail(),
status);
}
log.Flush();
}
// Test concurrent Record() from multiple threads.
// Keep total records (kNumThreads * kRecordsPerThread) under kMaxEntries
// to avoid write-write races from buffer wraparound, which are benign but
// would trigger TSAN warnings.
std::string raw = ReadRawLog(path);
constexpr size_t kNumEntries = 1100;
ASSERT_EQ(raw.size(), kNumEntries * sizeof(InjectedErrorLog::Entry));
auto first = DecodeEntry(raw, 0);
auto last = DecodeEntry(raw, kNumEntries - 1);
EXPECT_EQ(DecodeCString(first.file_name, sizeof(first.file_name)), "file0");
EXPECT_EQ(DecodeCString(last.file_name, sizeof(last.file_name)),
"file" + std::to_string(kNumEntries - 1));
}
// Goal: verify concurrent Record() calls append independent entries to the
// shared file. The test has several threads emit records concurrently and then
// checks the raw file size matches the total number of writes.
TEST_F(InjectedErrorLogTest, ConcurrentRecord) {
InjectedErrorLog log;
std::string path = test::PerThreadDBPath("injected_error_log_concurrent.bin");
constexpr int kNumThreads = 4;
constexpr int kRecordsPerThread = 200;
static_assert(kNumThreads * kRecordsPerThread <
static_cast<int>(InjectedErrorLog::kMaxEntries),
"total records must stay within buffer to avoid TSAN-visible "
"write-write races on overlapping slots");
{
InjectedErrorLog log(path);
IOStatus status = IOStatus::IOError("injected read error");
std::vector<std::thread> threads;
threads.reserve(kNumThreads);
for (int t = 0; t < kNumThreads; t++) {
threads.emplace_back([&log, t]() {
threads.emplace_back([&log, &status, t]() {
for (int i = 0; i < kRecordsPerThread; i++) {
log.Record("thread=%d iter=%d op=Get key=0x%08x", t, i, i * 17);
std::string file_name =
"thread" + std::to_string(t) + "_" + std::to_string(i);
log.Record("Read", file_name, fault_injection_detail::NoDetail(),
status);
}
});
}
@@ -101,12 +168,45 @@ TEST_F(InjectedErrorLogTest, ConcurrentRecord) {
t.join();
}
// PrintAll after all threads are done -- no race.
log.SetLogFilePath("/dev/null");
log.PrintAll();
log.Flush();
}
// Test HexHead utility.
std::string raw = ReadRawLog(path);
ASSERT_EQ(raw.size(), static_cast<size_t>(kNumThreads * kRecordsPerThread) *
sizeof(InjectedErrorLog::Entry));
}
// Goal: verify long file paths are suffix-truncated so the basename survives
// in the fixed-width record. The test logs a path longer than the file-name
// field and checks the stored sample matches the expected tail bytes.
TEST_F(InjectedErrorLogTest, LongFileNameKeepsSuffix) {
std::string path =
test::PerThreadDBPath("injected_error_log_suffix_truncation.bin");
const std::string long_file_name =
"/tmp/rocksdb_crashtest/artifacts/very/long/path/that/keeps/growing/"
"db_crashtest/fault_injection/000123.sst";
ASSERT_GT(long_file_name.size(), InjectedErrorLog::kMaxFileNameLen - 1);
{
InjectedErrorLog log(path);
IOStatus status = IOStatus::IOError("injected write error");
log.Record("Append", long_file_name, fault_injection_detail::NoDetail(),
status);
log.Flush();
}
std::string raw = ReadRawLog(path);
auto entry = DecodeEntry(raw, 0);
std::string stored = DecodeCString(entry.file_name, sizeof(entry.file_name));
std::string expected = long_file_name.substr(
long_file_name.size() - (InjectedErrorLog::kMaxFileNameLen - 1));
EXPECT_EQ(stored, expected);
EXPECT_EQ(stored.compare(stored.size() - strlen("000123.sst"),
strlen("000123.sst"), "000123.sst"),
0);
}
// Goal: verify the human-readable hex helper still formats the payload samples
// exactly as expected, since the Python decoder depends on the same output.
TEST_F(InjectedErrorLogTest, HexHead) {
const char data[] = "\x01\x02\xAB\xCD";
std::string result = InjectedErrorLog::HexHead(data, 4);