Merge 'scylla-sstable: add scrub operation' from Botond Dénes

Exposing scrub compaction to the command-line. Allows for offline scrub of sstables, in cases where online scrubbing (via scylla itself) is not possible or not desired. One such case recently was an sstable from a backup which turned out to be corrupt, `nodetool refresh --load-and-stream` refusing to load it.

Fixes: #14203

Closes #14260

* github.com:scylladb/scylladb:
  docs/operating-scylla/admin-tools: scylla-sstable: document scrub operation
  test/cql-pytest: test_tools.py: add test for scylla sstable scrub
  tools/scylla-sstable: add scrub operation
  tools/scylla-sstable: write operation: add none to valid validation levels
  tools/scylla-sstable: handle errors thrown by the operation
  test/cql-pytest: add option to omit scylla's output from the test output
  tools/scylla-sstable: s/option/operation_option/
  tool/scylla-sstable: add missing comments

docs/operating-scylla/admin-tools/scylla-sstable.rst

@@ -512,6 +512,8 @@ The content is dumped in JSON, using the following schema:
         "above_threshold": Uint
     }
 
+.. _scylla-sstable-validate-operation:
+
 validate
 ^^^^^^^^
 
@@ -527,6 +529,25 @@ The following things are validated:
 
 Any errors found will be logged with error level to ``stderr``.
 
+scrub
+^^^^^
+
+Rewrites the SStable, skipping or fixing corrupt parts. Not all kinds of corruption can be skipped or fixed by scrub.
+It is limited to ordering issues on the partition, row, or mutation-fragment level. See `sstable content <scylla-sstable-sstable-content_>`_ for more details.
+
+Scrub has several modes:
+
+* **abort** - Aborts the scrub as soon as any error is found (recognized or not). This mode is only included for the sake of completeness. We recommend using the **validate** mode so that all errors are reported.
+* **skip** - Skips over any corruptions found, thus omitting them from the output. Note that this mode can result in omitting more than is strictly necessary, but it guarantees that all detectable corruptions will be omitted.
+* **segregate** - Fixes partition/row/mutation-fragment out-of-order errors by segregating the output into as many SStables as required so that the content of each output SStable is properly ordered.
+* **validate** - Validates the content of the SStable, reporting any corruptions found. Writes no output SStables. In this mode, scrub has the same outcome as the `validate operation <scylla-sstable-validate-operation_>`_ - and the validate operation is recommended over scrub.
+
+Output SStables are written to the directory specified via ``--output-directory``. They will be written with the ``BIG`` format and the highest supported SStable format, with generations chosen by scylla-sstable. Generations are chosen such
+that they are unique among the SStables written by the current scrub.
+
+The output directory must be empty; otherwise, scylla-sstable will abort scrub. You can allow writing to a non-empty directory by setting the ``--unsafe-accept-nonempty-output-dir`` command line flag.
+Note that scrub will be aborted if an SStable cannot be written because its generation clashes with a pre-existing SStable in the output directory.
+
 validate-checksums
 ^^^^^^^^^^^^^^^^^^
 

test/cql-pytest/conftest.py

@@ -36,6 +36,11 @@ def pytest_addoption(parser):
         help='CQL server port to connect to')
     parser.addoption('--ssl', action='store_true',
         help='Connect to CQL via an encrypted TLSv1.2 connection')
+    # Used by the wrapper script only, not by pytest, added here so it appears
+    # in --help output and so that pytest's argparser won't protest against its
+    # presence.
+    parser.addoption('--omit-scylla-output', action='store_true',
+        help='Omit scylla\'s output from the test output')
 
 # "cql" fixture: set up client object for communicating with the CQL API.
 # The host/port combination of the server are determined by the --host and

test/cql-pytest/run

@@ -26,6 +26,12 @@ if '--raft' in sys.argv:
     run_with_raft.orig_cmd = cmd
     cmd = run_with_raft
 
+if "-h" in sys.argv or "--help" in sys.argv:
+    run.run_pytest(sys.path[0], sys.argv)
+    exit(0)
+
+run.omit_scylla_output = "--omit-scylla-output" in sys.argv
+
 pid = run.run_with_temporary_dir(cmd)
 ip = run.pid_to_ip(pid)
 

test/cql-pytest/run.py

