This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch harden-integration-image-pulls
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit 5bfe4f9777610ed75537f68f4aae2aec09abfec1
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Jul 27 16:17:28 2026 +0200

    Stop registry blips from failing integration test jobs
    
    A Docker Hub timeout while pulling the integration images fails the job
    before a single test runs, and the retry that follows is a full re-run of
    the suite that starts within a minute of the first one - too soon to
    outlast the outage that caused it. Pulling the images up front retries
    the operation that actually fails, and waiting before the re-run gives a
    transient outage time to clear.
---
 .../airflow_breeze/commands/testing_commands.py    |  7 ++
 .../airflow_breeze/utils/docker_command_utils.py   | 79 +++++++++++++++++++++
 dev/breeze/tests/test_docker_command_utils.py      | 82 ++++++++++++++++++++++
 .../ci/testing/run_integration_tests_with_retry.sh |  5 ++
 4 files changed, 173 insertions(+)

diff --git a/dev/breeze/src/airflow_breeze/commands/testing_commands.py 
b/dev/breeze/src/airflow_breeze/commands/testing_commands.py
index 7731c25a354..153de7034a3 100644
--- a/dev/breeze/src/airflow_breeze/commands/testing_commands.py
+++ b/dev/breeze/src/airflow_breeze/commands/testing_commands.py
@@ -104,6 +104,7 @@ from airflow_breeze.utils.docker_command_utils import (
     fix_ownership_using_docker,
     notify_on_unhealthy_backend_container,
     perform_environment_checks,
+    pull_images_with_retries,
     remove_docker_networks,
 )
 from airflow_breeze.utils.environment_check import is_ci_environment
