test(perf): cover GET fallback metric probes (#5291)

* test(perf): cover GET fallback metric probes

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(perf): prefer format fallback labels

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ilm): remove duplicate manual transition import

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/set_disk/mod.rs

@@ -1648,7 +1648,7 @@ fn classify_get_codec_streaming_object_class(
     if object_info.is_encrypted() {
         return GetCodecStreamingObjectClass::Encrypted;
     }
-    if object_info.is_compressed() {
+    if object_info.is_compressed() || fi.is_compressed() {
         return GetCodecStreamingObjectClass::Compressed;
     }
     if object_info.is_remote() {
@@ -1864,20 +1864,6 @@ fn get_codec_streaming_reader_gate(
         };
     }
 
-    let Ok(min_size) = i64::try_from(get_codec_streaming_min_size()) else {
-        return GetCodecStreamingGate {
-            object_class,
-            decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMinSize),
-            prefer_data_blocks_first_reader_setup: false,
-        };
-    };
-    if object_info.size < min_size {
-        return GetCodecStreamingGate {
-            object_class,
-            decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize),
-            prefer_data_blocks_first_reader_setup: false,
-        };
-    }
     if object_class == GetCodecStreamingObjectClass::Encrypted {
         return GetCodecStreamingGate {
             object_class,
@@ -1915,6 +1901,20 @@ fn get_codec_streaming_reader_gate(
             };
         }
     }
+    let Ok(min_size) = i64::try_from(get_codec_streaming_min_size()) else {
+        return GetCodecStreamingGate {
+            object_class,
+            decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::InvalidMinSize),
+            prefer_data_blocks_first_reader_setup: false,
+        };
+    };
+    if object_info.size < min_size {
+        return GetCodecStreamingGate {
+            object_class,
+            decision: GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::BelowMinSize),
+            prefer_data_blocks_first_reader_setup: false,
+        };
+    }
 
     GetCodecStreamingGate {
         object_class,

crates/ecstore/src/set_disk/read.rs

@@ -5121,7 +5121,13 @@ mod tests {
 
                 let mut compressed_fi = fi.clone();
                 insert_str(&mut compressed_fi.metadata, SUFFIX_COMPRESSION, "lz4".to_string());
-                let compressed = codec_streaming_test_object_info(&compressed_fi);
+                let mut compressed = codec_streaming_test_object_info(&compressed_fi);
+                assert_eq!(
+                    codec_streaming_reader_gate_for_test(&None, &compressed, &compressed_fi, true).decision,
+                    GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed)
+                );
+
+                compressed.user_defined = Arc::default();
                 assert_eq!(
                     codec_streaming_reader_gate_for_test(&None, &compressed, &compressed_fi, true).decision,
                     GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed)
@@ -5145,6 +5151,64 @@ mod tests {
         );
     }
 
