From 1054f2722ecfa10e532d4246672124ebd76212f8 Mon Sep 17 00:00:00 2001 From: Googler Date: Thu, 24 Sep 2026 16:57:29 -0700 Subject: [PATCH] Allow configuring fault injection rules via a plain text file. When the file is absent - no rules are installed. PiperOrigin-RevId: 987853071 --- tpu_sync/fault_injection/BUILD | 2 +- tpu_sync/fault_injection/fault_injector.cc | 144 ++++++++++++++++++ tpu_sync/fault_injection/fault_injector.h | 20 ++- .../fault_injection/fault_injector_test.cc | 97 +++++++++++- 4 files changed, 258 insertions(+), 5 deletions(-) diff --git a/tpu_sync/fault_injection/BUILD b/tpu_sync/fault_injection/BUILD index aa53d102d..cb685907e 100644 --- a/tpu_sync/fault_injection/BUILD +++ b/tpu_sync/fault_injection/BUILD @@ -33,6 +33,7 @@ cc_library( "@com_google_absl//absl/base:core_headers", "@com_google_absl//absl/base:no_destructor", "@com_google_absl//absl/container:flat_hash_map", + "@com_google_absl//absl/log", "@com_google_absl//absl/log:absl_check", "@com_google_absl//absl/random", "@com_google_absl//absl/status", @@ -50,7 +51,6 @@ cc_test( features = ["-use_header_modules"], deps = [ ":hooks", - "@com_google_absl//absl/container:flat_hash_set", "@com_google_absl//absl/status", "@com_google_absl//absl/strings", "@com_google_absl//absl/synchronization", diff --git a/tpu_sync/fault_injection/fault_injector.cc b/tpu_sync/fault_injection/fault_injector.cc index 227137113..a77f72ac0 100644 --- a/tpu_sync/fault_injection/fault_injector.cc +++ b/tpu_sync/fault_injection/fault_injector.cc @@ -16,14 +16,20 @@ #include #include +#include #include #include #include #include +#include +#include +#include +#include #include #include #include +#include #include #include @@ -33,8 +39,11 @@ #include "absl/log/absl_check.h" #include "absl/random/random.h" #include "absl/status/status.h" +#include "absl/strings/ascii.h" #include "absl/strings/match.h" +#include "absl/strings/numbers.h" #include "absl/strings/str_cat.h" +#include "absl/strings/str_split.h" #include "absl/strings/strip.h" #include "absl/synchronization/mutex.h" #include "absl/time/clock.h" @@ -59,8 +68,85 @@ absl::Span HooksFor(FaultInjectionType action) { return {}; } +constexpr std::string_view kFaultInjectionFileEnvVar = + "RAIDEN_FAULT_INJECTION_FILE"; + +__attribute__((constructor)) void InitFaultInjectorFromEnvAtLoad() { + if (const char* p = std::getenv(kFaultInjectionFileEnvVar.data()); + p != nullptr && p[0] != '\0') { + (void)GetFaultInjector(); + } +} + +// Plain-text format: one rule per line (`#` comments and blank lines ignored): +// [min_delay_ms [max_delay_ms]] +absl::Status LoadRulesFromText(std::string_view content, + FaultInjector& injector) { + FaultInjectionRules rules; + for (std::string_view line : + absl::StrSplit(content, '\n', absl::SkipEmpty())) { + line = absl::StripAsciiWhitespace(line); + if (line.empty() || line[0] == '#') continue; + std::vector cols = + absl::StrSplit(line, absl::ByAnyChar(" \t,"), absl::SkipEmpty()); + if (cols.size() < 3 || cols.size() > 5) { + return absl::InvalidArgumentError( + absl::StrCat("invalid rule line: ", line)); + } + FaultInjectionRule r; + r.hook = std::string(cols[0]); + if (cols[1] == "delay") { + r.action = FaultInjectionType::kDelay; + } else if (cols[1] == "fail") { + r.action = FaultInjectionType::kFail; + } else { + return absl::InvalidArgumentError( + absl::StrCat("unknown action: ", cols[1])); + } + if (!absl::SimpleAtod(cols[2], &r.probability)) { + return absl::InvalidArgumentError( + absl::StrCat("invalid probability: ", cols[2])); + } + if (cols.size() >= 4 && !absl::SimpleAtoi(cols[3], &r.min_delay_ms)) { + return absl::InvalidArgumentError( + absl::StrCat("invalid min_delay_ms: ", cols[3])); + } + if (cols.size() == 5 && !absl::SimpleAtoi(cols[4], &r.max_delay_ms)) { + return absl::InvalidArgumentError( + absl::StrCat("invalid max_delay_ms: ", cols[4])); + } + rules.push_back(std::move(r)); + } + return injector.Install(rules); +} + +void WriteStatusFile(std::string_view status_path, bool armed, + uint64_t total_hits, + const absl::flat_hash_map& hits) { + std::string out_str = + absl::StrCat("armed=", armed ? 1 : 0, "\ntotal_hits=", total_hits, "\n"); + for (const auto& [hook, count] : hits) { + absl::StrAppend(&out_str, hook, "=", count, "\n"); + } + std::string tmp_path = absl::StrCat(status_path, ".tmp"); + if (std::ofstream out(tmp_path); out) { + out << out_str; + out.close(); + (void)std::rename(tmp_path.c_str(), std::string(status_path).c_str()); + } +} + } // namespace +FaultInjector::FaultInjector() { + if (const char* p = std::getenv(kFaultInjectionFileEnvVar.data()); + p != nullptr && p[0] != '\0') { + StartFileWatcher(p); + } +} + +FaultInjector::~FaultInjector() { StopFileWatcher(); } + bool FaultInjector::IsHookActive(std::string_view hook) const noexcept { if (!HasActiveInjections()) return false; absl::ReaderMutexLock lock(mu_); @@ -181,6 +267,64 @@ absl::flat_hash_map FaultInjector::GetHitCounts() const { return rule_hits_; } +void FaultInjector::StartFileWatcher(std::string_view file_path, + absl::Duration poll_interval) { + StopFileWatcher(); + if (file_path.empty()) return; + { + absl::MutexLock lock(watcher_mu_); + watcher_stopping_ = false; + } + watcher_thread_ = std::thread(&FaultInjector::WatcherLoop, this, + std::string(file_path), poll_interval); +} + +void FaultInjector::StopFileWatcher() { + if (!watcher_thread_.joinable()) return; + { + absl::MutexLock lock(watcher_mu_); + watcher_stopping_ = true; + } + watcher_thread_.join(); +} + +void FaultInjector::WatcherLoop(std::string file_path, + absl::Duration poll_interval) { + const std::string status_path = absl::StrCat(file_path, ".status.", getpid()); + std::string last_content; + bool status_written = false; + bool last_armed = false; + uint64_t last_hits = 0; + + while (true) { + if (std::ifstream in(file_path); in) { + std::string content((std::istreambuf_iterator(in)), {}); + if (content != last_content) { + last_content = std::move(content); + LoadRulesFromText(last_content, *this).IgnoreError(); + } + } else if (!last_content.empty()) { + Install({}).IgnoreError(); + last_content.clear(); + } + + bool armed = HasActiveInjections(); + uint64_t total_hits = GetHitCount(); + if (!status_written || armed != last_armed || total_hits != last_hits) { + WriteStatusFile(status_path, armed, total_hits, GetHitCounts()); + status_written = true; + last_armed = armed; + last_hits = total_hits; + } + + absl::MutexLock lock(watcher_mu_); + if (watcher_mu_.AwaitWithTimeout(absl::Condition(&watcher_stopping_), + poll_interval)) { + break; + } + } +} + void FaultInjector::ExecuteDelay(std::string_view hook) { EvaluateAndSleep(hook); } diff --git a/tpu_sync/fault_injection/fault_injector.h b/tpu_sync/fault_injection/fault_injector.h index 8b173422e..d98e444ce 100644 --- a/tpu_sync/fault_injection/fault_injector.h +++ b/tpu_sync/fault_injection/fault_injector.h @@ -20,6 +20,7 @@ #include #include #include +#include #include #include "absl/base/optimization.h" @@ -28,6 +29,7 @@ #include "absl/random/random.h" #include "absl/status/status.h" #include "absl/synchronization/mutex.h" +#include "absl/time/time.h" #include "tpu_sync/fault_injection/hooks.h" // IWYU pragma: export namespace tpu_raiden { @@ -55,10 +57,10 @@ using FaultInjectionRules = std::vector; class FaultInjector final { public: - FaultInjector() = default; + FaultInjector(); FaultInjector(const FaultInjector&) = delete; FaultInjector& operator=(const FaultInjector&) = delete; - ~FaultInjector() = default; + ~FaultInjector(); // Returns true if any fault injection rules are currently // active. @@ -92,6 +94,13 @@ class FaultInjector final { absl::flat_hash_map GetHitCounts() const ABSL_LOCKS_EXCLUDED(mu_); + // Polls `file_path` every `poll_interval`, arming rules when the file exists, + // disarming when deleted, and writing status to `.status.`. + void StartFileWatcher(std::string_view file_path, + absl::Duration poll_interval = absl::Milliseconds(200)) + ABSL_LOCKS_EXCLUDED(watcher_mu_, mu_); + void StopFileWatcher() ABSL_LOCKS_EXCLUDED(watcher_mu_); + // Slow-path execution helpers invoked only when HasActiveInjections() is // true. A delay always runs to completion. void ExecuteDelay(std::string_view hook) ABSL_LOCKS_EXCLUDED(mu_); @@ -112,8 +121,15 @@ class FaultInjector final { void ExecuteSocket(std::string_view hook, int fd) ABSL_LOCKS_EXCLUDED(mu_); private: + void WatcherLoop(std::string file_path, absl::Duration poll_interval) + ABSL_LOCKS_EXCLUDED(watcher_mu_, mu_); + static inline std::atomic has_active_injections_{false}; + std::thread watcher_thread_; + absl::Mutex watcher_mu_; + bool watcher_stopping_ ABSL_GUARDED_BY(watcher_mu_) = false; + // Evaluates the hook and sleeps for the delay if a delay action is chosen. FaultInjectionAction EvaluateAndSleep(std::string_view hook) ABSL_LOCKS_EXCLUDED(mu_); diff --git a/tpu_sync/fault_injection/fault_injector_test.cc b/tpu_sync/fault_injection/fault_injector_test.cc index 086c347e9..ff7bc012c 100644 --- a/tpu_sync/fault_injection/fault_injector_test.cc +++ b/tpu_sync/fault_injection/fault_injector_test.cc @@ -19,6 +19,10 @@ #include #include +#include +#include +#include +#include #include #include #include @@ -27,6 +31,8 @@ #include #include #include "absl/status/status.h" +#include "absl/strings/match.h" +#include "absl/strings/str_cat.h" #include "absl/synchronization/notification.h" #include "absl/time/clock.h" #include "absl/time/time.h" @@ -63,8 +69,14 @@ void WaitForHit(std::string_view hook) { class FaultInjectorTest : public ::testing::Test { protected: - void SetUp() override { GetFaultInjector().Reset(); } - void TearDown() override { GetFaultInjector().Reset(); } + void SetUp() override { + GetFaultInjector().StopFileWatcher(); + GetFaultInjector().Reset(); + } + void TearDown() override { + GetFaultInjector().StopFileWatcher(); + GetFaultInjector().Reset(); + } }; TEST_F(FaultInjectorTest, DefaultStateIsInactive) { @@ -562,5 +574,86 @@ TEST_F(FaultInjectorTest, FastPathOverheadIsSubNanosecond) { EXPECT_LT(ns_per_check, 5.0) << "ns_per_check=" << ns_per_check; } +bool WaitUntil(const std::function& predicate) { + absl::Time deadline = absl::Now() + absl::Seconds(5); + while (absl::Now() < deadline) { + if (predicate()) return true; + absl::SleepFor(absl::Milliseconds(10)); + } + return predicate(); +} + +std::string ReadTextFile(const std::string& path) { + std::ifstream in(path); + if (!in) return ""; + return std::string((std::istreambuf_iterator(in)), {}); +} + +TEST_F(FaultInjectorTest, FileWatcherArmsDisarmsAndPreservesStatusMetrics) { + const std::string rules_path = + absl::StrCat(::testing::TempDir(), "/raiden_test_faults.txt"); + const std::string status_path = + absl::StrCat(rules_path, ".status.", getpid()); + (void)std::remove(rules_path.c_str()); + (void)std::remove(status_path.c_str()); + + GetFaultInjector().StartFileWatcher(rules_path, absl::Milliseconds(20)); + + // Create rules file to arm kTestHookAlpha. + { + std::ofstream out(rules_path); + out << "transfer_recv_session.pull.request fail 1.0\n"; + } + ASSERT_TRUE(WaitUntil([]() { return FaultInjector::HasActiveInjections(); })); + EXPECT_THAT(FaultInjectStatus(kTestHookAlpha), + StatusIs(absl::StatusCode::kInternal)); + EXPECT_THAT(FaultInjectStatus(kTestHookAlpha), + StatusIs(absl::StatusCode::kInternal)); + + ASSERT_TRUE(WaitUntil([&]() { + std::string s = ReadTextFile(status_path); + return absl::StrContains(s, "armed=1\n") && + absl::StrContains(s, "total_hits=2\n"); + })); + EXPECT_THAT(ReadTextFile(status_path), + HasSubstr("transfer_recv_session.pull.request=2\n")); + + // Delete rules file to disarm and verify cumulative metrics survive. + ASSERT_EQ(std::remove(rules_path.c_str()), 0); + ASSERT_TRUE( + WaitUntil([]() { return !FaultInjector::HasActiveInjections(); })); + ABSL_EXPECT_OK(FaultInjectStatus(kTestHookAlpha)); + + ASSERT_TRUE(WaitUntil([&]() { + return absl::StrContains(ReadTextFile(status_path), "armed=0\n"); + })); + std::string status_disarmed = ReadTextFile(status_path); + EXPECT_THAT(status_disarmed, HasSubstr("total_hits=2\n")); + EXPECT_THAT(status_disarmed, + HasSubstr("transfer_recv_session.pull.request=2\n")); + + // Recreate rules file and verify metrics accumulate across disarm -> re-arm. + { + std::ofstream out(rules_path); + out << absl::StrCat(kTestHookBeta, " delay 1.0 1 1\n"); + } + ASSERT_TRUE(WaitUntil( + []() { return GetFaultInjector().IsHookActive(kTestHookBeta); })); + FaultInjectDelay(kTestHookBeta); + + ASSERT_TRUE(WaitUntil([&]() { + std::string s = ReadTextFile(status_path); + return absl::StrContains(s, "armed=1\n") && + absl::StrContains(s, "total_hits=3\n"); + })); + std::string status_rearmed = ReadTextFile(status_path); + EXPECT_THAT(status_rearmed, HasSubstr(absl::StrCat(kTestHookAlpha, "=2\n"))); + EXPECT_THAT(status_rearmed, HasSubstr(absl::StrCat(kTestHookBeta, "=1\n"))); + + GetFaultInjector().StopFileWatcher(); + (void)std::remove(rules_path.c_str()); + (void)std::remove(status_path.c_str()); +} + } // namespace } // namespace tpu_raiden