diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index eaa9b21a1b343..03e5fbe45b595 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -370,6 +370,22 @@ jobs: with: save-if: false # set in linux-test shared-key: "amd-ci" + # The storage integration tests start MinIO containers. Pulling the image + # once up front keeps the pull off the critical path of the tests, which + # would otherwise pull it several times concurrently and occasionally fail + # with transient Docker transport errors. The tests retry the pull + # themselves, so a failure here is only a warning. + # + # MINIO_IMAGE must match what the tests start: the tag comes from the + # `minio` module of `testcontainers-modules`, the registry from + # MINIO_IMAGE_NAME in `datafusion-cli/tests/cli_integration.rs`. The + # `minio_image_matches_ci_prepull` test fails if either drifts. + - name: Pre-pull MinIO image + env: + MINIO_IMAGE: quay.io/minio/minio:RELEASE.2025-02-28T09-55-16Z + run: | + ci/scripts/retry timeout 120 docker pull "$MINIO_IMAGE" \ + || echo "::warning::Could not pre-pull $MINIO_IMAGE, the tests will pull it themselves" - name: Run tests (excluding doctests) env: RUST_BACKTRACE: 1 diff --git a/datafusion-cli/tests/cli_integration.rs b/datafusion-cli/tests/cli_integration.rs index 4dc244445a2eb..53f2ab89442ff 100644 --- a/datafusion-cli/tests/cli_integration.rs +++ b/datafusion-cli/tests/cli_integration.rs @@ -24,12 +24,13 @@ use insta::internals::SettingsBindDropGuard; use insta::{Settings, glob}; use insta_cmd::{assert_cmd_snapshot, get_cargo_bin}; use std::path::PathBuf; +use std::time::Duration; use std::{env, fs}; use testcontainers_modules::minio; use testcontainers_modules::testcontainers::core::{CmdWaitFor, ExecCommand, Mount}; use testcontainers_modules::testcontainers::runners::AsyncRunner; use testcontainers_modules::testcontainers::{ - ContainerAsync, ImageExt, TestcontainersError, + ContainerAsync, Image, ImageExt, TestcontainersError, }; fn cli() -> Command { @@ -45,10 +46,117 @@ fn make_settings() -> Settings { settings } +const MINIO_ROOT_USER: &str = "TEST-DataFusionLogin"; +const MINIO_ROOT_PASSWORD: &str = "TEST-DataFusionPassword"; + +/// Registry override for the image pinned by `testcontainers-modules`. +/// +/// MinIO withdrew `minio/minio` from Docker Hub on 2026-09-11. quay.io still +/// serves the same tag, so only the registry changes here. An unblock, not a +/// fix: see . +const MINIO_IMAGE_NAME: &str = "quay.io/minio/minio"; + +/// How many times to try bringing up the MinIO container before failing. +/// +/// Both the image pull and the `mc` calls that provision the bucket fail +/// intermittently on CI with transient errors such as +/// `bytes remaining on stream`. Retrying is much cheaper than a flaky run. +const MINIO_SETUP_ATTEMPTS: u32 = 3; + +/// Delay before the first retry of the MinIO setup, doubled on each attempt. +const MINIO_SETUP_RETRY_DELAY: Duration = Duration::from_secs(5); + +/// Time budget for a single MinIO setup attempt. A stalled image pull or `mc` +/// invocation is retried instead of hanging the whole test run. +const MINIO_SETUP_TIMEOUT: Duration = Duration::from_mins(3); + +/// Starts a MinIO container preloaded with the test data, retrying transient +/// Docker failures. +/// +/// Returns `None` when the test should be skipped, that is when +/// `TEST_STORAGE_INTEGRATION` is unset or the registry is rate limiting the +/// image pull. Panics if the container cannot be started for any other reason. +async fn start_minio_or_skip() -> Option> { + if env::var("TEST_STORAGE_INTEGRATION").is_err() { + eprintln!("Skipping external storages integration tests"); + return None; + } + + match setup_minio_container().await { + Ok(container) => Some(container), + Err(e) if is_docker_pull_rate_limit(&e) => { + eprintln!("Skipping test: Docker pull rate limit reached: {e}"); + None + } + Err(e) => panic!("{e}"), + } +} + +/// A registry pull rate limit does not clear up within a test run, so the +/// affected tests are skipped rather than retried. +fn is_docker_pull_rate_limit(error: &str) -> bool { + error.contains("toomanyrequests") +} + +/// Retrying only pays off for transient failures. An exhausted pull quota or a +/// Docker daemon that cannot be reached at all stays broken for the whole run. +fn is_retryable(error: &str) -> bool { + !is_docker_pull_rate_limit(error) + && !error.contains("failed to initialize a docker client") +} + async fn setup_minio_container() -> Result, String> { - const MINIO_ROOT_USER: &str = "TEST-DataFusionLogin"; - const MINIO_ROOT_PASSWORD: &str = "TEST-DataFusionPassword"; + let mut delay = MINIO_SETUP_RETRY_DELAY; + let mut last_error = String::from("MinIO container setup was not attempted at all"); + + for attempt in 1..=MINIO_SETUP_ATTEMPTS { + last_error = match tokio::time::timeout( + MINIO_SETUP_TIMEOUT, + try_setup_minio_container(), + ) + .await + { + Ok(Ok(container)) => return Ok(container), + Ok(Err(e)) => e, + Err(_) => format!( + "Timed out after {MINIO_SETUP_TIMEOUT:?} while starting the MinIO container" + ), + }; + + if attempt == MINIO_SETUP_ATTEMPTS || !is_retryable(&last_error) { + break; + } + + eprintln!( + "MinIO container setup failed (attempt {attempt}/{MINIO_SETUP_ATTEMPTS}), \ + retrying in {delay:?}: {last_error}" + ); + tokio::time::sleep(delay).await; + delay *= 2; + } + + Err(last_error) +} + +/// A single attempt at starting and provisioning a MinIO container. +/// +/// The container is removed again if provisioning fails, so that the next +/// attempt starts from a clean state. +async fn try_setup_minio_container() -> Result, String> { + let container = start_minio_container().await?; + + match provision_minio_container(&container).await { + Ok(()) => Ok(container), + Err(e) => { + if let Err(rm_error) = container.rm().await { + eprintln!("Failed to remove the MinIO container: {rm_error}"); + } + Err(e) + } + } +} +async fn start_minio_container() -> Result, String> { let data_path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../datafusion/core/tests/data"); @@ -56,7 +164,8 @@ async fn setup_minio_container() -> Result, String> .canonicalize() .expect("Failed to get absolute path for test data"); - let container = minio::MinIO::default() + minio::MinIO::default() + .with_name(MINIO_IMAGE_NAME) .with_env_var("MINIO_ROOT_USER", MINIO_ROOT_USER) .with_env_var("MINIO_ROOT_PASSWORD", MINIO_ROOT_PASSWORD) .with_mount(Mount::bind_mount( @@ -64,60 +173,93 @@ async fn setup_minio_container() -> Result, String> "/source", )) .start() - .await; - - match container { - Ok(container) => { - // We wait for MinIO to be healthy and prepare test files. We do it via CLI to avoid s3 dependency - let commands = [ - ExecCommand::new(["/usr/bin/mc", "ready", "local"]), - ExecCommand::new([ - "/usr/bin/mc", - "alias", - "set", - "localminio", - "http://localhost:9000", - MINIO_ROOT_USER, - MINIO_ROOT_PASSWORD, - ]), - ExecCommand::new(["/usr/bin/mc", "mb", "localminio/data"]), - ExecCommand::new([ - "/usr/bin/mc", - "cp", - "-r", - "/source/", - "localminio/data/", - ]), - ]; - - for command in commands { - let command = - command.with_cmd_ready_condition(CmdWaitFor::Exit { code: Some(0) }); - - let cmd_ref = format!("{command:?}"); - - if let Err(e) = container.exec(command).await { - let stdout = container.stdout_to_vec().await.unwrap_or_default(); - let stderr = container.stderr_to_vec().await.unwrap_or_default(); - - return Err(format!( - "Failed to execute command: {}\nError: {}\nStdout: {:?}\nStderr: {:?}", - cmd_ref, - e, - String::from_utf8_lossy(&stdout), - String::from_utf8_lossy(&stderr) - )); - } - } + .await + .map_err(|e| match e { + TestcontainersError::Client(e) => format!( + "Failed to start MinIO container. Ensure Docker is running and accessible: {e}" + ), + e => format!("Failed to start MinIO container: {e}"), + }) +} - Ok(container) +/// Waits for MinIO to be healthy and uploads the test files. +/// +/// This is done via the `mc` CLI shipped in the image to avoid an s3 dependency. +async fn provision_minio_container( + container: &ContainerAsync, +) -> Result<(), String> { + let commands = [ + ExecCommand::new(["/usr/bin/mc", "ready", "local"]), + ExecCommand::new([ + "/usr/bin/mc", + "alias", + "set", + "localminio", + "http://localhost:9000", + MINIO_ROOT_USER, + MINIO_ROOT_PASSWORD, + ]), + ExecCommand::new(["/usr/bin/mc", "mb", "localminio/data"]), + ExecCommand::new(["/usr/bin/mc", "cp", "-r", "/source/", "localminio/data/"]), + ]; + + for command in commands { + let command = + command.with_cmd_ready_condition(CmdWaitFor::Exit { code: Some(0) }); + + let cmd_ref = format!("{command:?}"); + + if let Err(e) = container.exec(command).await { + let stdout = container.stdout_to_vec().await.unwrap_or_default(); + let stderr = container.stderr_to_vec().await.unwrap_or_default(); + + return Err(format!( + "Failed to execute command: {}\nError: {}\nStdout: {:?}\nStderr: {:?}", + cmd_ref, + e, + String::from_utf8_lossy(&stdout), + String::from_utf8_lossy(&stderr) + )); } - - Err(TestcontainersError::Client(e)) => Err(format!( - "Failed to start MinIO container. Ensure Docker is running and accessible: {e}" - )), - Err(e) => Err(format!("Failed to start MinIO container: {e}")), } + + Ok(()) +} + +/// CI pre-pulls the MinIO image so that the storage integration tests do not +/// have to pull it themselves. Guard against that pre-pull going stale when +/// `testcontainers-modules` bumps the image it uses. +#[test] +fn minio_image_matches_ci_prepull() { + let workflow = + PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../.github/workflows/rust.yml"); + + // The workflow is not shipped with the published crate. + let Ok(contents) = fs::read_to_string(&workflow) else { + return; + }; + + let image = minio::MinIO::default(); + + // The override only redirects the registry, so it would silently stop + // tracking upstream if the crate ever pinned a different image. + assert!( + MINIO_IMAGE_NAME.ends_with(&format!("/{}", image.name())), + "`testcontainers-modules` now uses `{}`, which MINIO_IMAGE_NAME \ + (`{MINIO_IMAGE_NAME}`) no longer mirrors.", + image.name() + ); + + // Match the assignment: `minio/minio:` is a substring of the quay + // reference and would pass either way. + let image_ref = format!("{MINIO_IMAGE_NAME}:{}", image.tag()); + + assert!( + contents.contains(&format!("MINIO_IMAGE: {image_ref}")), + "{} does not pre-pull `{image_ref}`. Update MINIO_IMAGE in the \ + `Pre-pull MinIO image` step to match the image used by the tests.", + workflow.display() + ); } #[cfg(test)] @@ -522,18 +664,8 @@ fn test_cli_wide_result_set_no_crash() { #[tokio::test] async fn test_cli() { - if env::var("TEST_STORAGE_INTEGRATION").is_err() { - eprintln!("Skipping external storages integration tests"); + let Some(container) = start_minio_or_skip().await else { return; - } - - let container = match setup_minio_container().await { - Ok(c) => c, - Err(e) if e.contains("toomanyrequests") => { - eprintln!("Skipping test: Docker pull rate limit reached: {e}"); - return; - } - e @ Err(_) => e.unwrap(), }; let settings = make_settings(); @@ -546,8 +678,8 @@ async fn test_cli() { assert_cmd_snapshot!( cli() .env_clear() - .env("AWS_ACCESS_KEY_ID", "TEST-DataFusionLogin") - .env("AWS_SECRET_ACCESS_KEY", "TEST-DataFusionPassword") + .env("AWS_ACCESS_KEY_ID", MINIO_ROOT_USER) + .env("AWS_SECRET_ACCESS_KEY", MINIO_ROOT_PASSWORD) .env("AWS_ENDPOINT", format!("http://localhost:{port}")) .env("AWS_ALLOW_HTTP", "true") .pass_stdin(input) @@ -559,22 +691,13 @@ async fn test_cli() { async fn test_aws_options() { // Separate test is needed to pass aws as options in sql and not via env - if env::var("TEST_STORAGE_INTEGRATION").is_err() { - eprintln!("Skipping external storages integration tests"); + let Some(container) = start_minio_or_skip().await else { return; - } + }; let settings = make_settings(); let _bound = settings.bind_to_scope(); - let container = match setup_minio_container().await { - Ok(c) => c, - Err(e) if e.contains("toomanyrequests") => { - eprintln!("Skipping test: Docker pull rate limit reached: {e}"); - return; - } - e @ Err(_) => e.unwrap(), - }; let port = container.get_host_port_ipv4(9000).await.unwrap(); let input = format!( @@ -582,8 +705,8 @@ async fn test_aws_options() { STORED AS CSV LOCATION 's3://data/cars.csv' OPTIONS( - 'aws.access_key_id' 'TEST-DataFusionLogin', - 'aws.secret_access_key' 'TEST-DataFusionPassword', + 'aws.access_key_id' '{MINIO_ROOT_USER}', + 'aws.secret_access_key' '{MINIO_ROOT_PASSWORD}', 'aws.endpoint' 'http://localhost:{port}', 'aws.allow_http' 'true' ); @@ -658,18 +781,8 @@ fn test_backtrace_output(#[case] query: &str) { #[tokio::test] async fn test_s3_url_fallback() { - if env::var("TEST_STORAGE_INTEGRATION").is_err() { - eprintln!("Skipping external storages integration tests"); + let Some(container) = start_minio_or_skip().await else { return; - } - - let container = match setup_minio_container().await { - Ok(c) => c, - Err(e) if e.contains("toomanyrequests") => { - eprintln!("Skipping test: Docker pull rate limit reached: {e}"); - return; - } - e @ Err(_) => e.unwrap(), }; let mut settings = make_settings(); @@ -695,19 +808,10 @@ SELECT * FROM partitioned_data ORDER BY column_1, column_2 LIMIT 5; /// Validate object store profiling output #[tokio::test] async fn test_object_store_profiling() { - if env::var("TEST_STORAGE_INTEGRATION").is_err() { - eprintln!("Skipping external storages integration tests"); + let Some(container) = start_minio_or_skip().await else { return; - } - - let container = match setup_minio_container().await { - Ok(c) => c, - Err(e) if e.contains("toomanyrequests") => { - eprintln!("Skipping test: Docker pull rate limit reached: {e}"); - return; - } - e @ Err(_) => e.unwrap(), }; + let mut settings = make_settings(); // as the object store profiling contains timestamps and durations, we must @@ -769,8 +873,8 @@ impl MinioCommandExt for Command { let port = container.get_host_port_ipv4(9000).await.unwrap(); self.env_clear() - .env("AWS_ACCESS_KEY_ID", "TEST-DataFusionLogin") - .env("AWS_SECRET_ACCESS_KEY", "TEST-DataFusionPassword") + .env("AWS_ACCESS_KEY_ID", MINIO_ROOT_USER) + .env("AWS_SECRET_ACCESS_KEY", MINIO_ROOT_PASSWORD) .env("AWS_ENDPOINT", format!("http://localhost:{port}")) .env("AWS_ALLOW_HTTP", "true") }