Merge branch 'main' into ray-subsystembenchmark-checkpointing-save

# Conflicts:
#	gcsfs/tests/perf/subsystembenchmarks/tests/test_run_groups.py

Author: Yonghui-Lee

.github/workflows/ci.yml

@@ -75,6 +75,9 @@ jobs:
       - name: Run Release Automation Tests
         run: |
           pytest -vv .github/release/
+      - name: Run Cloud Build Test Report Unit Tests
+        run: |
+          pytest -vv cloudbuild/test_summarize_test_results.py
       - name: Run all tests (Standard, Zonal & HNS Enabled) with default ON extended feature support
         run: |
           export GOOGLE_APPLICATION_CREDENTIALS=$(pwd)/gcsfs/tests/fake-service-account-credentials.json
@@ -108,13 +111,15 @@ jobs:
           cache-dependency-path: |
             gcsfs/tests/perf/subsystembenchmarks/dataloading/huggingface_datasets/requirements.txt
             gcsfs/tests/perf/subsystembenchmarks/dataloading/webdataset/requirements.txt
+            gcsfs/tests/perf/subsystembenchmarks/dataloading/ray_data/requirements.txt
       - name: Install subsystem benchmark dependencies
         # Resolve dependencies together to catch version conflicts.
         run: |
           python -m pip install -e .
           python -m pip install \
             -r gcsfs/tests/perf/subsystembenchmarks/dataloading/huggingface_datasets/requirements.txt \
-            -r gcsfs/tests/perf/subsystembenchmarks/dataloading/webdataset/requirements.txt
+            -r gcsfs/tests/perf/subsystembenchmarks/dataloading/webdataset/requirements.txt \
+            -r gcsfs/tests/perf/subsystembenchmarks/dataloading/ray_data/requirements.txt
       - name: Run subsystem benchmark unit tests
         run: pytest gcsfs/tests/perf/subsystembenchmarks --run-benchmarks-infra
 

cloudbuild/benchmarks/benchmarks-cloudbuild.yaml

@@ -262,7 +262,7 @@ steps:
           echo '--- Installing pip packages ---'
           pip install --upgrade pip > /dev/null
           pip install pytest pytest-timeout pytest-subtests pytest-asyncio fusepy google-cloud-storage > /dev/null
-          pip install --upgrade 'git+https://github.com/googleapis/google-cloud-python.git@main#subdirectory=packages/google-cloud-storage'
+
           echo '--- Installing GCSFS packages ---'
           # Manual builds via 'gcloud builds submit' don't upload .git directory,
           # causing hatch-vcs to fail. We use a dummy version in that case.

cloudbuild/benchmarks/benchmarks_schema.json

@@ -35,6 +35,10 @@
       {"name": "seq_probability", "type": "STRING"},
       {"name": "engine", "type": "STRING"},
       {"name": "method", "type": "STRING"},
+      {"name": "batch_size", "type": "INTEGER"},
+      {"name": "max_gap", "type": "INTEGER"},
+      {"name": "num_ranges", "type": "INTEGER"},
+      {"name": "total_bytes", "type": "INTEGER"},
       {"name": "min", "type": "FLOAT"},
       {"name": "max", "type": "FLOAT"},
       {"name": "mean", "type": "FLOAT"},

cloudbuild/e2e-tests-cloudbuild.yaml

@@ -186,6 +186,45 @@ steps:
     waitFor: ["setup-vm", "create-buckets"]
     allowFailure: true
 