@@ -118,10 +118,12 @@ def abort_run_with_dir(pid, tmpdir):
 def abort_run_with_temporary_dir(pid):
     return abort_run_with_dir(pid, pid_to_dir(pid))
 
+omit_scylla_output = False
 summary=''
 run_pytest_pids = set()
 
 def cleanup_all():
+    global omit_scylla_output
     global summary
     global run_with_temporary_dir_pids
     global run_pytest_pids
@@ -136,9 +138,10 @@ def cleanup_all():
             pass
     for pid in run_with_temporary_dir_pids:
         f = abort_run_with_temporary_dir(pid)
-        print('\nSubprocess output:\n')
-        sys.stdout.flush()
-        shutil.copyfileobj(f, sys.stdout.buffer)
+        if not omit_scylla_output:
+            print('\nSubprocess output:\n')
+            sys.stdout.flush()
+            shutil.copyfileobj(f, sys.stdout.buffer)
     scylla_set = set()
     print(summary)
 

test/cql-pytest/test_tools.py

@@ -15,6 +15,7 @@
 import subprocess
 import tempfile
 import random
+import re
 import shutil
 import util
 
@@ -707,3 +708,142 @@ def test_external_dir_autodetect_conf_dir_home_env(self, scylla_path, system_scy
                 ext_sstable,
                 system_scylla_local_reference_dump,
                 env={"SCYLLA_HOME": scylla_home_dir})
