This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new da53196d812 Time-box and retry pre-extras SDK downloads (#71544)
da53196d812 is described below
commit da53196d812dd7245860e3d4281c8f3ab55bf016
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sat Oct 3 08:40:12 2026 +0200
Time-box and retry pre-extras SDK downloads (#71544)
A lowest-dependency providers job failed because IBM's download server
stopped answering: with no socket timeout, each of the five routes (DNS
plus the four published anycast IPs) sat in the kernel's TCP connect
timeout, and the single pass over them had no retry, so a transient
upstream outage took the whole job down after minutes of waiting.
Third-party servers hosting these SDKs are the least reliable part of
the run, and an outage measured in seconds should not cost a CI job.
---
contributing-docs/12_provider_distributions.rst | 6 +-
scripts/in_container/run_pre_extras_install.py | 64 ++++++----
.../in_container/test_run_pre_extras_install.py | 142 +++++++++++++++++++++
3 files changed, 188 insertions(+), 24 deletions(-)
diff --git a/contributing-docs/12_provider_distributions.rst
b/contributing-docs/12_provider_distributions.rst
index c7a234aaf05..b6be6afa537 100644
--- a/contributing-docs/12_provider_distributions.rst
+++ b/contributing-docs/12_provider_distributions.rst
@@ -206,7 +206,11 @@ must succeed first. Use the per-provider
pre-extras-install manifest:
Each ``downloads`` entry may also include ``fallback_ips`` (optional list
of IPv4 or
IPv6 address strings). The interpreter tries the URL with normal DNS
resolution first;
only on connection or resolution failure does it retry the same URL with
each listed
- IP, in order, by temporarily overriding ``socket.getaddrinfo`` for the
hostname. The
+ IP, in order, by temporarily overriding ``socket.getaddrinfo`` for the
hostname. Every
+ attempt uses a short socket timeout, and the whole set of routes is
retried in rounds
+ (see ``DOWNLOAD_TIMEOUT_SECONDS`` and ``DOWNLOAD_ROUNDS`` in the
interpreter), so an
+ unreachable or stalled upstream costs seconds per attempt rather than the
kernel's
+ multi-minute TCP connect timeout. A checksum mismatch is never retried.
The
TLS SNI and certificate verification stay bound to the URL hostname, and
the
``sha256`` check still runs end-to-end on whichever attempt succeeds, so
a fallback
entry only changes *which IP is dialled*, not what is trusted. Use this
when the
diff --git a/scripts/in_container/run_pre_extras_install.py
b/scripts/in_container/run_pre_extras_install.py
index cfd8c3db383..1b1f2e38658 100644
--- a/scripts/in_container/run_pre_extras_install.py
+++ b/scripts/in_container/run_pre_extras_install.py
@@ -39,6 +39,7 @@ import socket
import sys
import tarfile
import tempfile
+import time
import urllib.error
import urllib.request
import zipfile
@@ -51,6 +52,14 @@ import yaml
PROVIDERS_ROOT = Path("/opt/airflow/providers")
ALLOWED_EXTRACT_PREFIXES = ("/opt/", "/tmp/")
+# Socket timeout for a single connect or read. Without it, urllib inherits the
kernel's
+# TCP connect timeout (over two minutes), so an unreachable upstream burned
the whole
+# job before the first retry. Short timeouts plus rounds of retries recover
from a
+# blackholed IP or a stalled transfer quickly; a healthy server never comes
close to it.
+DOWNLOAD_TIMEOUT_SECONDS = 20
+# Each round tries the URL as resolved by DNS and then every fallback IP in
turn.
+DOWNLOAD_ROUNDS = 3
+SLEEP_BETWEEN_ROUNDS_SECONDS = 10
ENV_NAME_RE = re.compile(r"^[A-Z][A-Z0-9_]*$")
SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
ALLOWED_TOP_LEVEL_KEYS = {"downloads", "env"}
@@ -145,7 +154,10 @@ def override_dns(hostname: str, ip: str) -> Iterator[None]:
def _attempt_download(url: str, expected_sha256: str, dest: Path) -> None:
digest = hashlib.sha256()
- with urllib.request.urlopen(url) as response, dest.open("wb") as out:
+ with (
+ urllib.request.urlopen(url, timeout=DOWNLOAD_TIMEOUT_SECONDS) as
response,
+ dest.open("wb") as out,
+ ):
while True:
chunk = response.read(64 * 1024)
if not chunk:
@@ -163,34 +175,40 @@ def download_with_checksum(
dest: Path,
fallback_ips: list[str] | None = None,
) -> None:
- print(f"Downloading {url}")
- try:
- _attempt_download(url, expected_sha256, dest)
- return
- except (urllib.error.URLError, OSError) as primary_err:
- if not fallback_ips:
- raise
- print(
- f"Primary download failed ({type(primary_err).__name__}:
{primary_err}); "
- f"trying {len(fallback_ips)} fallback IP(s)"
- )
-
- hostname = urlparse(url).hostname
- if not hostname:
- fail(f"cannot extract hostname from url {url!r} for fallback
resolution")
+ """Download `url` to `dest`, retrying the whole set of routes in rounds.
+
+ A checksum mismatch is not retried - that means the manifest disagrees
with what
+ upstream serves, which no amount of retrying fixes.
+ """
+ fallback_routes: list[tuple[str, str]] = []
+ if fallback_ips:
+ hostname = urlparse(url).hostname
+ if not hostname:
+ fail(f"cannot extract hostname from url {url!r} for fallback
resolution")
+ fallback_routes = [(hostname, ip) for ip in fallback_ips]
last_err: BaseException | None = None
- for ip in fallback_ips:
- print(f" Retrying with {hostname} -> {ip}")
+ for round_number in range(1, DOWNLOAD_ROUNDS + 1):
+ print(f"Downloading {url} (round {round_number} of {DOWNLOAD_ROUNDS})")
try:
- with override_dns(hostname, ip):
- _attempt_download(url, expected_sha256, dest)
- print(f" Success via {ip}")
+ _attempt_download(url, expected_sha256, dest)
return
except (urllib.error.URLError, OSError) as e:
- print(f" {ip} failed: {type(e).__name__}: {e}")
+ print(f" failed: {type(e).__name__}: {e}")
last_err = e
- continue
+ for hostname, ip in fallback_routes:
+ print(f" Retrying with {hostname} -> {ip}")
+ try:
+ with override_dns(hostname, ip):
+ _attempt_download(url, expected_sha256, dest)
+ print(f" Success via {ip}")
+ return
+ except (urllib.error.URLError, OSError) as e:
+ print(f" {ip} failed: {type(e).__name__}: {e}")
+ last_err = e
+ if round_number < DOWNLOAD_ROUNDS:
+ print(f" Sleeping {SLEEP_BETWEEN_ROUNDS_SECONDS}s before the next
round")
+ time.sleep(SLEEP_BETWEEN_ROUNDS_SECONDS)
fail(f"all download attempts failed for {url}; last error: {last_err}")
diff --git a/scripts/tests/in_container/test_run_pre_extras_install.py
b/scripts/tests/in_container/test_run_pre_extras_install.py
new file mode 100644
index 00000000000..edaaf6ab787
--- /dev/null
+++ b/scripts/tests/in_container/test_run_pre_extras_install.py
@@ -0,0 +1,142 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+import hashlib
+import io
+import socket
+from unittest import mock
+
+import pytest
+import run_pre_extras_install as m
+
+ARCHIVE_CONTENT = b"pretend this is an SDK tarball"
+ARCHIVE_SHA256 = hashlib.sha256(ARCHIVE_CONTENT).hexdigest()
+URL = "https://example.com/sdk/9.4.0.0-Some-SDK-LinuxX64.tar.gz";
+
+
[email protected]
+def no_sleep(monkeypatch):
+ sleeps: list[float] = []
+ monkeypatch.setattr(m.time, "sleep", sleeps.append)
+ return sleeps
+
+
+class TestAttemptDownload:
+ @mock.patch("urllib.request.urlopen")
+ def test_uses_a_short_socket_timeout(self, mock_urlopen, tmp_path):
+ mock_urlopen.return_value = io.BytesIO(ARCHIVE_CONTENT)
+
+ m._attempt_download(URL, ARCHIVE_SHA256, tmp_path / "archive.tar.gz")
+
+ assert mock_urlopen.call_args.kwargs["timeout"] ==
m.DOWNLOAD_TIMEOUT_SECONDS
+ assert (tmp_path / "archive.tar.gz").read_bytes() == ARCHIVE_CONTENT
+
+
+class TestDownloadWithChecksum:
+ """The routes tried are recorded as either the url (DNS resolution) or the
fallback IP."""
+
+ @pytest.fixture(autouse=True)
+ def _record_routes(self, monkeypatch):
+ self.routes: list[str] = []
+ self.succeed_on: str | None = None
+
+ def fake_attempt(url, expected_sha256, dest):
+ route = socket.getaddrinfo(urlparse_host(url), 443)[0][4][0]
+ self.routes.append(route)
+ if route != self.succeed_on:
+ raise TimeoutError("timed out")
+ dest.write_bytes(ARCHIVE_CONTENT)
+
+ def urlparse_host(url):
+ return m.urlparse(url).hostname
+
+ # Resolve the hostname to itself unless override_dns is in effect, so
a route is
+ # identified by the fallback IP that was patched in - or by the
hostname otherwise.
+ monkeypatch.setattr(
+ m.socket,
+ "getaddrinfo",
+ lambda host, port, *args, **kwargs: [
+ (m.socket.AF_INET, m.socket.SOCK_STREAM, m.socket.IPPROTO_TCP,
"", (host, port))
+ ],
+ )
+ monkeypatch.setattr(m, "_attempt_download", fake_attempt)
+
+ def test_retries_in_rounds_and_gives_up(self, tmp_path, no_sleep):
+ with pytest.raises(SystemExit):
+ m.download_with_checksum(URL, ARCHIVE_SHA256, tmp_path /
"archive.tar.gz")
+
+ assert self.routes == ["example.com"] * m.DOWNLOAD_ROUNDS
+ assert no_sleep == [m.SLEEP_BETWEEN_ROUNDS_SECONDS] *
(m.DOWNLOAD_ROUNDS - 1)
+
+ def test_succeeds_on_a_later_round(self, tmp_path, no_sleep):
+ dest = tmp_path / "archive.tar.gz"
+ attempts = 0
+
+ def succeed_on_second_round(url, expected_sha256, dest):
+ nonlocal attempts
+ attempts += 1
+ if attempts < 2:
+ raise TimeoutError("timed out")
+ dest.write_bytes(ARCHIVE_CONTENT)
+
+ with mock.patch.object(m, "_attempt_download",
succeed_on_second_round):
+ m.download_with_checksum(URL, ARCHIVE_SHA256, dest)
+
+ assert attempts == 2
+ assert dest.read_bytes() == ARCHIVE_CONTENT
+ assert no_sleep == [m.SLEEP_BETWEEN_ROUNDS_SECONDS]
+
+ def test_tries_every_fallback_ip_before_the_next_round(self, tmp_path,
no_sleep):
+ with pytest.raises(SystemExit):
+ m.download_with_checksum(
+ URL, ARCHIVE_SHA256, tmp_path / "archive.tar.gz",
fallback_ips=["10.0.0.1", "10.0.0.2"]
+ )
+
+ assert self.routes == ["example.com", "10.0.0.1", "10.0.0.2"] *
m.DOWNLOAD_ROUNDS
+
+ def test_stops_at_the_fallback_ip_that_works(self, tmp_path, no_sleep):
+ self.succeed_on = "10.0.0.2"
+ dest = tmp_path / "archive.tar.gz"
+
+ m.download_with_checksum(URL, ARCHIVE_SHA256, dest,
fallback_ips=["10.0.0.1", "10.0.0.2"])
+
+ assert self.routes == ["example.com", "10.0.0.1", "10.0.0.2"]
+ assert dest.read_bytes() == ARCHIVE_CONTENT
+ assert no_sleep == []
+
+ def test_does_not_retry_a_checksum_mismatch(self, tmp_path, no_sleep):
+ def wrong_checksum(url, expected_sha256, dest):
+ self.routes.append(url)
+ m.fail(f"sha256 mismatch for {url}")
+
+ with mock.patch.object(m, "_attempt_download", wrong_checksum):
+ with pytest.raises(SystemExit):
+ m.download_with_checksum(
+ URL, ARCHIVE_SHA256, tmp_path / "archive.tar.gz",
fallback_ips=["10.0.0.1"]
+ )
+
+ assert self.routes == [URL]
+ assert no_sleep == []
+
+ def
test_rejects_a_url_without_a_hostname_when_fallbacks_are_configured(self,
tmp_path, no_sleep):
+ with pytest.raises(SystemExit):
+ m.download_with_checksum(
+ "https:///no-host.tar.gz";, ARCHIVE_SHA256, tmp_path /
"a.tar.gz", fallback_ips=["10.0.0.1"]
+ )
+
+ assert self.routes == []