+  # Test Report
+  # GitHub shows only the first ~65k chars of the build log, so the test steps write their full output to
+  # files on the VM. This step prints a size-capped summary (counts + failure tracebacks) first, then the
+  # full logs. It always exits 0 so that cleanup still runs.
+  - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
+    id: "test-report"
+    entrypoint: "bash"
+    args:
+      - "-c"
+      - |
+        mkdir -p /workspace/test-results
+        if gcloud compute instances describe gcsfs-test-vm-${_SHORT_BUILD_ID} --zone=${_ZONE} &>/dev/null; then
+          gcloud compute scp --recurse gcsfs-test-vm-${_SHORT_BUILD_ID}:~/test-results /workspace/ \
+            --zone=${_ZONE} \
+            --internal-ip \
+            --ssh-key-file=/workspace/.ssh/google_compute_engine > /dev/null || echo "⚠️ Could not copy test results from the VM"
+        fi
+
+        python3 cloudbuild/summarize_test_results.py /workspace/test-results \
+          --suites standard zonal zonal-core hns \
+          --failed-steps /workspace/FAILED \
+          --max-chars 30000 || echo "⚠️ Could not summarize test results"
+
+        for suite in standard zonal zonal-core hns; do
+          if [[ -f /workspace/test-results/$${suite}.log ]]; then
+            echo "===== FULL LOG: $${suite} ====="
+            cat /workspace/test-results/$${suite}.log
+          fi
+        done
+        exit 0
+    waitFor:
+      [
+        "run-standard-tests",
+        "run-zonal-tests",
+        "run-hns-tests",
+        "run-zonal-core-tests",
+      ]
+    allowFailure: true
+
   # Cleanup
   - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"
     id: "cleanup"
@@ -202,13 +241,7 @@ steps:
       - "WORKERS_HNS=${_WORKERS_HNS}"
       - "WORKERS_ZONAL=${_WORKERS_ZONAL}"
       - "WORKERS_ZONAL_CORE=${_WORKERS_ZONAL_CORE}"
-    waitFor:
-      [
-        "run-standard-tests",
-        "run-zonal-tests",
-        "run-hns-tests",
-        "run-zonal-core-tests",
-      ]
+    waitFor: ["test-report"]
 
   # Fail the build if any previous step failed
   - name: "gcr.io/google.com/cloudsdktool/cloud-sdk"

cloudbuild/run_tests.sh

@@ -24,14 +24,60 @@ if ! [[ "${PYTEST_XDIST_WORKERS}" =~ ^[0-9]+$ ]] || (( PYTEST_XDIST_WORKERS <= 0
 fi
 ARGS+=(-n "${PYTEST_XDIST_WORKERS}")
 
+# Full pytest output goes to files instead of the build log. GitHub shows only the first ~65k chars
+# of the Cloud Build log, so the test-report step prints a summary from the JUnit XML before these logs.
+RESULTS_DIR="${HOME}/test-results"
+RESULTS_LOG="${RESULTS_DIR}/${TEST_SUITE}.log"
+mkdir -p "${RESULTS_DIR}"
+ARGS+=("--junitxml=${RESULTS_DIR}/${TEST_SUITE}.xml")
+
+# While pytest runs, print the tests still in progress every HEARTBEAT_SECS. A hung test is failed by
+# pytest-timeout and a killed pytest leaves its log tail for test-report, but if the whole build times out
+# (or the VM is lost) test-report never runs, so these lines keep a stuck test visible in the step log.
+HEARTBEAT_SECS="${HEARTBEAT_SECS:-300}"
+if ! [[ "${HEARTBEAT_SECS}" =~ ^[0-9]+$ ]] || (( HEARTBEAT_SECS <= 0 )); then
+  HEARTBEAT_SECS=300
+fi
+
+in_progress_tests() {
+  # With -vv, xdist logs "<nodeid>" when a test starts and "[gwN] <OUTCOME> <nodeid>" when it finishes.
+  awk '
+    { sub(/ <- .*/, ""); sub(/[ \t]+$/, "") }
+    /^\[gw[0-9]+\] [A-Z]+ / { sub(/^\[gw[0-9]+\] [A-Z]+ /, ""); delete running[$0]; next }
+    /^[^ ]+\.py::/ { running[$0] = 1 }
+    END { for (t in running) printf "%s%s", (n++ ? ", " : ""), t; if (!n) printf "none" }
+  ' "${RESULTS_LOG}" 2>/dev/null | cut -c1-500
+}
+
+heartbeat() {
+  local elapsed=0
+  # Sleep in 1s steps so stopping the heartbeat never leaves a long sleep holding the ssh session open.
+  while sleep 1; do
+    elapsed=$((elapsed + 1))
+    if (( elapsed % HEARTBEAT_SECS == 0 )); then
+      echo "--- ${TEST_SUITE}: still running after $((elapsed / 60))m; in progress: $(in_progress_tests) ---"
+    fi
+  done
+}
+
+run_pytest() {
+  local status=0
+  heartbeat &
+  local heartbeat_pid=$!
+  pytest "$@" > "${RESULTS_LOG}" 2>&1 || status=$?
+  { kill "${heartbeat_pid}" && wait "${heartbeat_pid}"; } 2>/dev/null
+  return "${status}"
+}
+STATUS=0
+
 echo "--- Running Test Suite: ${TEST_SUITE} ---"
 
 case "$TEST_SUITE" in
   "standard")
     export GCSFS_TEST_BUCKET="gcsfs-test-standard-${SHORT_BUILD_ID}"
     export GCSFS_TEST_VERSIONED_BUCKET="gcsfs-test-versioned-${SHORT_BUILD_ID}"
     export GCSFS_TEST_REQ_PAYS_BUCKET="gcsfs-test-standard-req-pay-${SHORT_BUILD_ID}"
-    pytest "${ARGS[@]}" gcsfs/ --deselect gcsfs/tests/test_core.py::test_sign
+    run_pytest "${ARGS[@]}" gcsfs/ --deselect gcsfs/tests/test_core.py::test_sign || STATUS=$?
     ;;
 
   "zonal")
