Add remote storage liveness crash test signal (#15102)

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

Add a dedicated Skycastle signal that invokes liveness through the existing remote-storage test-driver path.

For remote DB runs, double default wrapper/no-progress timeouts while preserving explicit overrides. Default remote liveness `expected_values_dir` to local storage so `UniqueIdVerifier` never treats a remote DB path as a local path; local liveness behavior remains unchanged.

Normalize both equals and spaced `--env_uri`/`--fs_uri` forms once so remote detection, workload execution, and cleanup share the same arguments. Track the crash harness and internal release scripts as CI source triggers, scoped so release-script changes select only the remote-storage signal. Add command-level regression coverage and failure-safe expected-values cleanup. Use generic remote DB/storage terminology throughout public RocksDB code and comments.

Reviewed By: pdillinger

Differential Revision: D115936164

fbshipit-source-id: e4997bf19eec18e4914626d479690b78c6cd3ad2

crash_test.mk

@@ -25,6 +25,7 @@ CRASHTEST_PY=$(PYTHON) -u tools/db_crashtest.py --stress_cmd=$(DB_STRESS_CMD) --
 	crash_test_with_multiops_wup_txn \
 	crash_test_with_optimistic_txn \
 	crash_test_with_tiered_storage \
+	liveness_crash_test \
 	blackbox_crash_test blackbox_crash_test_with_atomic_flush \
 	blackbox_crash_test_with_wc_txn blackbox_crash_test_with_wp_txn \
 	blackbox_crash_test_with_wup_txn \
@@ -94,6 +95,9 @@ crash_test_with_multiops_wp_txn: $(DB_STRESS_CMD)
 crash_test_with_multiops_wup_txn: $(DB_STRESS_CMD)
 	$(CRASHTEST_MAKE) blackbox_crash_test_with_multiops_wup_txn
 
+liveness_crash_test: $(DB_STRESS_CMD)
+	$(CRASHTEST_PY) liveness $(CRASH_TEST_EXT_ARGS)
+
 blackbox_crash_test: $(DB_STRESS_CMD)
 	$(CRASHTEST_PY) --simple blackbox $(CRASH_TEST_EXT_ARGS)
 	$(CRASHTEST_PY) blackbox $(CRASH_TEST_EXT_ARGS)

db/deletefile_test.cc

@@ -262,8 +262,8 @@ TEST_F(DeleteFileTest, BackgroundPurgeIteratorTest) {
 // Regression test for a use-after-free where an obsolete-file purge started
 // DURING DB close -- after CloseHelper's early purge drain -- continued using
 // DBImpl after ~DBImpl destroyed mutex_. Seen as a vhost-user-blk SIGABRT when
-// many RocksDB instances closed concurrently under a Warm Storage fault, with
-// avoid_unnecessary_blocking_io=true on a shared, process-wide Env.
+// many RocksDB instances closed concurrently under a remote file-system fault
+// while using avoid_unnecessary_blocking_io=true on a shared, process-wide Env.
 //
 // Here we make the straggler deterministic: keep a dropped column family handle
 // alive, delete it only after CloseHelper has passed its early purge drain,

tools/db_crashtest.py

@@ -44,7 +44,10 @@ _TSAN_SUPPRESSIONS_FILE = os.path.abspath(
     os.path.join(os.path.dirname(__file__), "tsan_suppressions.txt")
 )
 DEFAULT_LIVENESS_TIMEOUT_SEC = 3600
+DEFAULT_LIVENESS_NO_PROGRESS_TIMEOUT_SEC = 300
+REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER = 2
 _FAULT_INJECTION_LOG_DIR_NAME = "fault_injection_logs"
+_REMOTE_DB_URI_FLAGS = ("--env_uri", "--fs_uri")
 
 
 def get_random_seed(override):
@@ -73,6 +76,43 @@ def stress_cmd_env():
     return env
 
 
+def normalize_remote_db_args(args):
+    """Normalize supported remote URI flags to --flag=value form."""
+    normalized_args = []
+    index = 0
+    while index < len(args):
+        arg = args[index]
+        if (
+            arg in _REMOTE_DB_URI_FLAGS
+            and index + 1 < len(args)
+            and args[index + 1]
+            and not args[index + 1].startswith("-")
+        ):
+            normalized_args.append(arg + "=" + args[index + 1])
+            index += 2
+            continue
+        normalized_args.append(arg)
+        index += 1
+    return normalized_args
+
+
+def parse_remote_db_uri_arg(arg):
+    flag, separator, value = arg.partition("=")
+    if flag not in _REMOTE_DB_URI_FLAGS or not separator:
+        return None
+    return flag, value
+
+
+def remote_db_enabled(args):
+    effective_values = {}
+    for arg in args:
+        parsed_arg = parse_remote_db_uri_arg(arg)
+        if parsed_arg is not None:
+            flag, value = parsed_arg
+            effective_values[flag] = value
+    return any(effective_values.values())
+
+
 def early_argument_parsing_before_main():
     parser = argparse.ArgumentParser()
     parser.add_argument(
@@ -93,16 +133,14 @@ def early_argument_parsing_before_main():
 
     global remain_args
     args, remain_args = parser.parse_known_args()
+    remain_args = normalize_remote_db_args(remain_args)
     init_random_seed = get_random_seed(args.initial_random_seed_override)
     global per_iteration_random_seed_override
     per_iteration_random_seed_override = args.per_iteration_random_seed_override
     global is_remote_db
-    # Set is_remote_db if remain_args has a non-empty --env_uri= or --fs_uri= argument
-    for arg in remain_args:
-        parts = arg.split("=", 1)
-        if parts[0] in ["--env_uri", "--fs_uri"] and len(parts) > 1 and parts[1]:
-            is_remote_db = True
-            break
+    # Remote URI arguments are normalized above so all later consumers see
+    # the same representation.
+    is_remote_db = remote_db_enabled(remain_args)
 
     print(f"Start with random seed {init_random_seed}")
     random.seed(init_random_seed)
@@ -699,12 +737,34 @@ liveness_default_params = {
     "duration": DEFAULT_LIVENESS_TIMEOUT_SEC,
     "enable_thread_tracking": 1,
     "liveness_check_interval_sec": 1,
-    "liveness_no_progress_timeout_sec": 300,
+    "liveness_no_progress_timeout_sec": DEFAULT_LIVENESS_NO_PROGRESS_TIMEOUT_SEC,
     "ops_per_thread": 100000000,
     "progress_reports": 1,
 }
 liveness_default_params.update(liveness_fault_injection_params)
 
+
+def liveness_default_timeout_sec():
+    multiplier = REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER if is_remote_db else 1
+    return DEFAULT_LIVENESS_TIMEOUT_SEC * multiplier
+
+
+def apply_liveness_remote_defaults(params, args):
+    if not is_remote_db:
+        return
+
+    # Keep the wrapper duration in the same ratio as the no-progress timeout so
+    # the longer remote DB timeout still has enough run time to exercise
+    # liveness.
+    if getattr(args, "duration", None) is None:
+        params["duration"] = liveness_default_timeout_sec()
+    if getattr(args, "liveness_no_progress_timeout_sec", None) is None:
+        params["liveness_no_progress_timeout_sec"] = (
+            DEFAULT_LIVENESS_NO_PROGRESS_TIMEOUT_SEC
+            * REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER
+        )
+
+
 simple_default_params = {
     "allow_concurrent_memtable_write": lambda: random.randint(0, 1),
     "column_families": 1,
@@ -1033,10 +1093,10 @@ def finalize_and_sanitize(src_params):
 
     # Blob direct write requires concurrent read visibility of files still open
     # for writing (BlobFileReader calls GetFileSize() on active partition files).
-    # Remote file systems such as Warm Storage do not guarantee that writes from
-    # a WritableFile are visible to a separate RandomAccessFile until the writer
-    # is closed, causing "Malformed blob file" corruption.  Disable BDW when a
-    # remote --env_uri / --fs_uri is in use.
+    # Some remote file systems do not guarantee that writes from a WritableFile
+    # are visible to a separate RandomAccessFile until the writer is closed,
+    # causing "Malformed blob file" corruption. Disable BDW when a remote
+    # --env_uri / --fs_uri is in use.
     if is_remote_db:
         dest_params["enable_blob_direct_write"] = 0
 
@@ -1736,6 +1796,7 @@ def gen_cmd_params(args):
         if v is not None:
             params[k] = v
     if args.test_type == "liveness":
+        apply_liveness_remote_defaults(params, args)
         params.update(liveness_fault_injection_params)
     return params
 
@@ -2152,8 +2213,7 @@ def cleanup_after_success(db_arg, num_dbs=1):
     ]
     # Pass through relevant arguments for remote DB access
     for arg in remain_args:
-        parts = arg.split("=", 1)
-        if parts[0] in ["--env_uri", "--fs_uri"]:
+        if parse_remote_db_uri_arg(arg) is not None:
             cleanup_cmd_parts.append(arg)
     print("Running DB cleanup command - %s\n" % " ".join(cleanup_cmd_parts))
     ret = subprocess.call(cleanup_cmd_parts, env=stress_cmd_env())
@@ -2202,7 +2262,7 @@ def print_fault_injection_log(pid):
 def liveness_timeout(cmd_params):
     duration = cmd_params.get("duration", 0)
     if duration is None or duration <= 0:
-        return DEFAULT_LIVENESS_TIMEOUT_SEC
+        return liveness_default_timeout_sec()
     return duration
 
 
@@ -2213,6 +2273,8 @@ def liveness_main(args, unknown_args):
 
     if not cmd_params.get("db"):
         cmd_params["db"] = db_parent_dir
+    if is_remote_db and not cmd_params.get("expected_values_dir"):
+        cmd_params["expected_values_dir"] = get_ev_parent_dir()
 
     apply_random_seed_per_iteration()
     wrapper_timeout = liveness_timeout(cmd_params)

tools/db_crashtest_test.py

@@ -13,6 +13,7 @@ import tempfile
 import unittest
 from contextlib import redirect_stdout
 from types import SimpleNamespace
+from unittest import mock
 
 
 _DB_CRASHTEST_PATH = os.path.join(os.path.dirname(__file__), "db_crashtest.py")
@@ -24,14 +25,14 @@ _TEST_EXPECTED_DIR_ENV_VAR = "TEST_TMPDIR_EXPECTED"
 _TSAN_OPTIONS_ENV_VAR = "TSAN_OPTIONS"
 
 
-def load_db_crashtest_module():
+def load_db_crashtest_module(args=None):
     spec = importlib.util.spec_from_file_location(
         "db_crashtest_under_test", _DB_CRASHTEST_PATH
     )
     module = importlib.util.module_from_spec(spec)
     old_argv = sys.argv[:]
     try:
-        sys.argv = [_DB_CRASHTEST_PATH]
+        sys.argv = [_DB_CRASHTEST_PATH] + list(args or [])
         spec.loader.exec_module(module)
     finally:
         sys.argv = old_argv
@@ -77,8 +78,18 @@ class DBCrashTestTest(unittest.TestCase):
 
         shutil.rmtree(self.test_tmpdir)
 
-    def load_db_crashtest(self):
-        return load_db_crashtest_module()
+    def load_db_crashtest(self, args=None):
+        return load_db_crashtest_module(args)
+
+    def register_expected_values_dir_cleanup(self, db_crashtest):
+        def cleanup_expected_values_dir():
+            if db_crashtest.ev_parent_dir_global:
+                shutil.rmtree(
+                    db_crashtest.ev_parent_dir_global,
+                    ignore_errors=True,
+                )
+
+        self.addCleanup(cleanup_expected_values_dir)
 
     def load_fault_injection_log_parser(self):
         return load_fault_injection_log_parser_module()
@@ -540,6 +551,33 @@ class DBCrashTestTest(unittest.TestCase):
         )
         self.assertEqual(30, db_crashtest.liveness_timeout({"duration": 30}))
 
+    def test_liveness_remote_db_uses_longer_timeouts(self):
+        db_crashtest = self.load_db_crashtest()
+        db_crashtest.is_remote_db = True
+
+        params = db_crashtest.gen_cmd_params(self.build_mode_args())
+
+        self.assertEqual(
+            db_crashtest.DEFAULT_LIVENESS_TIMEOUT_SEC
+            * db_crashtest.REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER,
+            params["duration"],
+        )
+        self.assertEqual(
+            db_crashtest.DEFAULT_LIVENESS_NO_PROGRESS_TIMEOUT_SEC
+            * db_crashtest.REMOTE_DB_LIVENESS_TIMEOUT_MULTIPLIER,
+            params["liveness_no_progress_timeout_sec"],
+        )
+        self.assertEqual(
+            params["duration"], db_crashtest.liveness_timeout({"duration": 0})
+        )
+
+        params = db_crashtest.gen_cmd_params(
+            self.build_mode_args(duration=17, liveness_no_progress_timeout_sec=7)
+        )
+
+        self.assertEqual(17, params["duration"])
+        self.assertEqual(7, params["liveness_no_progress_timeout_sec"])
+
     def test_liveness_wrapper_timeout_is_successful_end_of_run(self):
         db_crashtest = self.load_db_crashtest()
         execute_calls = []
@@ -567,8 +605,174 @@ class DBCrashTestTest(unittest.TestCase):
         self.assertEqual(17, execute_calls[0][1])
         self.assertFalse(execute_calls[0][2])
         self.assertFalse(execute_calls[0][3])
+        self.assertIn("--expected_values_dir=", execute_calls[0][0])
         self.assertEqual(1, len(cleanups))
 
+    def test_remote_liveness_uses_local_expected_values_dir(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.register_expected_values_dir_cleanup(db_crashtest)
+        execute_calls = []
+
+        def fake_execute_cmd(
+            cmd, timeout=None, timeout_pstack=False, expected_to_timeout=True
+        ):
+            execute_calls.append(cmd)
+            return (
+                True,
+                -15,
+                "Received signal 15 (Terminated)\n",
+                "",
+                123,
+            )
+
+        def expected_values_dir_from(cmd):
+            values = [
+                arg.split("=", 1)[1]
+                for arg in cmd
+                if arg.startswith("--expected_values_dir=")
+            ]
+            self.assertEqual(1, len(values))
+            return values[0]
+
+        db_crashtest.execute_cmd = fake_execute_cmd
+        db_crashtest.print_fault_injection_log = lambda pid: None
+        db_crashtest.cleanup_after_success = lambda db_arg, num_dbs=1: None
+
+        db_crashtest.liveness_main(self.build_mode_args(duration=17), [])
+
+        self.assertEqual(1, len(execute_calls))
+        expected_values_dir = expected_values_dir_from(execute_calls[0])
+        self.assertTrue(expected_values_dir)
+        self.assertEqual(
+            os.path.realpath(tempfile.gettempdir()),
+            os.path.dirname(os.path.realpath(expected_values_dir)),
+        )
+        self.assertTrue(os.path.isdir(expected_values_dir))
+
+        explicit_dir = os.path.join(self.test_tmpdir, "explicit_expected")
+        db_crashtest.get_ev_parent_dir = lambda: self.fail(
+            "explicit expected_values_dir should not allocate a default"
+        )
+        db_crashtest.liveness_main(
+            self.build_mode_args(duration=17, expected_values_dir=explicit_dir), []
+        )
+
+        self.assertEqual(2, len(execute_calls))
+        self.assertEqual(explicit_dir, expected_values_dir_from(execute_calls[1]))
+        self.assertTrue(os.path.isdir(explicit_dir))
+
+    def test_remote_uri_forms_are_normalized_for_workload_and_cleanup(self):
+        os.environ[_TEST_DIR_ENV_VAR] = "/dev_test/rocksdb_crash_test/job123"
+        uri = "remote://example/db"
+
+        for flag in ("--env_uri", "--fs_uri"):
+            for form in ("equals", "spaced"):
+                with self.subTest(flag=flag, form=form):
+                    remote_args = (
+                        [flag + "=" + uri] if form == "equals" else [flag, uri]
+                    )
+                    db_crashtest = self.load_db_crashtest(
+                        [
+                            "liveness",
+                            "--duration=17",
+                            "--stress_cmd=/bin/db_stress",
+                        ]
+                        + remote_args
+                    )
+                    self.register_expected_values_dir_cleanup(db_crashtest)
+                    workload_calls = []
+                    cleanup_calls = []
+                    expected_values_dirs = []
+
+                    def fake_execute_cmd(
+                        cmd,
+                        timeout=None,
+                        timeout_pstack=False,
+                        expected_to_timeout=True,
+                    ):
+                        workload_calls.append(cmd)
+                        expected_values_dirs.extend(
+                            arg.split("=", 1)[1]
+                            for arg in cmd
+                            if arg.startswith("--expected_values_dir=")
+                        )
+                        return (
+                            True,
+                            -15,
+                            "Received signal 15 (Terminated)\n",
+                            "",
+                            123,
+                        )
+
+                    def fake_cleanup_call(cmd, env=None):
+                        cleanup_calls.append(cmd)
+                        return 0
+
+                    db_crashtest.execute_cmd = fake_execute_cmd
+                    db_crashtest.print_fault_injection_log = lambda pid: None
+                    with mock.patch.object(
+                        db_crashtest.subprocess,
+                        "call",
+                        side_effect=fake_cleanup_call,
+                    ):
+                        db_crashtest.main()
+
+                    normalized_arg = flag + "=" + uri
+                    self.assertTrue(db_crashtest.is_remote_db)
+                    self.assertEqual(1, db_crashtest.remain_args.count(normalized_arg))
+                    self.assertEqual(1, len(workload_calls))
+                    self.assertEqual(1, workload_calls[0].count(normalized_arg))
+                    self.assertNotIn(flag, workload_calls[0])
+                    self.assertEqual(1, len(cleanup_calls))
+                    self.assertEqual(1, cleanup_calls[0].count(normalized_arg))
+                    self.assertNotIn(flag, cleanup_calls[0])
+                    self.assertEqual(1, len(expected_values_dirs))
+                    self.assertTrue(expected_values_dirs[0])
+                    self.assertEqual(
+                        os.path.realpath(tempfile.gettempdir()),
+                        os.path.dirname(os.path.realpath(expected_values_dirs[0])),
+                    )
+
+    def test_remote_uri_missing_value_does_not_consume_following_option(self):
+        db_crashtest = self.load_db_crashtest()
+
+        for flag in ("--env_uri", "--fs_uri"):
+            for following_option in ("-x", "--duration=17"):
+                with self.subTest(flag=flag, following_option=following_option):
+                    args = [flag, following_option]
+                    self.assertEqual(
+                        args,
+                        db_crashtest.normalize_remote_db_args(args),
+                    )
+                    self.assertFalse(db_crashtest.remote_db_enabled(args))
+
+    def test_remote_uri_last_value_wins_and_cleanup_preserves_order(self):
+        cases = (
+            (["--env_uri=first", "--env_uri="], False),
+            (["--env_uri=", "--env_uri=last"], True),
+            (["--env_uri=first", "--fs_uri="], True),
+            (["--env_uri=", "--fs_uri="], False),
+        )
+
+        for remote_args, expected_remote in cases:
+            with self.subTest(remote_args=remote_args):
+                db_crashtest = self.load_db_crashtest(
+                    ["liveness"] + remote_args
+                )
+                db_crashtest.stress_cmd = "/bin/db_stress"
+
+                self.assertEqual(expected_remote, db_crashtest.is_remote_db)
+                self.assertEqual(remote_args, db_crashtest.remain_args[1:])
+                with mock.patch.object(
+                    db_crashtest.subprocess,
+                    "call",
+                    return_value=0,
+                ) as cleanup_call:
+                    db_crashtest.cleanup_after_success("/remote/db")
+                self.assertEqual(remote_args, cleanup_call.call_args.args[0][4:])
+
     def test_liveness_can_target_transaction_lock_manager(self):
         db_crashtest = self.load_db_crashtest()
 

utilities/checkpoint/checkpoint_impl.cc

@@ -163,8 +163,7 @@ class SerialFileMover : public CheckpointFileMover {
 
 // Enqueues link/copy work on the CopyEngine pool; Finish() awaits it and does
 // the link->copy fallback for FileSystems that cannot hard-link (which costs an
-// extra round-trip per file; the intended warm-storage targets support
-// linking).
+// extra round-trip per file; the intended remote file systems support linking).
 class ParallelFileMover : public CheckpointFileMover {
  public:
   ParallelFileMover(CopyEngine* engine, Env* env, bool use_link, bool use_fsync,