@@ -275,6 +276,12 @@ def _run_test(
     run_cmd.extend(pytest_args)
     try:
         remove_docker_networks(networks=[f"{compose_project_name}_default"])
+        if shell_params.test_group in (GroupOfTests.INTEGRATION_CORE, 
GroupOfTests.INTEGRATION_PROVIDERS):
+            pull_images_with_retries(
+                compose_project_name=compose_project_name,
+                env=env,
+                skip_images={shell_params.airflow_image_name},
+            )
         result = run_command(
             run_cmd,
             output=output,
diff --git a/dev/breeze/src/airflow_breeze/utils/docker_command_utils.py 
b/dev/breeze/src/airflow_breeze/utils/docker_command_utils.py
index f1fa87c9493..9875697c0d1 100644
--- a/dev/breeze/src/airflow_breeze/utils/docker_command_utils.py
+++ b/dev/breeze/src/airflow_breeze/utils/docker_command_utils.py
@@ -26,6 +26,7 @@ import re
 import shlex
 import subprocess
 import sys
+import time
 from functools import lru_cache
 from subprocess import DEVNULL, CompletedProcess
 from typing import TYPE_CHECKING
@@ -702,6 +703,84 @@ def fix_ownership_using_docker(quiet: bool = True):
     run_command(cmd, text=True, check=False, quiet=quiet)
 
 
+IMAGE_PULL_ATTEMPTS = 5
+IMAGE_PULL_BACKOFF_SECONDS = 15
+
+
+def get_images_to_pull(compose_project_name: str, env: dict[str, str], 
skip_images: set[str]) -> list[str]:
+    """
+    Returns third-party images of the compose project that are not available 
locally yet.
+
+    :param compose_project_name: name of the docker compose project
+    :param env: environment variables to resolve the compose files with
+    :param skip_images: images that should never be pulled (the locally built 
CI/PROD image)
+    """
+    result = run_command(
+        ["docker", "compose", "--project-name", compose_project_name, 
"config", "--images"],
+        env=env,
+        text=True,
+        capture_output=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        return []
+    images_to_pull = []
+    for image in dict.fromkeys(line.strip() for line in 
result.stdout.splitlines() if line.strip()):
+        if image in skip_images:
+            continue
+        image_present = run_command(
+            ["docker", "image", "inspect", image],
+            text=True,
+            capture_output=True,
+            check=False,
+        )
+        if image_present.returncode != 0:
+            images_to_pull.append(image)
+    return images_to_pull
+
+
+def pull_images_with_retries(
+    compose_project_name: str,
+    env: dict[str, str],
+    skip_images: set[str] | None = None,
+    attempts: int = IMAGE_PULL_ATTEMPTS,
+) -> bool:
+    """
+    Pulls the third-party images the compose project needs, retrying with a 
growing backoff.
+
+    Registries (Docker Hub in particular) regularly time out or throttle CI 
runners. Left to
+    `docker compose run`, a single such blip fails the whole test job before 
any test executes,
+    and recovering from it costs a full re-run of the suite. Pulling up front 
instead retries
+    the operation that actually failed.
+
+    :param compose_project_name: name of the docker compose project
+    :param env: environment variables to resolve the compose files with
+    :param skip_images: images that should never be pulled (the locally built 
CI/PROD image)
+    :param attempts: how many times to attempt pulling each image
+
+    :return: True if every image needed is available locally
+    """
+    images = get_images_to_pull(compose_project_name, env, skip_images or 
set())
+    if not images:
+        return True
+    console_print(f"[info]Pulling {len(images)} image(s) before running the 
tests: {' '.join(images)}[/]")
+    all_pulled = True
+    for image in images:
+        for attempt in range(1, attempts + 1):
+            if run_command(["docker", "pull", image], env=env, 
check=False).returncode == 0:
+                break
+            if attempt == attempts:
+                console_print(f"[warning]Could not pull {image} in {attempts} 
attempts.[/]")
+                all_pulled = False
+                break
+            backoff = IMAGE_PULL_BACKOFF_SECONDS * attempt
+            console_print(
+                f"[warning]Failed to pull {image} (attempt 
{attempt}/{attempts}). Retrying in {backoff}s.[/]"
+            )
+            time.sleep(backoff)
+    return all_pulled
+
+
 def remove_docker_networks(networks: list[str] | None = None) -> None:
     """
     Removes specified docker networks. If no networks are specified, it 
removes all networks created by breeze.
diff --git a/dev/breeze/tests/test_docker_command_utils.py 
b/dev/breeze/tests/test_docker_command_utils.py
index 76293c350b0..dc373cf8c40 100644
--- a/dev/breeze/tests/test_docker_command_utils.py
+++ b/dev/breeze/tests/test_docker_command_utils.py
@@ -30,7 +30,9 @@ from airflow_breeze.utils.docker_command_utils import (
     check_docker_version,
     discover_running_compose_projects,
     enter_shell,
+    get_images_to_pull,
     is_known_breeze_compose_project,
+    pull_images_with_retries,
 )
 
 
@@ -471,3 +473,83 @@ def 
test_enter_shell_openlineage_rejects_non_postgres_backend(
         enter_shell(shell_params)
     assert exc_info.value.code == 1
     mock_run_command.assert_not_called()
+
+
+CI_IMAGE = "ghcr.io/apache/airflow/main/ci/python3.10:latest"
+
+
+def _fake_docker_calls(present_images: set[str], failing_pulls: dict[str, 
int]):
+    """
+    Builds a run_command side effect emulating docker for the pull helpers.
+
+    :param present_images: images `docker image inspect` reports as already 
available
+    :param failing_pulls: how many times `docker pull` fails for a given image 
before succeeding
+    """
+    remaining_failures = dict(failing_pulls)
+
+    def _run_command(cmd, **kwargs):
+        result = mock.MagicMock()
+        result.returncode = 0
+        if cmd[:2] == ["docker", "compose"]:
+            result.stdout = 
f"{CI_IMAGE}\npostgres:17\notel/opentelemetry-collector-contrib:0.155.0\n"
+        elif cmd[:3] == ["docker", "image", "inspect"]:
+            result.returncode = 0 if cmd[3] in present_images else 1
+        elif cmd[:2] == ["docker", "pull"]:
+            if remaining_failures.get(cmd[2], 0) > 0:
+                remaining_failures[cmd[2]] -= 1
+                result.returncode = 1
+        return result
+
+    return _run_command
+
+
[email protected]("airflow_breeze.utils.docker_command_utils.run_command")
+def test_get_images_to_pull_skips_present_and_skipped_images(mock_run_command):
+    mock_run_command.side_effect = 
_fake_docker_calls(present_images={"postgres:17"}, failing_pulls={})
+
+    images = get_images_to_pull("breeze-test", env={}, skip_images={CI_IMAGE})
+
+    assert images == ["otel/opentelemetry-collector-contrib:0.155.0"]
+
+
[email protected]("airflow_breeze.utils.docker_command_utils.run_command")
+def 
test_get_images_to_pull_returns_nothing_when_compose_config_fails(mock_run_command):
+    mock_run_command.return_value = mock.MagicMock(returncode=1, stdout="")
+
+    assert get_images_to_pull("breeze-test", env={}, skip_images=set()) == []
+
+
[email protected]("airflow_breeze.utils.docker_command_utils.time.sleep")
[email protected]("airflow_breeze.utils.docker_command_utils.run_command")
+def 
test_pull_images_with_retries_recovers_from_transient_failures(mock_run_command,
 mock_sleep):
+    otel_image = "otel/opentelemetry-collector-contrib:0.155.0"
+    mock_run_command.side_effect = _fake_docker_calls(
+        present_images={"postgres:17"}, failing_pulls={otel_image: 2}
+    )
+
+    assert pull_images_with_retries("breeze-test", env={}, 
skip_images={CI_IMAGE}) is True
+    assert mock_sleep.call_args_list == [call(15), call(30)]
+
+
[email protected]("airflow_breeze.utils.docker_command_utils.time.sleep")
[email protected]("airflow_breeze.utils.docker_command_utils.run_command")
+def 
test_pull_images_with_retries_gives_up_after_all_attempts(mock_run_command, 
mock_sleep):
+    otel_image = "otel/opentelemetry-collector-contrib:0.155.0"
+    mock_run_command.side_effect = _fake_docker_calls(
+        present_images={"postgres:17"}, failing_pulls={otel_image: 99}
+    )
+
+    assert pull_images_with_retries("breeze-test", env={}, 
skip_images={CI_IMAGE}, attempts=3) is False
+    assert mock_sleep.call_args_list == [call(15), call(30)]
+
+
[email protected]("airflow_breeze.utils.docker_command_utils.time.sleep")
[email protected]("airflow_breeze.utils.docker_command_utils.run_command")
+def 
test_pull_images_with_retries_does_not_pull_when_all_images_are_present(mock_run_command,
 mock_sleep):
+    mock_run_command.side_effect = _fake_docker_calls(
+        present_images={"postgres:17", 
"otel/opentelemetry-collector-contrib:0.155.0"}, failing_pulls={}
+    )
+
+    assert pull_images_with_retries("breeze-test", env={}, 
skip_images={CI_IMAGE}) is True
+    assert not any(c.args[0][:2] == ["docker", "pull"] for c in 
mock_run_command.call_args_list)
+    mock_sleep.assert_not_called()
diff --git a/scripts/ci/testing/run_integration_tests_with_retry.sh 
b/scripts/ci/testing/run_integration_tests_with_retry.sh
index afb3003ecef..8d37ffe76cc 100755
--- a/scripts/ci/testing/run_integration_tests_with_retry.sh
+++ b/scripts/ci/testing/run_integration_tests_with_retry.sh
@@ -27,6 +27,7 @@ fi
 
 TEST_GROUP=${1}
 INTEGRATION=${2}
+RETRY_DELAY_SECONDS=${RETRY_DELAY_SECONDS:-60}
 
 breeze down
 set +e
@@ -39,6 +40,10 @@ if [[ ${RESULT} != "0" ]]; then
     echo
     echo "This could be due to a flaky test, re-running once to re-check it 
After restarting docker."
     echo
+    # Both attempts used to land within a minute of each other, so any 
registry or network
+    # outage lasting longer than that failed the job even though it was not 
our failure.
+    echo "Waiting ${RETRY_DELAY_SECONDS} seconds before retrying."
+    sleep "${RETRY_DELAY_SECONDS}"
     sudo service docker restart
     breeze down
     set +e

Reply via email to