db_stress: db_stress fails on custom filesystems. (#9352)

Summary:
db_stress listener service always uses default filesystem to operate,
causing it to not recognize custom filesystem (like ZenFS plugin FS).
Pass the env to db_stress listener with the correct filesystem
information, so it can open the user intended filesystem.

Signed-off-by: Aravind Ramesh <Aravind.Ramesh@wdc.com>

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

Reviewed By: riversand963

Differential Revision: D33776762

Pulled By: pdillinger

fbshipit-source-id: e79bb9a544384f80ae9dd0108241ab9c83223954

db_stress_tool/db_stress_common.cc

@@ -16,6 +16,7 @@
 #include "util/file_checksum_helper.h"
 #include "util/xxhash.h"
 
+ROCKSDB_NAMESPACE::Env* db_stress_listener_env = nullptr;
 ROCKSDB_NAMESPACE::Env* db_stress_env = nullptr;
 #ifndef NDEBUG
 // If non-null, injects read error at a rate specified by the

db_stress_tool/db_stress_common.h

@@ -277,6 +277,7 @@ constexpr int kValueMaxLen = 100;
 
 // wrapped posix environment
 extern ROCKSDB_NAMESPACE::Env* db_stress_env;
+extern ROCKSDB_NAMESPACE::Env* db_stress_listener_env;
 #ifndef NDEBUG
 namespace ROCKSDB_NAMESPACE {
 class FaultInjectionTestFS;

db_stress_tool/db_stress_listener.cc

@@ -17,7 +17,7 @@ namespace ROCKSDB_NAMESPACE {
 
 // TODO: consider using expected_values_dir instead, but this is more
 // convenient for now.
-UniqueIdVerifier::UniqueIdVerifier(const std::string& db_name)
+UniqueIdVerifier::UniqueIdVerifier(const std::string& db_name, Env* env)
     : path_(db_name + "/.unique_ids") {
   // We expect such a small number of files generated during this test
   // (thousands?), checking full 192-bit IDs for uniqueness is a very
@@ -27,11 +27,10 @@ UniqueIdVerifier::UniqueIdVerifier(const std::string& db_name)
   // very good probability for the quantities in this test.
   offset_ = Random::GetTLSInstance()->Uniform(17);  // 0 to 16
 
-  // Use default FileSystem to avoid fault injection, etc.
+  const std::shared_ptr<FileSystem> fs = env->GetFileSystem();
-  FileSystem& fs = *FileSystem::Default();
   IOOptions opts;
 
-  Status st = fs.CreateDirIfMissing(db_name, opts, nullptr);
+  Status st = fs->CreateDirIfMissing(db_name, opts, nullptr);
   if (!st.ok()) {
     fprintf(stderr, "Failed to create directory %s: %s\n", db_name.c_str(),
             st.ToString().c_str());
@@ -41,7 +40,7 @@ UniqueIdVerifier::UniqueIdVerifier(const std::string& db_name)
   {
     std::unique_ptr<FSSequentialFile> reader;
     Status s =
-        fs.NewSequentialFile(path_, FileOptions(), &reader, /*dbg*/ nullptr);
+        fs->NewSequentialFile(path_, FileOptions(), &reader, /*dbg*/ nullptr);
     if (s.ok()) {
       // Load from file
       std::string id(24U, '\0');
@@ -61,7 +60,7 @@ UniqueIdVerifier::UniqueIdVerifier(const std::string& db_name)
             fprintf(stdout, "Warning: clearing corrupt unique id file\n");
             id_set_.clear();
             reader.reset();
-            s = fs.DeleteFile(path_, opts, /*dbg*/ nullptr);
+            s = fs->DeleteFile(path_, opts, /*dbg*/ nullptr);
             assert(s.ok());
           }
           break;
@@ -72,7 +71,7 @@ UniqueIdVerifier::UniqueIdVerifier(const std::string& db_name)
       // Newly created is ok.
       // But FileSystem doesn't tell us whether non-existence was the cause of
       // the failure. (Issue #9021)
-      Status s2 = fs.FileExists(path_, opts, /*dbg*/ nullptr);
+      Status s2 = fs->FileExists(path_, opts, /*dbg*/ nullptr);
       if (!s2.IsNotFound()) {
         fprintf(stderr, "Error opening unique id file: %s\n",
                 s.ToString().c_str());
@@ -81,8 +80,8 @@ UniqueIdVerifier::UniqueIdVerifier(const std::string& db_name)
     }
   }
   fprintf(stdout, "(Re-)verified %zu unique IDs\n", id_set_.size());
-  Status s = fs.ReopenWritableFile(path_, FileOptions(), &data_file_writer_,
+  Status s = fs->ReopenWritableFile(path_, FileOptions(), &data_file_writer_,
-                                   /*dbg*/ nullptr);
+                                    /*dbg*/ nullptr);
   if (!s.ok()) {
     fprintf(stderr, "Error opening unique id file for append: %s\n",
             s.ToString().c_str());

db_stress_tool/db_stress_listener.h

@@ -11,6 +11,7 @@
 
 #include "file/filename.h"
 #include "rocksdb/db.h"
+#include "rocksdb/env.h"
 #include "rocksdb/file_system.h"
 #include "rocksdb/listener.h"
 #include "rocksdb/table_properties.h"
@@ -26,7 +27,7 @@ namespace ROCKSDB_NAMESPACE {
 // Verify across process executions that all seen IDs are unique
 class UniqueIdVerifier {
  public:
-  explicit UniqueIdVerifier(const std::string& db_name);
+  explicit UniqueIdVerifier(const std::string& db_name, Env* env);
   ~UniqueIdVerifier();
 
   void Verify(const std::string& id);
@@ -49,12 +50,13 @@ class DbStressListener : public EventListener {
  public:
   DbStressListener(const std::string& db_name,
                    const std::vector<DbPath>& db_paths,
-                   const std::vector<ColumnFamilyDescriptor>& column_families)
+                   const std::vector<ColumnFamilyDescriptor>& column_families,
+                   Env* env)
       : db_name_(db_name),
         db_paths_(db_paths),
         column_families_(column_families),
         num_pending_file_creations_(0),
-        unique_ids_(db_name) {}
+        unique_ids_(db_name, env) {}
 
   const char* Name() const override { return kClassName(); }
   static const char* kClassName() { return "DBStressListener"; }

db_stress_tool/db_stress_test_base.cc

@@ -2551,8 +2551,8 @@ void StressTest::Open() {
     }
     options_.listeners.clear();
 #ifndef ROCKSDB_LITE
-    options_.listeners.emplace_back(
+    options_.listeners.emplace_back(new DbStressListener(
-        new DbStressListener(FLAGS_db, options_.db_paths, cf_descriptors));
+        FLAGS_db, options_.db_paths, cf_descriptors, db_stress_listener_env));
 #endif  // !ROCKSDB_LITE
     options_.create_missing_column_families = true;
     if (!FLAGS_use_txn) {

db_stress_tool/db_stress_tool.cc

@@ -32,6 +32,8 @@ namespace ROCKSDB_NAMESPACE {
 namespace {
 static std::shared_ptr<ROCKSDB_NAMESPACE::Env> env_guard;
 static std::shared_ptr<ROCKSDB_NAMESPACE::DbStressEnvWrapper> env_wrapper_guard;
+static std::shared_ptr<ROCKSDB_NAMESPACE::DbStressEnvWrapper>
+    dbsl_env_wrapper_guard;
 static std::shared_ptr<CompositeEnvWrapper> fault_env_guard;
 }  // namespace
 
@@ -77,6 +79,8 @@ int db_stress_tool(int argc, char** argv) {
             s.ToString().c_str());
     exit(1);
   }
+  dbsl_env_wrapper_guard = std::make_shared<DbStressEnvWrapper>(raw_env);
+  db_stress_listener_env = dbsl_env_wrapper_guard.get();
 
 #ifndef NDEBUG
   if (FLAGS_read_fault_one_in || FLAGS_sync_fault_injection ||