Enable cas s3 stress

ci/defs/job_configs.py

@@ -869,21 +869,41 @@ class JobConfigs:
             runs_on=RunnerLabels.FUNC_TESTER_AMD,
             requires=[ArtifactNames.DEB_AMD_DEBUG],
         ),
+        Job.ParamSet(
+            parameter="amd_debug, cas s3 storage",
+            runs_on=RunnerLabels.FUNC_TESTER_AMD,
+            requires=[ArtifactNames.DEB_AMD_DEBUG],
+        ),
         Job.ParamSet(
             parameter="amd_asan_ubsan",
             runs_on=RunnerLabels.FUNC_TESTER_AMD,
             requires=[ArtifactNames.DEB_AMD_ASAN_UBSAN],
         ),
+        Job.ParamSet(
+            parameter="amd_asan_ubsan, cas s3 storage",
+            runs_on=RunnerLabels.FUNC_TESTER_AMD,
+            requires=[ArtifactNames.DEB_AMD_ASAN_UBSAN],
+        ),
         Job.ParamSet(
             parameter="amd_tsan",
             runs_on=RunnerLabels.FUNC_TESTER_AMD,
             requires=[ArtifactNames.DEB_AMD_TSAN],
         ),
+        Job.ParamSet(
+            parameter="amd_tsan, cas s3 storage",
+            runs_on=RunnerLabels.FUNC_TESTER_AMD,
+            requires=[ArtifactNames.DEB_AMD_TSAN],
+        ),
         Job.ParamSet(
             parameter="amd_msan",
             runs_on=RunnerLabels.FUNC_TESTER_AMD,
             requires=[ArtifactNames.DEB_AMD_MSAN],
         ),
+        Job.ParamSet(
+            parameter="amd_msan, cas s3 storage",
+            runs_on=RunnerLabels.FUNC_TESTER_AMD,
+            requires=[ArtifactNames.DEB_AMD_MSAN],
+        ),
         Job.ParamSet(
             parameter="arm_release",
             runs_on=RunnerLabels.FUNC_TESTER_ARM,

ci/jobs/scripts/clickhouse_proc.py

@@ -1549,6 +1549,8 @@ def set_random_timezone():
             res = ch.start_minio(param)
         elif command == "start_azurite":
             res = ch.start_azurite()
+        elif command == "start_rustfs":
+            res = ch.start_rustfs()
         else:
             raise ValueError(f"Unknown command: {command}")
     except Exception:

ci/jobs/scripts/stress/stress.py

@@ -128,6 +128,12 @@ def get_options(i: int, upgrade_check: bool, encrypted_storage: bool) -> str:
     options = []
     client_options = []
 
+    # Honour the CAS-as-default MergeTree policy so clickhouse-test skips
+    # `no-cas-storage` / `no-object-storage` / `no-s3-storage` tests. Without
+    # this flag those tests still run and fail against a CAS disk.
+    if os.environ.get("USE_CAS_S3_STORAGE_FOR_MERGE_TREE") == "1":
+        options.append("--cas-s3-storage")
+
     if upgrade_check:
         # Disable settings randomization for upgrade checks to prevent test failures caused by missing settings in old version
         options.append("--no-random-settings")

ci/jobs/stress_job.py

@@ -1,6 +1,7 @@
 import csv
 import logging
 import os
+import shutil
 import socket
 import sys
 from pathlib import Path
@@ -142,7 +143,12 @@ def get_additional_envs(info, check_name: str) -> List[str]:
             "AZURE_STORAGE_ACCOUNT_URL=$AZURE_STORAGE_ACCOUNT_URL",
         ])
 
