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,14 @@ 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")
+    elif os.environ.get("USE_CAS_STORAGE_FOR_MERGE_TREE") == "1":
+        options.append("--cas-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

@@ -142,7 +142,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(

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,12 @@ start_server 10 || { echo "Failed to start server"; exit 1; }
 
 check_server_start
 
+# clickhouse-local cannot claim a CAS server-root (its uuid is all-zero), so
+# dump system logs via the still-running server that owns the root.
+if [[ "$USE_CAS_S3_STORAGE_FOR_MERGE_TREE" == "1" || "$USE_CAS_STORAGE_FOR_MERGE_TREE" == "1" ]]; then
+    collect_query_and_trace_logs --client
+fi
+
 stop_server
 
 [ -f /var/log/clickhouse-server/clickhouse-server.log ] || echo -e "Server log does not exist\tFAIL"
@@ -327,6 +341,14 @@ check_logs_for_critical_errors
 
 tar -chf /test_output/coordination.tar /var/lib/clickhouse/coordination ||:
 
-collect_query_and_trace_logs
+if [[ "$USE_CAS_S3_STORAGE_FOR_MERGE_TREE" != "1" && "$USE_CAS_STORAGE_FOR_MERGE_TREE" != "1" ]]; then
+    collect_query_and_trace_logs
+fi
 
 mv /var/log/clickhouse-server/stderr.log /test_output/
+
+# The job uploads /test_output only, and rustfs.log is the sole record of
+# pool-side S3 failures (412 mismatches, 503 saturation) behind a CAS disk.
+if [[ -f /repo/ci/tmp/rustfs.log ]]; then
+    zstd --threads=0 -o /test_output/rustfs.log.zst /repo/ci/tmp/rustfs.log ||:
+fi

tests/docker_scripts/stress_tests.lib

@@ -350,10 +350,20 @@ function check_logs_for_critical_errors()
 
 function collect_query_and_trace_logs()
 {
+    # `--client`: dump via the running server. Required on a CAS disk because
+    # clickhouse-local starts with uuid 0 and refuses to claim the server-root.
+    local use_client=0
+    if [[ "${1:-}" == "--client" ]]; then
+        use_client=1
+    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
-        clickhouse-local --config-file=/etc/clickhouse-server/config.xml --only-system-tables -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.tsv.zst
+        if [[ "$use_client" == "1" ]]; then
+            clickhouse-client -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.tsv.zst
+        else
+            clickhouse-local --config-file=/etc/clickhouse-server/config.xml --only-system-tables -q "select * from system.$table format TSVWithNamesAndTypes" | zstd --threads=0 > /test_output/$table.tsv.zst
+        fi
     done
 }