Add faulty file system for io failure injections (#9457)

Summary:
Add faulty file system to allow us to inject failure for low-level file operations in unit and fuzzer test.
It is implemented under velox filesystem. It is wrapper on top of the real file system, it either inject
errors or delegate the file operations to the underlying file system. We extend TempDirectoryPath or
TempFilePath to allow user to specify whether enable fault injection or not in test. If it does, it returns
the file or directory path with faulty file system scheme prefix (faulty:). This allows the velox file system
to open the file through faulty file system and by removing the faulty fs scheme prefix, it can delegate
the file operation to the corresponding underlying file system.

We support three kinds of file fault injections: (1) error injection which throws for a set (or all) file operation
types; (2) delay injection for a set (or all) file operation types; (3) custom injection with user provided fault
injection hook. We define the base structure FaultFileOperation to capture the file fault injection parameters
and each file api will extend with its own operation type.

This PR only supports fault injection for read file operations. Next we will extend to other file operation types
as well as fs operation types.

Pull Request resolved: #9457

Reviewed By: tanjialiang, oerling

Differential Revision: D56079500

Pulled By: xiaoxmeng

velox/common/base/SpillConfig.h

@@ -35,7 +35,7 @@ namespace facebook::velox::common {
 
 /// Defining type for a callback function that returns the spill directory path.
 /// Implementations can use it to ensure the path exists before returning.
-using GetSpillDirectoryPathCB = std::function<const std::string&()>;
+using GetSpillDirectoryPathCB = std::function<const std::string()>;
 
 /// The callback used to update the aggregated spill bytes of a query. If the
 /// query spill limit is set, the callback throws if the aggregated spilled

velox/common/base/tests/FsTest.cpp

@@ -24,7 +24,8 @@ namespace facebook::velox::common {
 class FsTest : public testing::Test {};
 
 TEST_F(FsTest, createDirectory) {
-  auto rootPath = exec::test::TempDirectoryPath::createTempDirectory();
+  auto dir = exec::test::TempDirectoryPath::create();
+  auto rootPath = dir->path();
   auto tmpDirectoryPath = rootPath + "/first/second/third";
   // First time should generate directory successfully.
   EXPECT_FALSE(fs::exists(tmpDirectoryPath.c_str()));
@@ -34,7 +35,7 @@ TEST_F(FsTest, createDirectory) {
   // Directory already exist, not creating but should return success.
   EXPECT_TRUE(generateFileDirectory(tmpDirectoryPath.c_str()));
   EXPECT_TRUE(fs::exists(tmpDirectoryPath.c_str()));
-  boost::filesystem::remove_all(rootPath);
+  dir.reset();
   EXPECT_FALSE(fs::exists(rootPath.c_str()));
 }
 

velox/common/caching/tests/AsyncDataCacheTest.cpp

@@ -106,7 +106,7 @@ class AsyncDataCacheTest : public testing::Test {
         tempDirectory_ = exec::test::TempDirectoryPath::create();
       }
       ssdCache = std::make_unique<SsdCache>(
-          fmt::format("{}/cache", tempDirectory_->path),
+          fmt::format("{}/cache", tempDirectory_->path()),
           ssdBytes,
           4,
           executor(),
@@ -816,7 +816,7 @@ TEST_F(AsyncDataCacheTest, DISABLED_ssd) {
 
   cache_->ssdCache()->clear();
   // We cut the tail off one of the cache shards.
-  corruptFile(fmt::format("{}/cache0.cpt", tempDirectory_->path));
+  corruptFile(fmt::format("{}/cache0.cpt", tempDirectory_->path()));
   // We open the cache from checkpoint. Reading checks the data integrity, here
   // we check that more data was read than written.
   initializeCache(kRamBytes, kSsdBytes);

velox/common/caching/tests/SsdFileTest.cpp

@@ -67,7 +67,7 @@ class SsdFileTest : public testing::Test {
 
     tempDirectory_ = exec::test::TempDirectoryPath::create();
     ssdFile_ = std::make_unique<SsdFile>(
-        fmt::format("{}/ssdtest", tempDirectory_->path),
+        fmt::format("{}/ssdtest", tempDirectory_->path()),
         0, // shardId
         bits::roundUp(ssdBytes, SsdFile::kRegionSize) / SsdFile::kRegionSize,
         0, // checkpointInternalBytes

velox/common/file/File.cpp

@@ -238,10 +238,8 @@ LocalWriteFile::LocalWriteFile(
   {
     if (shouldThrowOnFileAlreadyExists) {
       FILE* exists = fopen(buf.get(), "rb");
-      VELOX_CHECK(
-          !exists,
-          "Failure in LocalWriteFile: path '{}' already exists.",
-          path);
+      VELOX_CHECK_NULL(
+          exists, "Failure in LocalWriteFile: path '{}' already exists.", path);
     }
   }
   auto* file = fopen(buf.get(), "ab");

velox/common/file/File.h

@@ -225,14 +225,15 @@ class InMemoryWriteFile final : public WriteFile {
   std::string* file_;
 };
 
-// Current implementation for the local version is quite simple (e.g. no
-// internal arenaing), as local disk writes are expected to be cheap. Local
-// files match against any filepath starting with '/'.
-
+/// Current implementation for the local version is quite simple (e.g. no
+/// internal arenaing), as local disk writes are expected to be cheap. Local
+/// files match against any filepath starting with '/'.
 class LocalReadFile final : public ReadFile {
  public:
   explicit LocalReadFile(std::string_view path);
 
+  /// TODO: deprecate this after creating local file all through velox fs
+  /// interface.
   explicit LocalReadFile(int32_t fd);
 
   ~LocalReadFile();

velox/common/file/tests/CMakeLists.txt

@@ -12,7 +12,9 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-add_library(velox_file_test_utils TestUtils.cpp)
+add_library(velox_file_test_utils TestUtils.cpp FaultyFile.cpp
+                                  FaultyFileSystem.cpp)
+
 target_link_libraries(velox_file_test_utils PUBLIC velox_file)
 
 add_executable(velox_file_test FileTest.cpp UtilsTest.cpp)

velox/common/file/tests/FaultyFile.cpp

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#include "velox/common/file/tests/FaultyFile.h"
+
+namespace facebook::velox::tests::utils {
+
+FaultyReadFile::FaultyReadFile(
+    const std::string& path,
+    std::shared_ptr<ReadFile> delegatedFile,
+    FileFaultInjectionHook injectionHook)
+    : path_(path),
+      delegatedFile_(std::move(delegatedFile)),
+      injectionHook_(std::move(injectionHook)) {
+  VELOX_CHECK_NOT_NULL(delegatedFile_);
+}
+
+std::string_view
+FaultyReadFile::pread(uint64_t offset, uint64_t length, void* buf) const {
+  if (injectionHook_ != nullptr) {
+    FaultFileReadOperation op(path_, offset, length, buf);
+    injectionHook_(&op);
+    if (!op.delegate) {
+      return std::string_view(static_cast<char*>(op.buf), op.length);
+    }
+  }
+  return delegatedFile_->pread(offset, length, buf);
+}
+
+uint64_t FaultyReadFile::preadv(
+    uint64_t offset,
+    const std::vector<folly::Range<char*>>& buffers) const {
+  if (injectionHook_ != nullptr) {
+    FaultFileReadvOperation op(path_, offset, buffers);
+    injectionHook_(&op);
+    if (!op.delegate) {
+      return op.readBytes;
+    }
+  }
+  return delegatedFile_->preadv(offset, buffers);
+}
+
+FaultyWriteFile::FaultyWriteFile(
+    std::shared_ptr<WriteFile> delegatedFile,
+    FileFaultInjectionHook injectionHook)
+    : delegatedFile_(std::move(delegatedFile)),
+      injectionHook_(std::move(injectionHook)) {
+  VELOX_CHECK_NOT_NULL(delegatedFile_);
+}
+
+void FaultyWriteFile::append(std::string_view data) {
+  delegatedFile_->append(data);
+}
+
+void FaultyWriteFile::append(std::unique_ptr<folly::IOBuf> data) {
+  delegatedFile_->append(std::move(data));
+}
+
+void FaultyWriteFile::flush() {
+  delegatedFile_->flush();
+}
+
+void FaultyWriteFile::close() {
+  delegatedFile_->close();
+}
+} // namespace facebook::velox::tests::utils

velox/common/file/tests/FaultyFile.h

@@ -0,0 +1,150 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include "velox/common/file/File.h"
+
+namespace facebook::velox::tests::utils {
+
+/// Defines the per-file operation fault injection.
+struct FaultFileOperation {
+  enum class Type {
+    /// Injects faults for file read operations.
+    kRead,
+    kReadv,
+    /// TODO: add to support fault injections for the other file operation
+    /// types.
+  };
+
+  const Type type;
+
+  /// The delegated file path.
+  const std::string path;
+
+  /// Indicates to forward this operation to the delegated file or not. If not,
+  /// then the file fault injection hook must have processed the request. For
+  /// instance, if this is a file read injection, then the hook must have filled
+  /// the fake read data for data corruption tests.
+  bool delegate{true};
+
+  FaultFileOperation(Type _type, const std::string& _path)
+      : type(_type), path(_path) {}
+};
+
+/// Fault injection parameters for file read API.
+struct FaultFileReadOperation : FaultFileOperation {
+  const uint64_t offset;
+  const uint64_t length;
+  void* const buf;
+
+  FaultFileReadOperation(
+      const std::string& _path,
+      uint64_t _offset,
+      uint64_t _length,
+      void* _buf)
+      : FaultFileOperation(FaultFileOperation::Type::kRead, _path),
+        offset(_offset),
+        length(_length),
+        buf(_buf) {}
+};
+
+/// Fault injection parameters for file readv API.
+struct FaultFileReadvOperation : FaultFileOperation {
+  const uint64_t offset;
+  const std::vector<folly::Range<char*>>& buffers;
+  uint64_t readBytes{0};
+
+  FaultFileReadvOperation(
+      const std::string& _path,
+      uint64_t _offset,
+      const std::vector<folly::Range<char*>>& _buffers)
+      : FaultFileOperation(FaultFileOperation::Type::kReadv, _path),
+        offset(_offset),
+        buffers(_buffers) {}
+};
+
+/// The fault injection hook on the file operation path.
+using FileFaultInjectionHook = std::function<void(FaultFileOperation*)>;
+
+class FaultyReadFile : public ReadFile {
+ public:
+  FaultyReadFile(
+      const std::string& path,
+      std::shared_ptr<ReadFile> delegatedFile,
+      FileFaultInjectionHook injectionHook);
+
+  ~FaultyReadFile() override{};
+
+  uint64_t size() const override {
+    return delegatedFile_->size();
+  }
+
+  std::string_view pread(uint64_t offset, uint64_t length, void* buf)
+      const override;
+
+  uint64_t preadv(
+      uint64_t offset,
+      const std::vector<folly::Range<char*>>& buffers) const override;
+
+  uint64_t memoryUsage() const override {
+    return delegatedFile_->memoryUsage();
+  }
+
+  bool shouldCoalesce() const override {
+    return delegatedFile_->shouldCoalesce();
+  }
+
+  std::string getName() const override {
+    return delegatedFile_->getName();
+  }
+
+  uint64_t getNaturalReadSize() const override {
+    return delegatedFile_->getNaturalReadSize();
+  }
+
+ private:
+  const std::string path_;
+  const std::shared_ptr<ReadFile> delegatedFile_;
+  const FileFaultInjectionHook injectionHook_;
+};
+
+class FaultyWriteFile : public WriteFile {
+ public:
+  FaultyWriteFile(
+      std::shared_ptr<WriteFile> delegatedFile,
+      FileFaultInjectionHook injectionHook);
+
+  ~FaultyWriteFile() override{};
+
+  void append(std::string_view data) override;
+
+  void append(std::unique_ptr<folly::IOBuf> data) override;
+
+  void flush() override;
+
+  void close() override;
+
+  uint64_t size() const override {
+    return delegatedFile_->size();
+  }
+
+ private:
+  const std::shared_ptr<WriteFile> delegatedFile_;
+  const FileFaultInjectionHook injectionHook_;
+};
+
+} // namespace facebook::velox::tests::utils