@@ -44,13 +90,13 @@ case "$TEST_SUITE" in
     export GCSFS_RUN_RAPID_TESTS="true"
     export GCSFS_EXPERIMENTAL_ZB_HNS_SUPPORT='true'
     # Excludes tests related to requster pays as Zonal buckets do not support requester pays feature
-    pytest "${ARGS[@]}" \
-      gcsfs/tests/test_extended_gcsfs.py \
+    run_pytest "${ARGS[@]}" \
+      gcsfs/tests/test_zonal.py \
       gcsfs/tests/test_zonal_file.py \
-      gcsfs/tests/integration/test_async_gcsfs.py \
-      gcsfs/tests/integration/test_extended_hns.py \
-      --deselect gcsfs/tests/integration/test_extended_hns.py::TestExtendedGcsFileSystemHnsRequesterPays::test_hns_mkdir_fails_without_quota_project \
-      --deselect gcsfs/tests/integration/test_extended_hns.py::TestExtendedGcsFileSystemHnsRequesterPays::test_hns_bucket_type_detection_with_req_pays
+      gcsfs/tests/test_async.py \
+      gcsfs/tests/test_hns.py \
+      --deselect gcsfs/tests/test_hns.py::TestExtendedGcsFileSystemHnsRequesterPays::test_hns_mkdir_fails_without_quota_project \
+      --deselect gcsfs/tests/test_hns.py::TestExtendedGcsFileSystemHnsRequesterPays::test_hns_bucket_type_detection_with_req_pays || STATUS=$?
     ;;
 
   "hns")
@@ -62,20 +108,20 @@ case "$TEST_SUITE" in
     export GCSFS_RUN_HNS_TESTS="true"
     export GCSFS_EXPERIMENTAL_ZB_HNS_SUPPORT='true'
     # Excludes tests that are not applicable to HNS buckets:
-    # - test_extended_gcsfs.py, test_zonal_file.py: Zonal bucket specific tests which won't work on HNS bucket.
-    # - test_extended_gcsfs_unit.py: Unit tests for zonal bucket features.
-    # - test_core_versioned.py: HNS buckets do not support versioning.
+    # - test_zonal.py, test_zonal_file.py: Zonal bucket specific tests which won't work on HNS bucket.
+    # - test_zonal_unit.py: Unit tests for zonal bucket features.
+    # - test_flat_versioned.py: HNS buckets do not support versioning.
     # - test_core.py::test_sign: Current Cloud Build auth setup does not support this.
     # - test_core.py::test_mv_file_cache: Integration test only applicable for regional buckets.
     # - test_core.py::test_rm_wildcards_non_recursive: HNS buckets have different behavior for non-recursive wildcard deletion.
