From 09a991e7351889f46bafef15008ebbbf95ab3a38 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 26 Jul 2026 19:44:53 +0800 Subject: [PATCH] test(perf): cover GET fallback metric probes (#5291) * test(perf): cover GET fallback metric probes Co-Authored-By: heihutu * test(perf): prefer format fallback labels Co-Authored-By: heihutu * test(ilm): remove duplicate manual transition import Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- crates/ecstore/src/set_disk/mod.rs | 30 +++++----- crates/ecstore/src/set_disk/read.rs | 66 ++++++++++++++++++++- scripts/run_get_1mib_abba_stage_metrics.sh | 8 +++ scripts/run_get_codec_streaming_smoke.sh | 13 ++-- scripts/test_get_1mib_abba_stage_metrics.sh | 7 +++ 5 files changed, 103 insertions(+), 21 deletions(-) diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 4c7308a81..e52099664 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/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, diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index c55d4ed60..99774e244 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/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>, || { diff --git a/scripts/run_get_1mib_abba_stage_metrics.sh b/scripts/run_get_1mib_abba_stage_metrics.sh index eee1c16a2..48b3341fa 100755 --- a/scripts/run_get_1mib_abba_stage_metrics.sh +++ b/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: /v1/metrics) --diagnostic-obs-meter-interval --diagnostic-obs-service-name-prefix + --compressed-fallback-probe Forward the disk-compressed object fallback + probe to the underlying smoke runner Binary/options: --rustfs-bin 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 diff --git a/scripts/run_get_codec_streaming_smoke.sh b/scripts/run_get_codec_streaming_smoke.sh index f325ddc48..b8631634e 100755 --- a/scripts/run_get_codec_streaming_smoke.sh +++ b/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_ok = encrypted_get["status"] == 200 and sha256_hex(encrypted_body) == sha256_hex(body) + encrypted_get, encrypted_get_body = request("GET", encrypted_path, extra_headers=ssec_headers) + 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", } diff --git a/scripts/test_get_1mib_abba_stage_metrics.sh b/scripts/test_get_1mib_abba_stage_metrics.sh index 805ab38e9..50a63fec3 100755 --- a/scripts/test_get_1mib_abba_stage_metrics.sh +++ b/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"