+
+
+@pytest.fixture(scope="module")
+def scrub_schema_file():
+    """Create a schema.cql for the scrub tests"""
+    with tempfile.NamedTemporaryFile("w+t") as f:
+        f.write("CREATE TABLE ks.tbl (pk int, ck int, v text, PRIMARY KEY (pk, ck))")
+        f.flush()
+        yield f.name
+
+
+@pytest.fixture(scope="module")
+def scrub_good_sstable(scylla_path, scrub_schema_file):
+    """A good sstable used by the scrub tests."""
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        sst_json_path = os.path.join(tmp_dir, "sst.json")
+        with open(sst_json_path, "w") as f:
+            sst_json = [
+                {
+                    "key": { "raw": "0004000000c8" },
+                    "clustering_elements": [
+                        { "type": "clustering-row", "key": { "raw": "000400000001" }, "columns": { "v": { "is_live": True, "type": "regular", "timestamp": 1686815362417553, "value": "vv" } } }
+                    ]
+                }
+            ]
+            json.dump(sst_json, f)
+        subprocess.check_call([scylla_path, "sstable", "write", "--schema-file", scrub_schema_file, "--output-dir", tmp_dir, "--generation", "1", "--input-file", sst_json_path])
+        ssts = glob.glob(os.path.join(tmp_dir, "*-Data.db"))
+        assert len(ssts) == 1
+        yield ssts[0]
+
+
+@pytest.fixture(scope="module")
+def scrub_bad_sstable(scylla_path, scrub_schema_file):
+    """A bad sstable (out-of-order rows) used by the scrub tests."""
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        sst_json_path = os.path.join(tmp_dir, "sst.json")
+        with open(sst_json_path, "w") as f:
+            # rows are out-of-order
+            sst_json = [
+                {
+                    "key": { "raw": "0004000000c8" },
+                    "clustering_elements": [
+                        { "type": "clustering-row", "key": { "raw": "000400000002" }, "columns": { "v": { "is_live": True, "type": "regular", "timestamp": 1686815362417553, "value": "vv" } } },
+                        { "type": "clustering-row", "key": { "raw": "000400000001" }, "columns": { "v": { "is_live": True, "type": "regular", "timestamp": 1686815362417553, "value": "vv" } } }
+                    ]
+                }
+            ]
+            json.dump(sst_json, f)
+        subprocess.check_call([scylla_path, "sstable", "write", "--schema-file", scrub_schema_file, "--output-dir", tmp_dir, "--generation", "1", "--input-file", sst_json_path, "--validation-level", "none"])
+        ssts = glob.glob(os.path.join(tmp_dir, "*-Data.db"))
+        assert len(ssts) == 1
+        yield ssts[0]
+
+
+def subprocess_check_error(args, pattern):
+    """Invoke scubprocess.run() with the provided args and check that it fails with stderr matching the provided pattern."""
+    res = subprocess.run(args, check=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+    assert res.returncode != 0
+    err = res.stderr.decode('utf-8')
+    assert re.search(pattern, err) is not None
+
+
+def check_scrub_output_dir(sst_dir, num_sstables):
+    assert len(glob.glob(os.path.join(sst_dir, "*-Data.db"))) == num_sstables
+
+
+def test_scrub_no_sstables(scylla_path, scrub_schema_file):
+    subprocess_check_error([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "validate"], "error processing arguments: no sstables specified on the command line")
+
+
+def test_scrub_missing_scrub_mode_cli_arg(scylla_path, scrub_schema_file, scrub_bad_sstable, scrub_good_sstable):
+    subprocess_check_error([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, scrub_good_sstable], "error processing arguments: missing mandatory command-line argument --scrub-mode")
+
+
+def test_scrub_output_dir(scylla_path, scrub_schema_file, scrub_good_sstable):
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        # Empty output directory is accepted.
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "abort", "--output-dir", tmp_dir, scrub_good_sstable])
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        with open(os.path.join(tmp_dir, "dummy.txt"), "w") as f:
+            f.write("dummy")
+            f.flush()
+
+        # Non-empty output directory is rejected.
+        subprocess_check_error([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "abort", "--output-dir", tmp_dir, scrub_good_sstable], "error processing arguments: output-directory is not empty, pass --unsafe-accept-nonempty-output-dir if you are sure you want to write into this directory\n")
+
+        # Validate doesn't write output sstables, so it doesn't care if output dir is non-empty.
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "validate", "--output-dir", tmp_dir, scrub_good_sstable])
+
+        # Check that overriding with --unsafe-accept-nonempty-output-dir works.
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "abort", "--output-dir", tmp_dir, "--unsafe-accept-nonempty-output-dir", scrub_good_sstable])
+
+
+def test_scrub_output_dir_sstable_clash(scylla_path, scrub_schema_file, scrub_good_sstable):
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "abort", "--output-dir", tmp_dir, "--unsafe-accept-nonempty-output-dir", scrub_good_sstable])
+        check_scrub_output_dir(tmp_dir, 1)
+        subprocess_check_error([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "abort", "--output-dir", tmp_dir, "--unsafe-accept-nonempty-output-dir", scrub_good_sstable], "cannot create output sstable .*, file already exists")
+
+
+def test_scrub_abort_mode(scylla_path, scrub_schema_file, scrub_good_sstable, scrub_bad_sstable):
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "abort", "--output-dir", tmp_dir, scrub_good_sstable])
+        check_scrub_output_dir(tmp_dir, 1)
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        subprocess_check_error([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "abort", "--output-dir", tmp_dir, scrub_bad_sstable], "compaction_aborted_exception \\(Compaction for ks/tbl was aborted due to: scrub compaction found invalid data\\)")
+        check_scrub_output_dir(tmp_dir, 0)
+
+
+def test_scrub_skip_mode(scylla_path, scrub_schema_file, scrub_good_sstable, scrub_bad_sstable):
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "skip", "--output-dir", tmp_dir, scrub_good_sstable])
+        check_scrub_output_dir(tmp_dir, 1)
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "skip", "--output-dir", tmp_dir, scrub_bad_sstable])
+        check_scrub_output_dir(tmp_dir, 1)
+
+
+def test_scrub_segregate_mode(scylla_path, scrub_schema_file, scrub_good_sstable, scrub_bad_sstable):
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "segregate", "--output-dir", tmp_dir, scrub_good_sstable])
+        check_scrub_output_dir(tmp_dir, 1)
+
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "segregate", "--output-dir", tmp_dir, scrub_bad_sstable])
+        check_scrub_output_dir(tmp_dir, 2)
+
+
+def test_scrub_validate_mode(scylla_path, scrub_schema_file, scrub_good_sstable, scrub_bad_sstable):
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "validate", "--output-dir", tmp_dir, scrub_good_sstable])
+        check_scrub_output_dir(tmp_dir, 0)
+
+        subprocess.check_call([scylla_path, "sstable", "scrub", "--schema-file", scrub_schema_file, "--scrub-mode", "validate", "--output-dir", tmp_dir, scrub_bad_sstable])
+        check_scrub_output_dir(tmp_dir, 0)