-    pytest "${ARGS[@]}" gcsfs/ \
-      --deselect gcsfs/tests/test_extended_gcsfs.py \
+    run_pytest "${ARGS[@]}" gcsfs/ \
+      --deselect gcsfs/tests/test_zonal.py \
       --deselect gcsfs/tests/test_zonal_file.py \
-      --deselect gcsfs/tests/test_extended_gcsfs_unit.py \
-      --deselect gcsfs/tests/test_core_versioned.py \
+      --deselect gcsfs/tests/test_zonal_unit.py \
+      --deselect gcsfs/tests/test_flat_versioned.py \
       --deselect gcsfs/tests/test_core.py::test_sign \
       --deselect gcsfs/tests/test_core.py::test_mv_file_cache \
-      --deselect gcsfs/tests/test_core.py::test_rm_wildcards_non_recursive
+      --deselect gcsfs/tests/test_core.py::test_rm_wildcards_non_recursive || STATUS=$?
     ;;
 
   "zonal-core")
@@ -137,24 +183,16 @@ case "$TEST_SUITE" in
       "--deselect=gcsfs/tests/test_core.py::test_put_file_resumable_upload_cleanup_on_chunk_failure"
     )
 
-    # The prefetcher engine is not integrated for zonal in this bucket.
-    # It will be integrated in a separate PR, after which this will be removed.
-    ZONAL_DESELECTS+=(
-      "--deselect=gcsfs/tests/test_core.py::test_cat_file_routing_and_thresholds"
-      "--deselect=gcsfs/tests/test_core.py::test_cat_file_concurrent_data_integrity"
-      "--deselect=gcsfs/tests/test_core.py::test_cat_file_concurrent_exception_cancellation"
-      "--deselect=gcsfs/tests/test_core.py::test_gcsfile_prefetch_disabled_fallback"
-      "--deselect=gcsfs/tests/test_core.py::test_gcsfile_prefetch_sequential_integrity"
-      "--deselect=gcsfs/tests/test_core.py::test_gcsfile_prefetch_random_seek_integrity"
-      "--deselect=gcsfs/tests/test_core.py::test_gcsfile_multithreaded_read_integrity"
-      "--deselect=gcsfs/tests/test_core.py::test_gcsfile_not_satisfiable_range"
-    )
-
     # Zonal buckets do not support the requester pays feature
     ZONAL_DESELECTS+=(
       "--deselect=gcsfs/tests/test_core.py::test_requester_pays_fails_without_user_project"
     )
 
-    pytest "${ARGS[@]}" "${ZONAL_DESELECTS[@]}" gcsfs/tests/test_core.py
+    run_pytest "${ARGS[@]}" "${ZONAL_DESELECTS[@]}" gcsfs/tests/test_core.py || STATUS=$?
     ;;
 esac
+
+# Pytest's final "=== N passed, M skipped in Xs ===" line, so the step log still shows the counts.
+RESULT_LINE=$(grep -E '^=+ .* in [0-9.]+s' "${RESULTS_LOG}" 2>/dev/null | tail -n 1 | sed -E 's/^=+ | =+$//g')
+echo "--- ${TEST_SUITE}: pytest exit ${STATUS} ${RESULT_LINE} ---"
+exit "${STATUS}"

cloudbuild/subsystembenchmarks/subsystembenchmarks_schema.json

