1
0
mirror of https://github.com/facebook/rocksdb.git synced 2026-10-23 08:09:02 +03:00

Add injected error log ring buffer for fault injection diagnostics (#14431)

Summary:
Add a circular ring buffer (InjectedErrorLog) to FaultInjectionTestFS that records the last 1000 injected errors. Each entry captures the timestamp, thread ID, FS API name with all arguments (file path, offset, buffer size, first 8 bytes of data in hex), and the injected error status. For example:

  Append("/path/035354.sst", size=4096, head=[1a 2b 3c ...]) -> IOError (Retryable): injected write error
  RenameFile("/path/tmp.sst", "/path/035355.sst") -> IOError: injected metadata write error
  Read(offset=16384, size=4096) -> IOError (Retryable): injected read error

The ring buffer is printed to a file automatically:
- On any fatal signal (SIGABRT, SIGSEGV, SIGTERM, SIGINT, SIGHUP, SIGFPE, SIGBUS, SIGILL, SIGQUIT, SIGXCPU, SIGXFSZ, SIGSYS) via a registered crash callback
- At the end of db_stress main(), for diagnostic visibility even when the test completes normally

This addresses a key debugging gap: when write fault injection causes secondary failures (e.g., the builder error propagation issue in T257612259), the injected errors were previously completely silent with no logging trail. The ring buffer provides the missing diagnostic context to correlate fault injection with downstream failures.

Changes:
- port/stack_trace.h/.cc: Add RegisterCrashCallback() API; extend InstallStackTraceHandler() to catch all catchable termination signals
- utilities/fault_injection_fs.h: Add InjectedErrorLog class with printf-style Record(), HexHead() for data bytes, and signal-safe PrintAll()
- utilities/fault_injection_fs.cc: Record full API arguments and error status at all 31 fault injection call sites
- db_stress_tool/db_stress_tool.cc: Register crash callback and print ring buffer at end of main()

Pull Request resolved: https://github.com/facebook/rocksdb/pull/14431

Reviewed By: hx235

Differential Revision: D95435430

Pulled By: xingbowang

fbshipit-source-id: 6c18e1b072044575d6c8c3f198070127b0f80608
This commit is contained in:
Xingbo Wang
2026-03-11 18:15:04 -07:00
committed by meta-codesync[bot]
parent cb8bc56d14
commit 91f227d9be
6 changed files with 492 additions and 73 deletions

View File

@@ -24,6 +24,7 @@
#include "db_stress_tool/db_stress_common.h"
#include "db_stress_tool/db_stress_driver.h"
#include "db_stress_tool/db_stress_shared_state.h"
#include "port/stack_trace.h"
#include "rocksdb/convenience.h"
#include "utilities/fault_injection_fs.h"
@@ -92,6 +93,15 @@ int db_stress_tool(int argc, char** argv) {
fault_env_guard =
std::make_shared<CompositeEnvWrapper>(raw_env, fault_fs_guard);
raw_env = fault_env_guard.get();
// Register a crash callback so that recently injected errors are
// printed to stderr when the process crashes (SIGABRT, SIGSEGV, etc.).
// This helps diagnose stress test failures caused by fault injection.
port::RegisterCrashCallback([]() {
if (fault_fs_guard) {
fault_fs_guard->PrintRecentInjectedErrors();
}
});
}
auto db_stress_fs =
@@ -301,6 +311,25 @@ int db_stress_tool(int argc, char** argv) {
FLAGS_db = default_db_path;
}
// Now that FLAGS_db is resolved, set the fault injection log file path
// so that PrintAll() writes to a file instead of stderr (signal-safe).
// Store the log in TEST_TMPDIR (outside the DB directory) so it survives
// DB reopen (which cleans untracked files) and gets included in the
// sandcastle db.tar.gz artifact for post-failure analysis.
if (fault_fs_guard) {
std::string log_dir;
const char* test_tmpdir = getenv("TEST_TMPDIR");
if (test_tmpdir && test_tmpdir[0] != '\0') {
log_dir = test_tmpdir;
} else {
log_dir = "/tmp";
}
std::string log_path = log_dir + "/fault_injection_" +
std::to_string(getpid()) + "_" +
std::to_string(time(nullptr)) + ".log";
fault_fs_guard->SetInjectedErrorLogPath(log_path);
}
if ((FLAGS_test_secondary || FLAGS_continuous_verification_interval > 0) &&
FLAGS_secondaries_base.empty()) {
std::string default_secondaries_path;

View File

@@ -18,6 +18,7 @@ void PrintAndFreeStack(void* /*callstack*/, int /*num_frames*/) {}
void* SaveStack(int* /*num_frames*/, int /*first_frames_to_skip*/) {
return nullptr;
}
void RegisterCrashCallback(CrashCallback /*callback*/) {}
} // namespace port
} // namespace ROCKSDB_NAMESPACE
@@ -320,6 +321,7 @@ void* SaveStack(int* num_frames, int first_frames_to_skip) {
static std::atomic<uint64_t> g_thread_handling_stack_trace{0};
static int g_recursion_count = 0;
static std::atomic<bool> g_at_exit_called{false};
static std::atomic<CrashCallback> g_crash_callback{nullptr};
static void StackTraceHandler(int sig) {
fprintf(stderr, "Received signal %d (%s)\n", sig, strsignal(sig));
@@ -360,6 +362,12 @@ static void StackTraceHandler(int sig) {
fprintf(stderr, "In a race with process already exiting...\n");
}
// Invoke registered crash callback before printing stack trace
auto callback = g_crash_callback.load(std::memory_order_acquire);
if (callback) {
callback();
}
// skip the top three signal handler related frames
PrintStack(3);
@@ -398,13 +406,41 @@ static void AtExit() {
g_at_exit_called.store(true, std::memory_order_release);
}
// Lightweight handler for graceful termination signals (SIGTERM, SIGINT).
// Prints the crash callback (e.g., ring buffer) but skips the
// expensive GDB/LLDB stack trace, since these are intentional terminations.
static void TerminationHandler(int sig) {
char buf[64];
int len = snprintf(buf, sizeof(buf), "Received signal %d (%s)\n", sig,
strsignal(sig));
if (len > 0) {
auto unused __attribute__((unused)) = write(STDOUT_FILENO, buf, len);
}
auto callback = g_crash_callback.load(std::memory_order_acquire);
if (callback) {
callback();
}
signal(sig, SIG_DFL);
raise(sig);
}
void RegisterCrashCallback(CrashCallback callback) {
g_crash_callback.store(callback, std::memory_order_release);
}
void InstallStackTraceHandler() {
// just use the plain old signal as it's simple and sufficient
// for this use case
// Crash signals — invoke full stack trace + ring buffer
signal(SIGILL, StackTraceHandler);
signal(SIGSEGV, StackTraceHandler);
signal(SIGBUS, StackTraceHandler);
signal(SIGABRT, StackTraceHandler);
signal(SIGFPE, StackTraceHandler);
signal(SIGQUIT, StackTraceHandler);
// Termination signals — print ring buffer only, no stack trace
signal(SIGTERM, TerminationHandler);
signal(SIGINT, TerminationHandler);
atexit(AtExit);
// Allow ouside debugger to attach, even with Yama security restrictions.
// This is needed even outside of PrintStack() so that external mechanisms

View File

@@ -27,5 +27,14 @@ void PrintAndFreeStack(void* callstack, int num_frames);
// Save the current callstack
void* SaveStack(int* num_frame, int first_frames_to_skip = 0);
// Register a callback to be invoked when a fatal signal is received,
// before the stack trace is printed. This is useful for printing diagnostic
// information (e.g., recently injected errors) when a crash occurs.
// The callback must only call async-signal-safe functions (write, snprintf,
// etc.) or functions that are safe enough in practice (fprintf to stderr).
// Only one callback is supported; subsequent calls overwrite the previous one.
using CrashCallback = void (*)();
void RegisterCrashCallback(CrashCallback callback);
} // namespace port
} // namespace ROCKSDB_NAMESPACE

View File

@@ -2,6 +2,7 @@
# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved.
import argparse
import glob
import math
import os
import random
@@ -1418,11 +1419,16 @@ def execute_cmd(cmd, timeout=None, timeout_pstack=False):
hit_timeout = True
if timeout_pstack:
os.system("pstack %d" % pid)
child.kill()
print("KILLED %d\n" % child.pid)
outs, errs = child.communicate()
child.terminate() # SIGTERM — triggers TerminationHandler
try:
outs, errs = child.communicate(timeout=3)
print("TERMINATED %d\n" % child.pid)
except subprocess.TimeoutExpired:
child.kill()
print("KILLED %d (SIGTERM did not work)\n" % child.pid)
outs, errs = child.communicate()
return hit_timeout, child.returncode, outs.decode("utf-8"), errs.decode("utf-8")
return hit_timeout, child.returncode, outs.decode("utf-8"), errs.decode("utf-8"), pid
def print_output_and_exit_on_error(stdout, stderr, print_stderr_separately=False):
@@ -1453,6 +1459,54 @@ def cleanup_after_success(dbname):
sys.exit(2)
def print_and_cleanup_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)
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="",
)
print("".join(entries[-max_tail_entries:]), end="")
print(
"=== Showed %d of %d injected error entries ===\n"
% (max_tail_entries, total_entries),
end="",
)
except OSError:
pass
# This script runs and kills db_stress multiple times. It checks consistency
# in case of unsafe crashes in RocksDB.
def blackbox_crash_main(args, unknown_args):
@@ -1476,7 +1530,9 @@ def blackbox_crash_main(args, unknown_args):
dict(list(cmd_params.items()) + list({"db": dbname}.items())), unknown_args
)
hit_timeout, retcode, outs, errs = execute_cmd(cmd, cmd_params["interval"])
hit_timeout, retcode, outs, errs, pid = execute_cmd(cmd, cmd_params["interval"])
print_and_cleanup_fault_injection_log(pid)
# Reset destroy_db_initially after each run (it may have been set by
# command line for first run only)
@@ -1502,10 +1558,12 @@ def blackbox_crash_main(args, unknown_args):
cmd = gen_cmd(
dict(list(cmd_params.items()) + list({"db": dbname}.items())), unknown_args
)
hit_timeout, retcode, outs, errs = execute_cmd(
hit_timeout, retcode, outs, errs, pid = execute_cmd(
cmd, cmd_params["verify_timeout"], True
)
print_and_cleanup_fault_injection_log(pid)
# For the final run
print_output_and_exit_on_error(outs, errs, args.print_stderr_separately)
@@ -1646,7 +1704,7 @@ def whitebox_crash_main(args, unknown_args):
# for job scheduling or execution.
# TODO detect a hanging condition. The job might run too long as RocksDB
# hits a hanging bug.
hit_timeout, retncode, stdoutdata, stderrdata = execute_cmd(
hit_timeout, retncode, stdoutdata, stderrdata, pid = execute_cmd(
cmd, exit_time - time.time() + 900
)

View File

@@ -98,7 +98,7 @@ IOStatus TestFSDirectory::Fsync(const IOOptions& options, IODebugContext* dbg) {
}
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataWrite, options);
FaultInjectionIOType::kMetadataWrite, options, "Fsync", dirname_);
if (!s.ok()) {
return s;
}
@@ -114,7 +114,7 @@ IOStatus TestFSDirectory::Close(const IOOptions& options, IODebugContext* dbg) {
}
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataWrite, options);
FaultInjectionIOType::kMetadataWrite, options, "Close", dirname_);
if (!s.ok()) {
return s;
}
@@ -130,7 +130,8 @@ IOStatus TestFSDirectory::FsyncWithDirOptions(
return fs_->GetError();
}
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataWrite, options);
FaultInjectionIOType::kMetadataWrite, options, "FsyncWithDirOptions",
dirname_);
if (!s.ok()) {
return s;
}
@@ -169,7 +170,8 @@ IOStatus TestFSWritableFile::Append(const Slice& data, const IOOptions& options,
}
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kWrite, options, state_.filename_,
FaultInjectionIOType::kWrite, options, "Append", state_.filename_,
fault_injection_detail::SizeAndHead(data),
FaultInjectionTestFS::ErrorOperation::kAppend);
if (!s.ok()) {
return s;
@@ -205,7 +207,8 @@ IOStatus TestFSWritableFile::Append(
}
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kWrite, options, state_.filename_,
FaultInjectionIOType::kWrite, options, "Append", state_.filename_,
fault_injection_detail::SizeAndHead(data),
FaultInjectionTestFS::ErrorOperation::kAppend);
if (!s.ok()) {
return s;
@@ -244,8 +247,9 @@ IOStatus TestFSWritableFile::Truncate(uint64_t size, const IOOptions& options,
if (!fs_->IsFilesystemActive()) {
return fs_->GetError();
}
IOStatus s = fs_->MaybeInjectThreadLocalError(FaultInjectionIOType::kWrite,
options, state_.filename_);
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kWrite, options, "Truncate", state_.filename_,
fault_injection_detail::Size(size));
if (!s.ok()) {
return s;
}
@@ -269,7 +273,8 @@ IOStatus TestFSWritableFile::PositionedAppend(const Slice& data,
return IOStatus::Corruption("Data is corrupted!");
}
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kWrite, options, state_.filename_,
FaultInjectionIOType::kWrite, options, "PositionedAppend",
state_.filename_, fault_injection_detail::OffsetSizeAndHead(offset, data),
FaultInjectionTestFS::ErrorOperation::kPositionedAppend);
if (!s.ok()) {
return s;
@@ -296,7 +301,8 @@ IOStatus TestFSWritableFile::PositionedAppend(
return IOStatus::Corruption("Data is corrupted!");
}
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kWrite, options, state_.filename_,
FaultInjectionIOType::kWrite, options, "PositionedAppend",
state_.filename_, fault_injection_detail::OffsetSizeAndHead(offset, data),
FaultInjectionTestFS::ErrorOperation::kPositionedAppend);
if (!s.ok()) {
return s;
@@ -332,7 +338,7 @@ IOStatus TestFSWritableFile::Close(const IOOptions& options,
return fs_->GetError();
}
IOStatus io_s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataWrite, options);
FaultInjectionIOType::kMetadataWrite, options, "Close", state_.filename_);
if (!io_s.ok()) {
return io_s;
}
@@ -460,7 +466,10 @@ IOStatus TestFSRandomRWFile::Sync(const IOOptions& options,
TestFSRandomAccessFile::TestFSRandomAccessFile(
const std::string& fname, std::unique_ptr<FSRandomAccessFile>&& f,
FaultInjectionTestFS* fs)
: target_(std::move(f)), fs_(fs), is_sst_(EndsWith(fname, ".sst")) {
: fname_(fname),
target_(std::move(f)),
fs_(fs),
is_sst_(EndsWith(fname, ".sst")) {
assert(target_ != nullptr);
}
@@ -473,7 +482,8 @@ IOStatus TestFSRandomAccessFile::Read(uint64_t offset, size_t n,
return fs_->GetError();
}
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kRead, options, "",
FaultInjectionIOType::kRead, options, "Read", fname_,
fault_injection_detail::OffsetAndSize(offset, n),
FaultInjectionTestFS::ErrorOperation::kRead, result, use_direct_io(),
scratch, /*need_count_increase=*/true,
/*fault_injected=*/nullptr);
@@ -498,7 +508,8 @@ IOStatus TestFSRandomAccessFile::ReadAsync(
}
if (res_status.ok()) {
res_status = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kRead, opts, "",
FaultInjectionIOType::kRead, opts, "ReadAsync", fname_,
fault_injection_detail::OffsetAndSize(req.offset, req.len),
FaultInjectionTestFS::ErrorOperation::kRead, &res.result,
use_direct_io(), req.scratch, /*need_count_increase=*/true,
/*fault_injected=*/nullptr);
@@ -538,7 +549,9 @@ IOStatus TestFSRandomAccessFile::MultiRead(FSReadRequest* reqs, size_t num_reqs,
}
bool this_injected_error;
reqs[i].status = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kRead, options, "",
FaultInjectionIOType::kRead, options, "MultiRead", fname_,
fault_injection_detail::ReqOffsetAndSize(i, reqs[i].offset,
reqs[i].len),
FaultInjectionTestFS::ErrorOperation::kRead, &(reqs[i].result),
use_direct_io(), reqs[i].scratch,
/*need_count_increase=*/true,
@@ -547,7 +560,8 @@ IOStatus TestFSRandomAccessFile::MultiRead(FSReadRequest* reqs, size_t num_reqs,
}
if (s.ok()) {
s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kRead, options, "",
FaultInjectionIOType::kRead, options, "MultiRead", "",
fault_injection_detail::Count(num_reqs),
FaultInjectionTestFS::ErrorOperation::kMultiRead, nullptr,
use_direct_io(), nullptr, /*need_count_increase=*/!injected_error,
/*fault_injected=*/nullptr);
@@ -617,7 +631,8 @@ IOStatus TestFSSequentialFile::Read(size_t n, const IOOptions& options,
Slice* result, char* scratch,
IODebugContext* dbg) {
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kRead, options, "",
FaultInjectionIOType::kRead, options, "Read", fname_,
fault_injection_detail::Size(n),
FaultInjectionTestFS::ErrorOperation::kRead, result, use_direct_io(),
scratch, true /*need_count_increase=*/, nullptr /* fault_injected*/);
if (!s.ok()) {
@@ -768,7 +783,8 @@ IOStatus TestFSSequentialFile::PositionedRead(uint64_t offset, size_t n,
Slice* result, char* scratch,
IODebugContext* dbg) {
IOStatus s = fs_->MaybeInjectThreadLocalError(
FaultInjectionIOType::kRead, options, "",
FaultInjectionIOType::kRead, options, "PositionedRead", fname_,
fault_injection_detail::OffsetAndSize(offset, n),
FaultInjectionTestFS::ErrorOperation::kRead, result, use_direct_io(),
scratch, true /*need_count_increase=*/, nullptr /* fault_injected */);
if (!s.ok()) {
@@ -800,8 +816,8 @@ IOStatus FaultInjectionTestFS::FileExists(const std::string& fname,
return GetError();
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataRead, options, "FileExists", fname);
if (!io_s.ok()) {
return io_s;
}
@@ -818,8 +834,8 @@ IOStatus FaultInjectionTestFS::GetChildren(const std::string& dir,
return GetError();
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataRead, options, "GetChildren", dir);
if (!io_s.ok()) {
return io_s;
}
@@ -836,7 +852,8 @@ IOStatus FaultInjectionTestFS::GetChildrenFileAttributes(
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options,
"GetChildrenFileAttributes", dir);
if (!io_s.ok()) {
return io_s;
}
@@ -857,8 +874,8 @@ IOStatus FaultInjectionTestFS::NewWritableFile(
}
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kWrite, file_opts.io_options, fname,
FaultInjectionTestFS::ErrorOperation::kOpen);
FaultInjectionIOType::kWrite, file_opts.io_options, "NewWritableFile",
fname, {}, FaultInjectionTestFS::ErrorOperation::kOpen);
if (!io_s.ok()) {
return io_s;
}
@@ -894,7 +911,8 @@ IOStatus FaultInjectionTestFS::ReopenWritableFile(
return target()->ReopenWritableFile(fname, file_opts, result, dbg);
}
IOStatus io_s = MaybeInjectThreadLocalError(FaultInjectionIOType::kWrite,
file_opts.io_options, fname);
file_opts.io_options,
"ReopenWritableFile", fname);
if (!io_s.ok()) {
return io_s;
}
@@ -970,7 +988,8 @@ IOStatus FaultInjectionTestFS::NewRandomRWFile(
return target()->NewRandomRWFile(fname, file_opts, result, dbg);
}
IOStatus io_s = MaybeInjectThreadLocalError(FaultInjectionIOType::kWrite,
file_opts.io_options, fname);
file_opts.io_options,
"NewRandomRWFile", fname);
if (!io_s.ok()) {
return io_s;
}
@@ -1002,10 +1021,10 @@ IOStatus FaultInjectionTestFS::NewRandomAccessFile(
return GetError();
}
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kRead, file_opts.io_options, fname,
ErrorOperation::kOpen, nullptr /* result */, false /* direct_io */,
nullptr /* scratch */, true /*need_count_increase*/,
nullptr /*fault_injected*/);
FaultInjectionIOType::kRead, file_opts.io_options, "NewRandomAccessFile",
fname, {}, ErrorOperation::kOpen, nullptr /* result */,
false /* direct_io */, nullptr /* scratch */,
true /*need_count_increase*/, nullptr /*fault_injected*/);
if (!io_s.ok()) {
return io_s;
}
@@ -1025,10 +1044,10 @@ IOStatus FaultInjectionTestFS::NewSequentialFile(
return GetError();
}
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kRead, file_opts.io_options, fname,
ErrorOperation::kOpen, nullptr /* result */, false /* direct_io */,
nullptr /* scratch */, true /*need_count_increase*/,
nullptr /*fault_injected*/);
FaultInjectionIOType::kRead, file_opts.io_options, "NewSequentialFile",
fname, {}, ErrorOperation::kOpen, nullptr /* result */,
false /* direct_io */, nullptr /* scratch */,
true /*need_count_increase*/, nullptr /*fault_injected*/);
if (!io_s.ok()) {
return io_s;
}
@@ -1048,7 +1067,7 @@ IOStatus FaultInjectionTestFS::DeleteFile(const std::string& f,
return GetError();
}
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataWrite, options);
FaultInjectionIOType::kMetadataWrite, options, "DeleteFile", f);
if (!io_s.ok()) {
return io_s;
}
@@ -1071,8 +1090,8 @@ IOStatus FaultInjectionTestFS::GetFileSize(const std::string& f,
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataRead, options, "GetFileSize", f);
if (!io_s.ok()) {
return io_s;
}
@@ -1101,7 +1120,8 @@ IOStatus FaultInjectionTestFS::GetFileModificationTime(const std::string& fname,
return GetError();
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options,
"GetFileModificationTime", fname);
if (!io_s.ok()) {
return io_s;
}
@@ -1118,12 +1138,13 @@ IOStatus FaultInjectionTestFS::RenameFile(const std::string& s,
return GetError();
}
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataWrite, options);
FaultInjectionIOType::kMetadataWrite, options, "RenameFile", s,
fault_injection_detail::TwoFiles(s, t));
if (!io_s.ok()) {
return io_s;
}
// We preserve contents of overwritten files up to a size threshold.
// We preserve contents of overwritten files
// We could keep previous file in another name, but we need to worry about
// garbage collect the those files. We do it if it is needed later.
// We ignore I/O errors here for simplicity.
@@ -1164,7 +1185,8 @@ IOStatus FaultInjectionTestFS::LinkFile(const std::string& s,
return GetError();
}
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataWrite, options);
FaultInjectionIOType::kMetadataWrite, options, "LinkFile", s,
fault_injection_detail::TwoFiles(s, t));
if (!io_s.ok()) {
return io_s;
}
@@ -1208,8 +1230,8 @@ IOStatus FaultInjectionTestFS::NumFileLinks(const std::string& fname,
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataRead, options, "NumFileLinks", fname);
if (!io_s.ok()) {
return io_s;
}
@@ -1225,8 +1247,9 @@ IOStatus FaultInjectionTestFS::AreFilesSame(const std::string& first,
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataRead, options, "AreFilesSame", first,
fault_injection_detail::TwoFiles(first, second));
if (!io_s.ok()) {
return io_s;
}
@@ -1242,8 +1265,8 @@ IOStatus FaultInjectionTestFS::GetAbsolutePath(const std::string& db_path,
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataRead, options, "GetAbsolutePath", db_path);
if (!io_s.ok()) {
return io_s;
}
@@ -1258,8 +1281,8 @@ IOStatus FaultInjectionTestFS::IsDirectory(const std::string& path,
if (!IsFilesystemActive()) {
return GetError();
}
IOStatus io_s =
MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead, options);
IOStatus io_s = MaybeInjectThreadLocalError(
FaultInjectionIOType::kMetadataRead, options, "IsDirectory", path);
if (!io_s.ok()) {
return io_s;
}
@@ -1389,9 +1412,10 @@ void FaultInjectionTestFS::UntrackFile(const std::string& f) {
}
IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalReadError(
const IOOptions& io_options, ErrorOperation op, Slice* result,
bool direct_io, char* scratch, bool need_count_increase,
bool* fault_injected) {
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) {
bool dummy_bool;
bool& ret_fault_injected = fault_injected ? *fault_injected : dummy_bool;
ret_fault_injected = false;
@@ -1461,6 +1485,17 @@ 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());
}
}
return ret;
}
@@ -1472,14 +1507,14 @@ bool FaultInjectionTestFS::TryParseFileName(const std::string& file_name,
}
IOStatus FaultInjectionTestFS::MaybeInjectThreadLocalError(
FaultInjectionIOType type, const IOOptions& io_options,
const std::string& file_name, ErrorOperation op, Slice* result,
bool direct_io, char* scratch, bool need_count_increase,
bool* fault_injected) {
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) {
if (type == FaultInjectionIOType::kRead) {
return MaybeInjectThreadLocalReadError(io_options, op, result, direct_io,
scratch, need_count_increase,
fault_injected);
return MaybeInjectThreadLocalReadError(
io_options, op_name, file_name, std::move(detail_fn), op, result,
direct_io, scratch, need_count_increase, fault_injected);
}
ErrorContext* ctx = GetErrorContextFromFaultInjectionIOType(type);
@@ -1501,6 +1536,17 @@ 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());
}
}
if (type == FaultInjectionIOType::kWrite) {
TEST_SYNC_POINT(
"FaultInjectionTestFS::InjectMetadataWriteError:Injected");

View File

@@ -17,9 +17,25 @@
#pragma once
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cstdarg>
#include <functional>
#include <map>
#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 "rocksdb/file_system.h"
@@ -29,9 +45,215 @@
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).
class InjectedErrorLog {
public:
static constexpr size_t kMaxEntries = 1000;
static constexpr size_t kMaxMessageLen = 256;
struct Entry {
uint64_t timestamp_us;
uint64_t thread_id;
char context[kMaxMessageLen];
};
InjectedErrorLog() : head_(0), entries_{} { log_path_[0] = '\0'; }
// 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';
}
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();
va_list args;
va_start(args, fmt);
vsnprintf(e.context, kMaxMessageLen, fmt, args);
va_end(args);
}
// Format the first few bytes of a buffer as hex for logging.
// Returns a string like "ab cd ef 01 02 ..."
static std::string HexHead(const char* data, size_t size,
size_t max_bytes = 8) {
std::string result;
size_t n = std::min(size, max_bytes);
char buf[4];
for (size_t i = 0; i < n; i++) {
snprintf(buf, sizeof(buf), "%02x ", (unsigned char)data[i]);
result += buf;
}
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.
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;
const Entry& e = entries_[idx];
if (e.timestamp_us == 0) continue;
uint64_t secs = e.timestamp_us / 1000000;
uint64_t usecs = e.timestamp_us % 1000000;
len = snprintf(buf, sizeof(buf), "[%llu.%06llu] thread=%llu: %s\n",
(unsigned long long)secs, (unsigned long long)usecs,
(unsigned long long)e.thread_id, e.context);
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
}
private:
std::atomic<size_t> head_;
Entry entries_[kMaxEntries];
char log_path_[PATH_MAX];
};
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.
namespace fault_injection_detail {
inline std::function<std::string()> NoDetail() { return {}; }
inline std::function<std::string()> 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);
};
}
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 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 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 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 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 std::function<std::string()> 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);
};
}
} // namespace fault_injection_detail
enum class FaultInjectionIOType {
kRead = 0,
kWrite,
@@ -161,6 +383,7 @@ class TestFSRandomAccessFile : public FSRandomAccessFile {
IOStatus GetFileSize(uint64_t* file_size) override;
private:
std::string fname_;
std::unique_ptr<FSRandomAccessFile> target_;
FaultInjectionTestFS* fs_;
const bool is_sst_;
@@ -332,7 +555,7 @@ class FaultInjectionTestFS : public FileSystemWrapper {
*disk_free = 0;
} else {
io_s = MaybeInjectThreadLocalError(FaultInjectionIOType::kMetadataRead,
options);
options, "GetFreeSpace", path);
if (io_s.ok()) {
io_s = target()->GetFreeSpace(path, options, disk_free, dbg);
}
@@ -545,7 +768,8 @@ class FaultInjectionTestFS : public FileSystemWrapper {
IOStatus MaybeInjectThreadLocalError(
FaultInjectionIOType type, const IOOptions& io_options,
const std::string& file_name = "", ErrorOperation op = kUnknown,
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,
bool need_count_increase = false, bool* fault_injected = nullptr);
@@ -610,6 +834,22 @@ 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);
}
inline static const std::string kInjected = "injected";
private:
@@ -676,6 +916,7 @@ class FaultInjectionTestFS : public FileSystemWrapper {
bool fail_get_file_unique_id_ = false;
bool fail_random_access_get_file_size_sst_ = false;
bool fail_fs_get_file_size_sst_ = false;
InjectedErrorLog injected_error_log_;
// Inject an error. For a READ operation, a status of IOError(), a
// corruption in the contents of scratch, or truncation of slice
@@ -683,11 +924,11 @@ class FaultInjectionTestFS : public FileSystemWrapper {
// its always an IOError.
// fault_injected returns whether a fault is injected. It is needed
// because some fault is inected with IOStatus to be OK.
IOStatus MaybeInjectThreadLocalReadError(const IOOptions& io_options,
ErrorOperation op, Slice* slice,
bool direct_io, char* scratch,
bool need_count_increase,
bool* fault_injected);
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);
bool ShouldExcludeFromWriteFaultInjection(const std::string& file_name) {
MutexLock l(&mutex_);