+    #[test]
+    fn codec_streaming_format_fallbacks_precede_min_size() {
+        temp_env::with_vars(
+            [
+                (ENV_RUSTFS_GET_CODEC_STREAMING_ENABLE, Some("true")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_ROLLOUT, Some("benchmark")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_BODY_COMPAT_CONFIRMED, Some("true")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_HEADER_COMPAT_CONFIRMED, Some("true")),
+                (ENV_RUSTFS_GET_CODEC_STREAMING_MIN_SIZE, Some("1048576")),
+            ],
+            || {
+                let mut encrypted_fi = codec_streaming_test_fileinfo(16 * 1024, 1);
+                encrypted_fi
+                    .metadata
+                    .insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
+                let encrypted = codec_streaming_test_object_info(&encrypted_fi);
+                assert_eq!(
+                    codec_streaming_reader_gate_for_test(&None, &encrypted, &encrypted_fi, true).decision,
+                    GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted)
+                );
+
+                let mut compressed_fi = codec_streaming_test_fileinfo(16 * 1024, 1);
+                insert_str(&mut compressed_fi.metadata, SUFFIX_COMPRESSION, "lz4".to_string());
+                let compressed = codec_streaming_test_object_info(&compressed_fi);
+
+                assert_eq!(
+                    codec_streaming_reader_gate_for_test(&None, &compressed, &compressed_fi, true).decision,
+                    GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Compressed)
+                );
+
+                let mut encrypted_fi = codec_streaming_test_fileinfo(16 * 1024, 1);
+                encrypted_fi
+                    .metadata
+                    .insert("x-amz-server-side-encryption".to_string(), "AES256".to_string());
+                let encrypted = codec_streaming_test_object_info(&encrypted_fi);
+                assert_eq!(
+                    codec_streaming_reader_gate_for_test(&None, &encrypted, &encrypted_fi, true).decision,
+                    GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Encrypted)
+                );
+
+                let mut remote_fi = codec_streaming_test_fileinfo(16 * 1024, 1);
+                remote_fi.transition_status = TRANSITION_COMPLETE.to_string();
+                let remote = codec_streaming_test_object_info(&remote_fi);
+                assert_eq!(
+                    codec_streaming_reader_gate_for_test(&None, &remote, &remote_fi, true).decision,
+                    GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Remote)
+                );
+
+                let multipart_fi = codec_streaming_test_fileinfo(16 * 1024, 2);
+                let multipart = codec_streaming_test_object_info(&multipart_fi);
+                assert_eq!(
+                    codec_streaming_reader_gate_for_test(&None, &multipart, &multipart_fi, true).decision,
+                    GetCodecStreamingDecision::Fallback(GetCodecStreamingFallbackReason::Multipart)
+                );
+            },
+        );
+    }
+
     #[test]
     fn codec_streaming_engine_defaults_to_legacy_and_parses_rustfs() {
         temp_env::with_var(ENV_RUSTFS_GET_CODEC_STREAMING_ENGINE, None::<&str>, || {

scripts/run_get_1mib_abba_stage_metrics.sh

@@ -47,6 +47,7 @@ DIAGNOSTIC_OBS_METRIC_ENDPOINT="${RUSTFS_OBS_METRIC_ENDPOINT:-}"
 DIAGNOSTIC_OBS_METER_INTERVAL="${RUSTFS_OBS_METER_INTERVAL:-1}"
 DIAGNOSTIC_OBS_SERVICE_NAME_PREFIX="${RUSTFS_OBS_SERVICE_NAME:-RustFS-get-1mib-abba}"
 RESOURCE_SAMPLE_INTERVAL_SECS="${RUSTFS_GET_BENCH_RESOURCE_SAMPLE_INTERVAL_SECS:-5}"
+COMPRESSED_FALLBACK_PROBE=false
 HEALTH_TIMEOUT_SECS=60
 SKIP_BUILD=false
 DRY_RUN=false
@@ -103,6 +104,8 @@ Diagnostics:
                                           (default: <obs-endpoint>/v1/metrics)
   --diagnostic-obs-meter-interval <secs>
   --diagnostic-obs-service-name-prefix <name>
+  --compressed-fallback-probe    Forward the disk-compressed object fallback
+                                 probe to the underlying smoke runner
 
 Binary/options:
   --rustfs-bin <path>            RustFS binary (default: target/release/rustfs)
@@ -166,6 +169,7 @@ while [[ $# -gt 0 ]]; do
     --diagnostic-obs-metric-endpoint) DIAGNOSTIC_OBS_METRIC_ENDPOINT="$2"; shift 2 ;;
     --diagnostic-obs-meter-interval) DIAGNOSTIC_OBS_METER_INTERVAL="$2"; shift 2 ;;
     --diagnostic-obs-service-name-prefix) DIAGNOSTIC_OBS_SERVICE_NAME_PREFIX="$2"; shift 2 ;;
+    --compressed-fallback-probe) COMPRESSED_FALLBACK_PROBE=true; shift ;;
     --rustfs-bin) RUSTFS_BIN="$2"; shift 2 ;;
     --warp-bin) WARP_BIN="$2"; shift 2 ;;
     --python-bin) PYTHON_BIN="$2"; shift 2 ;;