@@ -10,7 +10,7 @@
     "fields": [
       {"name": "benchmark_case_id", "type": "STRING", "description": "Stable benchmark case identifier encoding every swept parameter."},
       {"name": "benchmark_group", "type": "STRING", "description": "pytest-benchmark's group label, which publish_case_metadata sets to the case scenario: read for dataloading, checkpoint_write for checkpointing. Same value as workload_scenario; filter on workload_family and workload_implementation to tell the loaders within a subsystem apart."},
-      {"name": "workload_implementation", "type": "STRING", "description": "Library or integration executing the workload, matching the run.py --group suffix: huggingface_datasets, webdataset, or pytorch_lightning."},
+      {"name": "workload_implementation", "type": "STRING", "description": "Library or integration executing the workload, matching the run.py --group suffix: huggingface_datasets, webdataset, pytorch_lightning, or ray_data."},
       {"name": "workload_scenario", "type": "STRING", "description": "Operation exercised within the workload implementation, named by the scenario in the group's configs.yaml: read for dataloading, checkpoint_write for checkpointing."},
       {"name": "config_sweep_axis", "type": "STRING", "description": "Configuration axis that produced this case, or baseline for the implicit baseline case."},
       {"name": "workload_family", "type": "STRING", "description": "Broad benchmark row family: data_loading or checkpointing. With workload_implementation it identifies the runnable group, which benchmark_group also carries verbatim (note data_loading here against the dataloading directory the group name uses)."},
@@ -25,11 +25,11 @@
       {"name": "requirements_resolved", "type": "STRING", "description": "JSON list from pip describing final installed distribution names and resolved versions."},
       {"name": "config_sweep_axes_requested", "type": "STRING", "description": "Normalized requested config sweep axes, or all when the complete group was selected; baseline is included automatically."},
       {"name": "measurement_round_count", "type": "INTEGER", "description": "Number of measured rounds represented by the round-duration statistics. A round is one timed unit of the workload: one full-corpus iteration for dataloading, one checkpoint operation for checkpointing."},
-      {"name": "dataloader_num_workers", "type": "INTEGER", "description": "Populated by dataloading. DataLoader worker process count per distributed rank; 0 means in-process loading. Shared name with the macrobenchmarks schema."},
+      {"name": "dataloader_num_workers", "type": "INTEGER", "description": "Populated by dataloading. DataLoader worker process count per distributed rank (or Ray Data reader task concurrency); 0 means in-process loading or library-default concurrency. Shared name with the macrobenchmarks schema."},
       {"name": "batch_size_samples", "type": "INTEGER", "description": "Populated by dataloading. Number of samples yielded per data batch."},
-      {"name": "sample_sequence_length_tokens", "type": "INTEGER", "description": "Populated by dataloading/huggingface_datasets. Number of tokens in each generated dataset sample."},
+      {"name": "sample_sequence_length_tokens", "type": "INTEGER", "description": "Populated by dataloading/huggingface_datasets and dataloading/ray_data. Number of tokens in each generated dataset sample."},
       {"name": "dataset_format", "type": "STRING", "description": "Populated by dataloading. On-GCS representation of the generated dataset, pairing stored payload with container: pretok_parquet, text_parquet, pretok_jsonl, image_tar_jpeg, image_tar_png, image_tar_npy, or image_tar_gz_jpeg. Image payload encoding rides here rather than in a column of its own, matching how pretok_ and text_ already distinguish payloads sharing a container."},
-      {"name": "parquet_row_group_size_rows", "type": "INTEGER", "description": "Populated by dataloading/huggingface_datasets. Number of rows per Parquet row group; not applicable to non-Parquet formats."},
+      {"name": "parquet_row_group_size_rows", "type": "INTEGER", "description": "Populated by dataloading/huggingface_datasets and dataloading/ray_data. Number of rows per Parquet row group; not applicable to non-Parquet formats."},
       {"name": "dataset_file_count", "type": "INTEGER", "description": "Populated by dataloading. Number of files in the generated dataset."},
       {"name": "dataset_size_bytes", "type": "INTEGER", "description": "Populated by dataloading. Total stored size in bytes of the generated GCS dataset objects. Shared name with the macrobenchmarks schema."},
       {"name": "dataset_sample_count", "type": "INTEGER", "description": "Populated by dataloading. Total number of samples in the generated dataset; every measured round must yield this count."},