-    if "s3" in check_name:
+    # "cas s3" must win over the plain "s3" substring. Otherwise
+    # `USE_S3_STORAGE_FOR_MERGE_TREE=1` is also set and install.sh's
+    # if/elif chain installs the plain-S3 default policy instead of CAS.
+    if "cas s3" in check_name:
+        result.append("USE_CAS_S3_STORAGE_FOR_MERGE_TREE=1")
+    elif "s3" in check_name:
         result.append("USE_S3_STORAGE_FOR_MERGE_TREE=1")
 
     result.append(
@@ -308,6 +314,14 @@ def run_stress_test(upgrade_check: bool = False) -> None:
 
     Utils.fix_ownership_after_docker(temp_path, docker_image)
 
+    # ci/tmp is not uploaded. Copying here rather than from stress_runner.sh runs
+    # whatever the container's exit code, so a job that aborts early still publishes
+    # rustfs.log -- the only record of what the pool side did behind a CAS disk.
+    for helper_log in ("rustfs.log", "minio.log", "azurite.log", "kafka.log"):
+        src = temp_path / helper_log
+        if src.is_file():
+            shutil.copy(src, result_path / helper_log)
+
     core_files = ClickHouseService.collect_cores(cores_path)
 
     is_oom = False

ci/tests/test_stress_job.py

@@ -35,6 +35,7 @@
 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "../.."))
 
 from ci.jobs.stress_job import (
+    get_additional_envs,
     process_results,
     read_test_results,
     sanitize_test_result_line,
@@ -245,5 +246,35 @@ def test_dpkg_progress_in_info_does_not_split_row(tmp_path):
     assert malformed == []
 
 
+@pytest.mark.parametrize(
+    "check_name",
+    [
+        "Stress test (amd_debug, cas s3 storage)",
+        "Stress test (amd_asan_ubsan, cas s3 storage)",
+        "Stress test (amd_tsan, cas s3 storage)",
+        "Stress test (amd_msan, cas s3 storage)",
+    ],
+)
+def test_cas_s3_env_is_exclusive_of_plain_s3(monkeypatch, check_name):
+    monkeypatch.setattr("ci.jobs.ci_utils.is_extended_run", lambda: False)
+    envs = get_additional_envs(None, check_name)
+    assert "USE_CAS_S3_STORAGE_FOR_MERGE_TREE=1" in envs
+    assert "USE_S3_STORAGE_FOR_MERGE_TREE=1" not in envs
+
+
+def test_plain_s3_env_is_unchanged(monkeypatch):
+    monkeypatch.setattr("ci.jobs.ci_utils.is_extended_run", lambda: False)
+    envs = get_additional_envs(None, "Stress test (arm_asan_ubsan, s3)")
+    assert "USE_S3_STORAGE_FOR_MERGE_TREE=1" in envs
+    assert "USE_CAS_S3_STORAGE_FOR_MERGE_TREE=1" not in envs
+
+
+def test_default_stress_env_has_neither_s3_flag(monkeypatch):
+    monkeypatch.setattr("ci.jobs.ci_utils.is_extended_run", lambda: False)
+    envs = get_additional_envs(None, "Stress test (amd_asan_ubsan)")
+    assert "USE_S3_STORAGE_FOR_MERGE_TREE=1" not in envs
+    assert "USE_CAS_S3_STORAGE_FOR_MERGE_TREE=1" not in envs
+
+
 if __name__ == "__main__":
     sys.exit(pytest.main([__file__, "-v"]))

src/Disks/DiskObjectStorage/MetadataStorages/ContentAddressed/README.md

@@ -190,6 +190,12 @@ not to multipart-capable blob publication.
   (local object storage) and "`cas s3 storage`" run the whole
   stateless suite with `MergeTree` defaulting to a CAS disk. Tests that
   legitimately cannot run there carry the `no-cas-storage` tag.
+- **Stress**: `Stress test (amd_{debug,asan_ubsan,tsan,msan}, cas s3 storage)`
+  run the same `clickhouse-test --stress-tests` loop as the other stress
+  jobs (stateless tests picked at random, several clients in parallel, no
+  result validation) with `cas_s3` as the default `MergeTree` policy
+  and RustFS backing the pool. One job per AMD sanitizer build plus
+  debug (the non-sanitizer stress package).
 - **Soak / chaos**: `utils/ca-soak/` — multi-replica docker-compose
   harnesses (fault proxies, GC sharding variants, AWS S3/GCS backends) and
   adversarial scenarios.

tests/docker_scripts/stress_runner.sh

@@ -52,6 +52,11 @@ cd /repo && python3 /repo/ci/jobs/scripts/clickhouse_proc.py logs_export_config
 
 cd /repo && python3 /repo/ci/jobs/scripts/clickhouse_proc.py start_minio stateless || { echo "Failed to start minio"; exit 1; }
 cd /repo && python3 /repo/ci/jobs/scripts/clickhouse_proc.py start_azurite || { echo "Failed to start azurite"; exit 1; }
+if [[ "$USE_CAS_S3_STORAGE_FOR_MERGE_TREE" == "1" ]]; then
+    # CAS-over-S3 needs RustFS (enforced If-Match deletes). MinIO stays up for
+    # the non-CAS s3 disks that EXPORT_S3_STORAGE_POLICIES still installs.
+    cd /repo && python3 /repo/ci/jobs/scripts/clickhouse_proc.py start_rustfs || { echo "Failed to start rustfs"; exit 1; }
+fi
 
 # Start Redpanda (Kafka-compatible broker) so that Kafka engine tests work and
 # do not leave behind broken StorageKafka tables whose background threads cause
@@ -100,7 +105,10 @@ clickhouse-client --query "SHOW TABLES FROM tpcds"
 clickhouse-client --query "SHOW TABLES FROM tpch"
 clickhouse-client --query "SHOW TABLES FROM test"
 
-if [[ "$USE_S3_STORAGE_FOR_MERGE_TREE" == "1" ]]; then
+if [[ "$USE_CAS_S3_STORAGE_FOR_MERGE_TREE" == "1" ]]; then
+    TEMP_POLICY="cas_s3"
+    echo "Using cas_s3 storage policy"
+elif [[ "$USE_S3_STORAGE_FOR_MERGE_TREE" == "1" ]]; then
     TEMP_POLICY="s3_cache"
 elif [[ "$USE_AZURE_STORAGE_FOR_MERGE_TREE" == "1" ]]; then
     TEMP_POLICY="azure_cache"
@@ -315,6 +323,9 @@ start_server 10 || { echo "Failed to start server"; exit 1; }
 
 check_server_start
 
+# The server may be killed rather than shut down, so don't rely on the shutdown flush.
+clickhouse-client --receive_timeout 30 -q "SYSTEM FLUSH LOGS" ||:
+
 stop_server
 
 [ -f /var/log/clickhouse-server/clickhouse-server.log ] || echo -e "Server log does not exist\tFAIL"

tests/docker_scripts/stress_tests.lib

@@ -350,6 +350,18 @@ function check_logs_for_critical_errors()
 
 function collect_query_and_trace_logs()
 {
+    # A writable open of a CAS disk claims the server-root and fails closed against the
+    # real server's owner uuid; read-only skips the claim, which is all a dump needs.
+    # `--follow-symlinks`: install.sh symlinks these configs. See dump_system_tables.
+    grep -Rl '<metadata_type>cas</metadata_type>' /etc/clickhouse-server/ 2>/dev/null \
+        | xargs -r sed -i --follow-symlinks 's|<metadata_type>cas</metadata_type>|<metadata_type>cas</metadata_type><readonly>true</readonly>|g'
+
+    # A declared but not read-only CAS disk means this scrape is about to die on ownership.
+    if grep -Rlq '<metadata_type>cas</metadata_type>' /etc/clickhouse-server/ 2>/dev/null \
+        && ! grep -Rlq '<metadata_type>cas</metadata_type><readonly>true</readonly>' /etc/clickhouse-server/ 2>/dev/null; then
+        echo "WARNING: a CAS disk is declared but the read-only marker was not inserted -- clickhouse-local will claim server-root ownership and this scrape will fail"
+    fi
+
     for table in query_log trace_log metric_log aggregated_zookeeper_log
     do
         # Don't ignore errors here, it leads to ignore sanitizer reports when running clickhouse-local