@@ -276,6 +280,7 @@ diagnostic_metrics_url=${DIAGNOSTIC_METRICS_URL}
 diagnostic_obs_endpoint=${DIAGNOSTIC_OBS_ENDPOINT}
 diagnostic_obs_metric_endpoint=${DIAGNOSTIC_OBS_METRIC_ENDPOINT}
 diagnostic_obs_meter_interval=${DIAGNOSTIC_OBS_METER_INTERVAL}
+compressed_fallback_probe=${COMPRESSED_FALLBACK_PROBE}
 stage_metrics_artifacts=service_metrics_summary.csv,service_metrics_round_summary.csv,service_metrics_stage_distribution.csv,service_metrics_round_percentiles.csv
 body_header_parity_artifacts=compat_summary.csv,response_headers_legacy.json,response_headers_codec_legacy.json,body_sha256_legacy.txt,body_sha256_codec_legacy.txt
 performance_conclusion=not_encoded_by_harness_collect_raw_abba_stage_metrics_first
@@ -366,6 +371,9 @@ run_cell() {
   if [[ "$WARP_WARMUP_GET_BEFORE_BENCH" == "true" ]]; then
     cmd+=(--warp-warmup-get-before-bench)
   fi
+  if [[ "$COMPRESSED_FALLBACK_PROBE" == "true" ]]; then
+    cmd+=(--compressed-fallback-probe)
+  fi
   if [[ "$SKIP_BUILD" == "true" ]]; then
     cmd+=(--skip-build)
   fi

scripts/run_get_codec_streaming_smoke.sh

@@ -3034,7 +3034,9 @@ else:
 compressed_key = object_key + compressed_probe_extension
 compressed_path = bucket_path + "/" + urllib.parse.quote(compressed_key, safe="/-_.~")
 if compressed_fallback_probe:
-    compressed_body = (b"RustFS compressed fallback probe\n" * 4096)[:128 * 1024]
+    compressed_size = max(object_size, codec_min_size, 128 * 1024)
+    compressed_pattern = b"RustFS compressed fallback probe\n"
+    compressed_body = (compressed_pattern * ((compressed_size // len(compressed_pattern)) + 1))[:compressed_size]
     compressed_put, _ = request(
         "PUT",
         compressed_path,
@@ -3325,6 +3327,7 @@ else:
 
 encrypted_key = object_key + ".encrypted"
 encrypted_path = bucket_path + "/" + urllib.parse.quote(encrypted_key, safe="/-_.~")
+encrypted_probe_body = payload(max(object_size, codec_min_size, 1))
 ssec_key = b"0123456789abcdef0123456789abcdef"
 ssec_key_b64 = base64.b64encode(ssec_key).decode()
 ssec_key_md5_b64 = base64.b64encode(hashlib.md5(ssec_key).digest()).decode()
@@ -3333,7 +3336,7 @@ ssec_headers = [
     ("x-amz-server-side-encryption-customer-key", ssec_key_b64),
     ("x-amz-server-side-encryption-customer-key-md5", ssec_key_md5_b64),
 ]
-encrypted_put, _ = request("PUT", encrypted_path, body, [("content-type", "application/octet-stream"), *ssec_headers])
+encrypted_put, _ = request("PUT", encrypted_path, encrypted_probe_body, [("content-type", "application/octet-stream"), *ssec_headers])
 if encrypted_put["status"] not in (200, 204):
     fallback_rows.append(
         {
@@ -3348,8 +3351,8 @@ if encrypted_put["status"] not in (200, 204):
         }
     )
 else:
-    encrypted_get, encrypted_body = request("GET", encrypted_path, extra_headers=ssec_headers)
+    encrypted_get, encrypted_get_body = request("GET", encrypted_path, extra_headers=ssec_headers)
-    encrypted_ok = encrypted_get["status"] == 200 and sha256_hex(encrypted_body) == sha256_hex(body)
+    encrypted_ok = encrypted_get["status"] == 200 and sha256_hex(encrypted_get_body) == sha256_hex(encrypted_probe_body)
     fallback_rows.append(
         {
             "profile": profile,
@@ -3357,7 +3360,7 @@ else:
             "object_key": encrypted_key,
             "status": "ok" if encrypted_ok else "unexpected_status_or_body",
             "status_code": encrypted_get["status"],
-            "body_len": len(encrypted_body),
+            "body_len": len(encrypted_get_body),
             "expected_runtime_fallback_reason": "encrypted",
             "note": "SSE-C encrypted object GET should stay on the legacy fallback path for codec profiles",
         }

scripts/test_get_1mib_abba_stage_metrics.sh

@@ -21,6 +21,7 @@ trap cleanup EXIT
   --round-cooldown-secs 0 \
   --out-dir "$OUT_DIR" \
   --warp-bin true \
+  --compressed-fallback-probe \
   --skip-build \
   --dry-run >/dev/null
 
@@ -38,9 +39,12 @@ rg -qx 'diagnostic_metrics=true' "${OUT_DIR}/manifest.env"
 rg -qx 'diagnostic_obs_endpoint=http://127.0.0.1:4318' "${OUT_DIR}/manifest.env"
 rg -qx 'diagnostic_obs_metric_endpoint=http://127.0.0.1:4318/v1/metrics' "${OUT_DIR}/manifest.env"
 rg -qx 'diagnostic_obs_meter_interval=1' "${OUT_DIR}/manifest.env"
+rg -qx 'compressed_fallback_probe=true' "${OUT_DIR}/manifest.env"
 rg -qx 'performance_conclusion=not_encoded_by_harness_collect_raw_abba_stage_metrics_first' "${OUT_DIR}/manifest.env"
 rg -Fq '("service.name", "service_name", "job", "otel_scope_name")' "${SCRIPT_DIR}/run_get_codec_streaming_smoke.sh"
 rg -Fq '("service_name", "service.name", "job", "otel_scope_name")' "${SCRIPT_DIR}/run_get_codec_streaming_smoke.sh"
+rg -Fq 'compressed_size = max(object_size, codec_min_size, 128 * 1024)' "${SCRIPT_DIR}/run_get_codec_streaming_smoke.sh"
+rg -Fq 'encrypted_probe_body = payload(max(object_size, codec_min_size, 1))' "${SCRIPT_DIR}/run_get_codec_streaming_smoke.sh"
 
 matrix_rows="$(awk -F',' 'NR > 1 { count++ } END { print count + 0 }' "${OUT_DIR}/abba_matrix.csv")"
 if [[ "$matrix_rows" != "4" ]]; then
@@ -76,6 +80,9 @@ for outer_size in 65536 1048576; do
       rg -qx 'RUSTFS_OBS_ENDPOINT=http://127.0.0.1:4318' "$profile_manifest"
       rg -qx 'RUSTFS_OBS_METRIC_ENDPOINT=http://127.0.0.1:4318/v1/metrics' "$profile_manifest"
       rg -qx 'RUSTFS_GET_CODEC_STREAMING_MIN_SIZE=1048576' "$profile_manifest"
+      rg -qx 'RUSTFS_COMPRESSION_ENABLED=true' "$profile_manifest"
+      rg -qx 'RUSTFS_COMPRESSION_EXTENSIONS=.compressed-probe.txt' "$profile_manifest"
+      rg -qx 'RUSTFS_COMPRESSION_MIME_TYPES=text/plain' "$profile_manifest"
       test -f "${cell_dir}/${profile}/service_metrics_round_summary.csv"
       test -f "${cell_dir}/${profile}/service_metrics_stage_distribution.csv"
       test -f "${cell_dir}/${profile}/service_metrics_round_percentiles.csv"