This is an automated email from the ASF dual-hosted git repository.
henry3260 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 8cc830d4ef9 Fix Kubernetes TCP keepalive not being applied with
urllib3 2.x (#74921)
8cc830d4ef9 is described below
commit 8cc830d4ef93a51cebd4018171d7a2f76cadd3b0
Author: Shahar Epstein <[email protected]>
AuthorDate: Wed Oct 21 09:36:53 2026 +0300
Fix Kubernetes TCP keepalive not being applied with urllib3 2.x (#74921)
* Use Configuration.socket_options instead of monkey-patching urllib3 in
cncf-kubernetes
In urllib3 2.x, HTTPConnection.__init__ binds socket_options to
default_socket_options as a default argument at import time, so
reassigning the class attribute later never reaches new connections and
[kubernetes_executor] enable_tcp_keepalive was silently a no-op.
Pass the options through the Kubernetes client's
Configuration.socket_options,
which RESTClientObject forwards to urllib3.PoolManager.
Closes: #68396
Supersedes: #68557
* Address review findings for cncf.kubernetes TCP keepalive fix
- Mark the google cncf.kubernetes floor with 'use next version'.
- enable_tcp_keepalive: drop the urllib3 monkey-patch fallback; warn and
return when Configuration has no socket_options.
- Merge keepalive options into existing socket_options (or urllib3
defaults) without duplicates.
- Deprecated _enable_tcp_keepalive() now sets the default Configuration.
- Tidy docstrings and revert the spelling_wordlist additions.
- Add tests for merging, idempotency, unsupported clients, get_kube_client,
_TimeoutK8sApiClient, the deprecated helper and the hook; use autospec in hook
tests.
* Require kubernetes client 36.0.1 for TCP keepalive support
Configuration.socket_options, the only way to apply keepalive options with
urllib3 2.x, does not exist in kubernetes 35.0.0, so on that floor keepalive
could only ever be logged as unavailable and the lowest-dependency tests failed
on it.
* Parametrize cncf.kubernetes TCP keepalive client tests
Cover custom tcp_keep_idle/intvl/cnt values, and exercise
_TimeoutK8sApiClient with and without a supplied configuration and
disable_verify_ssl, asserting the global default stays unchanged.
* Align TCP keepalive kwarg naming in cncf.kubernetes client
Rename the _TimeoutK8sApiClient and KubernetesHook kwarg
should_enable_tcp_keepalive to enable_tcp_keepalive, matching
disable_verify_ssl and the GKE hook. The unreleased public helper
becomes configure_tcp_keepalive so the kwarg does not shadow it.
The socket-options test now derives urllib3's defaults instead of
hard-coding TCP_NODELAY.
Also restore uv.lock revision 5 from main, which an older local uv
had rewritten to 3.
---------
Co-authored-by: anmolxlight <[email protected]>
---
providers/cncf/kubernetes/README.rst | 2 +-
providers/cncf/kubernetes/docs/index.rst | 2 +-
providers/cncf/kubernetes/pyproject.toml | 9 +-
.../providers/cncf/kubernetes/hooks/kubernetes.py | 19 +++-
.../providers/cncf/kubernetes/kube_client.py | 86 ++++++++------
.../unit/cncf/kubernetes/hooks/test_kubernetes.py | 38 ++++++-
.../tests/unit/cncf/kubernetes/test_client.py | 124 +++++++++++++++++++--
providers/google/pyproject.toml | 2 +-
.../google/cloud/hooks/kubernetes_engine.py | 4 +-
.../google/cloud/hooks/test_kubernetes_engine.py | 2 +-
uv.lock | 2 +-
11 files changed, 230 insertions(+), 60 deletions(-)
diff --git a/providers/cncf/kubernetes/README.rst
b/providers/cncf/kubernetes/README.rst
index 75da4158d35..53fd9ce1799 100644
--- a/providers/cncf/kubernetes/README.rst
+++ b/providers/cncf/kubernetes/README.rst
@@ -59,7 +59,7 @@ PIP package Version required
``asgiref`` ``>=3.5.2; python_version <
"3.14"``
``asgiref`` ``>=3.11.1; python_version >=
"3.14"``
``cryptography`` ``>=44.0.3``
-``kubernetes`` ``>=35.0.0,!=36.0.0,<37.0.0``
+``kubernetes`` ``>=36.0.1,<37.0.0``
``urllib3`` ``>=2.1.0,!=2.6.0``
``kubernetes_asyncio`` ``>=32.0.0,<37.0.0``
==========================================
======================================
diff --git a/providers/cncf/kubernetes/docs/index.rst
b/providers/cncf/kubernetes/docs/index.rst
index f7aaafba334..1bc453b2b85 100644
--- a/providers/cncf/kubernetes/docs/index.rst
+++ b/providers/cncf/kubernetes/docs/index.rst
@@ -118,7 +118,7 @@ PIP package Version required
``asgiref`` ``>=3.5.2; python_version <
"3.14"``
``asgiref`` ``>=3.11.1; python_version >=
"3.14"``
``cryptography`` ``>=44.0.3``
-``kubernetes`` ``>=35.0.0,!=36.0.0,<37.0.0``
+``kubernetes`` ``>=36.0.1,<37.0.0``
``urllib3`` ``>=2.1.0,!=2.6.0``
``kubernetes_asyncio`` ``>=32.0.0,<37.0.0``
==========================================
======================================
diff --git a/providers/cncf/kubernetes/pyproject.toml
b/providers/cncf/kubernetes/pyproject.toml
index 40d8cc655b8..0aabf10dd8a 100644
--- a/providers/cncf/kubernetes/pyproject.toml
+++ b/providers/cncf/kubernetes/pyproject.toml
@@ -76,11 +76,10 @@ dependencies = [
# limiting minimum airflow version supported in cncf.kubernetes provider,
due to the
# potential breaking changes in Airflow Core as well (kubernetes is added
as extra, so Airflow
# core is not hard-limited via install-requirements, only by extra).
- # 36.0.0 is excluded because it shipped an in-cluster auth regression
- # (https://github.com/kubernetes-client/python/issues/2582), fixed in
36.0.1. 35.0.0 is the only
- # 35.x release and ignores NO_PROXY
(https://github.com/kubernetes-client/python/issues/2520),
- # fixed in the 36.x line, so a proxied install resolves to 36.0.1+.
- "kubernetes>=35.0.0,!=36.0.0,<37.0.0",
+ # 36.0.1 is the minimum: Configuration.socket_options (needed for TCP
keepalive with urllib3 2.x)
+ # first appears in 36.x, and 36.0.0 shipped an in-cluster auth regression
+ # (https://github.com/kubernetes-client/python/issues/2582).
+ "kubernetes>=36.0.1,<37.0.0",
# Urllib 2.6.0 breaks kubernetes client because kubernetes client uses
deprecated in 2.0.0 and
# removed in 2.6.0 `getheaders()` call (instead of `headers` property.
# Tracked in https://github.com/kubernetes-client/python/issues/2477
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index 8c911363860..b4809db6cf3 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -38,7 +38,6 @@ from airflow.models import Connection
from airflow.providers.cncf.kubernetes.exceptions import KubernetesApiError,
KubernetesApiPermissionError
from airflow.providers.cncf.kubernetes.kube_client import (
_disable_verify_ssl,
- _enable_tcp_keepalive,
_TimeoutAsyncK8sApiClient,
_TimeoutK8sApiClient,
)
@@ -279,8 +278,7 @@ class KubernetesHook(BaseHook, PodOperatorHookProtocol):
if disable_verify_ssl is True:
_disable_verify_ssl()
- if disable_tcp_keepalive is not True:
- _enable_tcp_keepalive()
+ enable_tcp_keepalive = disable_tcp_keepalive is not True
if in_cluster:
self.log.debug("loading kube_config from: in_cluster
configuration")
@@ -289,6 +287,7 @@ class KubernetesHook(BaseHook, PodOperatorHookProtocol):
return _TimeoutK8sApiClient(
configuration=self.client_configuration,
disable_verify_ssl=disable_verify_ssl is True,
+ enable_tcp_keepalive=enable_tcp_keepalive,
)
if kubeconfig_path is not None:
@@ -302,6 +301,7 @@ class KubernetesHook(BaseHook, PodOperatorHookProtocol):
return _TimeoutK8sApiClient(
configuration=self.client_configuration,
disable_verify_ssl=disable_verify_ssl is True,
+ enable_tcp_keepalive=enable_tcp_keepalive,
)
if kubeconfig is not None:
@@ -320,6 +320,7 @@ class KubernetesHook(BaseHook, PodOperatorHookProtocol):
return _TimeoutK8sApiClient(
configuration=self.client_configuration,
disable_verify_ssl=disable_verify_ssl is True,
+ enable_tcp_keepalive=enable_tcp_keepalive,
)
if self.config_dict:
@@ -333,14 +334,21 @@ class KubernetesHook(BaseHook, PodOperatorHookProtocol):
return _TimeoutK8sApiClient(
configuration=self.client_configuration,
disable_verify_ssl=disable_verify_ssl is True,
+ enable_tcp_keepalive=enable_tcp_keepalive,
)
return self._get_default_client(
- cluster_context=cluster_context,
disable_verify_ssl=disable_verify_ssl
+ cluster_context=cluster_context,
+ disable_verify_ssl=disable_verify_ssl,
+ enable_tcp_keepalive=enable_tcp_keepalive,
)
def _get_default_client(
- self, *, cluster_context: str | None = None, disable_verify_ssl: bool
| None = None
+ self,
+ *,
+ cluster_context: str | None = None,
+ disable_verify_ssl: bool | None = None,
+ enable_tcp_keepalive: bool = False,
) -> client.ApiClient:
# if we get here, then no configuration has been supplied
# we should try in_cluster since that's most likely
@@ -359,6 +367,7 @@ class KubernetesHook(BaseHook, PodOperatorHookProtocol):
return _TimeoutK8sApiClient(
configuration=self.client_configuration,
disable_verify_ssl=disable_verify_ssl is True,
+ enable_tcp_keepalive=enable_tcp_keepalive,
)
@property
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py
index 6c3cad4c428..9b6acbfdb75 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/kube_client.py
@@ -19,10 +19,14 @@
from __future__ import annotations
import logging
+import socket
+import warnings
from typing import Any
import urllib3.util
+from urllib3.connection import HTTPConnection
+from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.common.compat.sdk import conf
log = logging.getLogger(__name__)
@@ -60,9 +64,10 @@ try:
"""
Wrapper around kubernetes sync ApiClient to set default timeout.
- When *disable_verify_ssl* is True the TLS certificate check is turned
off
- on the *client_configuration* that is passed (or on a fresh default
copy)
- so that callers do not need to repeat this logic at every call-site.
+ When *disable_verify_ssl* is True the TLS certificate check is turned
off, and when
+ *enable_tcp_keepalive* is True TCP keepalive is enabled, on the
*configuration*
+ that is passed (or on a fresh default copy) so that callers do not
need to repeat
+ this logic at every call-site.
"""
def __init__(
@@ -70,11 +75,15 @@ try:
configuration: client.Configuration | None = None,
*,
disable_verify_ssl: bool = False,
+ enable_tcp_keepalive: bool = False,
) -> None:
- if disable_verify_ssl:
+ if disable_verify_ssl or enable_tcp_keepalive:
if configuration is None:
configuration = client.Configuration.get_default_copy()
- configuration.verify_ssl = False
+ if disable_verify_ssl:
+ configuration.verify_ssl = False
+ if enable_tcp_keepalive:
+ configure_tcp_keepalive(configuration)
super().__init__(configuration=configuration)
def call_api(self, *args: Any, **kwargs: Any) -> Any:
@@ -115,53 +124,64 @@ except ImportError as e:
_import_err = e
-def _enable_tcp_keepalive() -> None:
+def configure_tcp_keepalive(configuration: Configuration) -> None:
"""
- Enable TCP keepalive mechanism.
-
- This prevents urllib3 connection to hang indefinitely when idle connection
- is time-outed on services like cloud load balancers or firewalls.
+ Enable TCP keepalive on the Kubernetes client ``configuration``.
- See https://github.com/apache/airflow/pull/11406 for detailed explanation.
+ This prevents connections from hanging indefinitely when an idle
connection is timed out
+ by services like cloud load balancers or firewalls. The options are set
through
+ ``Configuration.socket_options``, which the client passes on to its
urllib3 pool manager.
- Please ping @michalmisiewicz or @dimberman in the PR if you want to modify
this function.
+ See https://github.com/apache/airflow/pull/11406 for the original
discussion.
"""
- import socket
-
- from urllib3.connection import HTTPConnection, HTTPSConnection
-
tcp_keep_idle = conf.getint("kubernetes_executor", "tcp_keep_idle")
tcp_keep_intvl = conf.getint("kubernetes_executor", "tcp_keep_intvl")
tcp_keep_cnt = conf.getint("kubernetes_executor", "tcp_keep_cnt")
- socket_options = [(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)]
+ keepalive_options: list[tuple[int, int, int | bytes]] =
[(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)]
if hasattr(socket, "TCP_KEEPIDLE"):
- socket_options.append((socket.IPPROTO_TCP, socket.TCP_KEEPIDLE,
tcp_keep_idle))
+ keepalive_options.append((socket.IPPROTO_TCP, socket.TCP_KEEPIDLE,
tcp_keep_idle))
else:
log.debug("Unable to set TCP_KEEPIDLE on this platform")
if hasattr(socket, "TCP_KEEPINTVL"):
- socket_options.append((socket.IPPROTO_TCP, socket.TCP_KEEPINTVL,
tcp_keep_intvl))
+ keepalive_options.append((socket.IPPROTO_TCP, socket.TCP_KEEPINTVL,
tcp_keep_intvl))
else:
log.debug("Unable to set TCP_KEEPINTVL on this platform")
if hasattr(socket, "TCP_KEEPCNT"):
- socket_options.append((socket.IPPROTO_TCP, socket.TCP_KEEPCNT,
tcp_keep_cnt))
+ keepalive_options.append((socket.IPPROTO_TCP, socket.TCP_KEEPCNT,
tcp_keep_cnt))
else:
log.debug("Unable to set TCP_KEEPCNT on this platform")
- # Cast both the default options and our socket options
- socket_options_cast: list[tuple[int, int, int | bytes]] = [
- (level, opt, val) for level, opt, val in socket_options
- ]
- default_options_cast: list[tuple[int, int, int | bytes]] = [
- (level, opt, val) for level, opt, val in
HTTPSConnection.default_socket_options
- ]
+ base_options = configuration.socket_options
+ if base_options is None:
+ base_options = HTTPConnection.default_socket_options
+ overridden = {(level, optname) for level, optname, _ in keepalive_options}
+ configuration.socket_options = [
+ option for option in base_options if (option[0], option[1]) not in
overridden
+ ] + keepalive_options
+
+
+def _enable_tcp_keepalive(configuration: Configuration | None = None) -> None:
+ """
+ Enable TCP keepalive mechanism.
- # Then use the cast versions for both HTTPS and HTTP
- HTTPSConnection.default_socket_options = default_options_cast +
socket_options_cast
- HTTPConnection.default_socket_options = default_options_cast +
socket_options_cast
+ .. deprecated::
+ Use :func:`configure_tcp_keepalive` instead.
+ """
+ warnings.warn(
+ "_enable_tcp_keepalive is deprecated, use
configure_tcp_keepalive(configuration) instead.",
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
+ set_default = configuration is None
+ if configuration is None:
+ configuration = _get_default_configuration()
+ configure_tcp_keepalive(configuration)
+ if set_default:
+ Configuration.set_default(configuration)
def get_kube_client(
@@ -182,10 +202,10 @@ def get_kube_client(
if not has_kubernetes:
raise _import_err
- if conf.getboolean("kubernetes_executor", "enable_tcp_keepalive"):
- _enable_tcp_keepalive()
-
configuration = _get_default_configuration()
+
+ if conf.getboolean("kubernetes_executor", "enable_tcp_keepalive"):
+ configure_tcp_keepalive(configuration)
api_client_retry_configuration = conf.getjson(
"kubernetes_executor", "api_client_retry_configuration", fallback={}
)
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
index ff9ec0b8dd3..4fed8799e57 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/hooks/test_kubernetes.py
@@ -20,6 +20,7 @@ from __future__ import annotations
import asyncio
import json
import os
+import socket
import tempfile
from asyncio import Future
from unittest import mock
@@ -428,7 +429,7 @@ class TestKubernetesHook:
),
)
@patch("kubernetes.config.incluster_config.InClusterConfigLoader",
new=MagicMock())
- @patch(f"{HOOK_MODULE}._enable_tcp_keepalive")
+
@patch("airflow.providers.cncf.kubernetes.kube_client.configure_tcp_keepalive",
autospec=True)
def test_disable_tcp_keepalive(
self,
mock_enable,
@@ -445,6 +446,37 @@ class TestKubernetesHook:
assert mock_enable.called is expected
assert isinstance(api_conn, kubernetes.client.api_client.ApiClient)
+ @pytest.mark.platform("linux")
+ @pytest.mark.parametrize(
+ "config_source",
+ [
+ pytest.param("config_dict", id="config_dict"),
+ pytest.param("default_client", id="default_client"),
+ ],
+ )
+ @pytest.mark.parametrize(
+ ("disable_tcp_keepalive", "expected"),
+ [(True, False), (False, True)],
+ )
+ @patch("kubernetes.config.incluster_config.InClusterConfigLoader",
new=MagicMock())
+ @patch("kubernetes.config.kube_config.KubeConfigLoader", new=MagicMock())
+ @patch("kubernetes.config.kube_config.KubeConfigMerger", new=MagicMock())
+ def test_tcp_keepalive_reaches_client(self, config_source,
disable_tcp_keepalive, expected):
+ """The keepalive socket options must end up on the client
configuration, not only be requested."""
+ if config_source == "config_dict":
+ kubernetes_hook = KubernetesHook(
+ config_dict={"apiVersion": "v1", "kind": "Config"},
+ disable_tcp_keepalive=disable_tcp_keepalive,
+ )
+ else:
+ kubernetes_hook =
KubernetesHook(disable_tcp_keepalive=disable_tcp_keepalive)
+
+ api_conn = kubernetes_hook.get_conn()
+
+ keepalive = (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+ options = api_conn.configuration.socket_options or []
+ assert (keepalive in options) is expected
+
@pytest.mark.parametrize(
("config_path_param", "conn_id", "call_path"),
(
@@ -619,7 +651,9 @@ class TestKubernetesHook:
with mock.patch.dict("os.environ",
AIRFLOW_CONN_KUBERNETES_DEFAULT=conn_uri):
kubernetes_hook = KubernetesHook(conn_id="kubernetes_default")
kubernetes_hook.get_conn()
- mock_get_client.assert_called_with(cluster_context="test",
disable_verify_ssl=None)
+ mock_get_client.assert_called_with(
+ cluster_context="test", disable_verify_ssl=None,
enable_tcp_keepalive=True
+ )
assert kubernetes_hook.get_namespace() == "test"
def test_missing_default_connection_is_ok(self, remove_default_conn,
sdk_connection_not_found):
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_client.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_client.py
index 5bb18a0992f..19909341135 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_client.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/test_client.py
@@ -21,17 +21,27 @@ from unittest import mock
import pytest
from kubernetes.client import Configuration
-from urllib3.connection import HTTPConnection, HTTPSConnection
+from urllib3.connection import HTTPConnection
+from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.providers.cncf.kubernetes.kube_client import (
_disable_verify_ssl,
_enable_tcp_keepalive,
+ _TimeoutK8sApiClient,
+ configure_tcp_keepalive,
get_kube_client,
)
from tests_common.test_utils.config import conf_vars
[email protected]
+def restore_default_configuration():
+ original = Configuration.get_default_copy()
+ yield
+ Configuration.set_default(original)
+
+
class TestClient:
@mock.patch("airflow.providers.cncf.kubernetes.kube_client.config")
def test_load_cluster_config(self, config):
@@ -64,20 +74,118 @@ class TestClient:
assert client.api_client.configuration.ssl_ca_cert == "/path/to/ca.crt"
@pytest.mark.platform("linux")
- def test_enable_tcp_keepalive(self):
- socket_options = [
+ @pytest.mark.parametrize(
+ ("config_overrides", "expected_idle", "expected_intvl",
"expected_cnt"),
+ [
+ pytest.param({}, 120, 30, 6, id="defaults"),
+ pytest.param(
+ {
+ ("kubernetes_executor", "tcp_keep_idle"): "60",
+ ("kubernetes_executor", "tcp_keep_intvl"): "10",
+ ("kubernetes_executor", "tcp_keep_cnt"): "3",
+ },
+ 60,
+ 10,
+ 3,
+ id="custom",
+ ),
+ ],
+ )
+ def test_configure_tcp_keepalive(self, config_overrides, expected_idle,
expected_intvl, expected_cnt):
+ configuration = Configuration()
+ configuration.socket_options = None
+
+ with conf_vars(config_overrides):
+ configure_tcp_keepalive(configuration)
+
+ assert configuration.socket_options == [
+ *HTTPConnection.default_socket_options,
+ (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
+ (socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, expected_idle),
+ (socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, expected_intvl),
+ (socket.IPPROTO_TCP, socket.TCP_KEEPCNT, expected_cnt),
+ ]
+
+ @pytest.mark.platform("linux")
+ def test_configure_tcp_keepalive_preserves_existing_options(self):
+ configuration = Configuration()
+ configuration.socket_options = [
+ (socket.SOL_SOCKET, socket.SO_RCVBUF, 1234),
+ (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 0),
+ (socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 5),
+ ]
+
+ configure_tcp_keepalive(configuration)
+
+ assert configuration.socket_options == [
+ (socket.SOL_SOCKET, socket.SO_RCVBUF, 1234),
(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1),
(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 120),
(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, 30),
(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, 6),
]
- expected_http_connection_options =
HTTPConnection.default_socket_options + socket_options
- expected_https_connection_options =
HTTPSConnection.default_socket_options + socket_options
- _enable_tcp_keepalive()
+ @pytest.mark.platform("linux")
+ def test_configure_tcp_keepalive_is_idempotent(self):
+ configuration = Configuration()
+ configuration.socket_options = None
+
+ configure_tcp_keepalive(configuration)
+ first_call_options = list(configuration.socket_options)
+ configure_tcp_keepalive(configuration)
+
+ assert configuration.socket_options == first_call_options
+
+ @pytest.mark.platform("linux")
+ @pytest.mark.parametrize("enabled", [True, False])
+ @mock.patch("airflow.providers.cncf.kubernetes.kube_client.config",
autospec=True)
+ def test_get_kube_client_tcp_keepalive(self, mock_config, enabled):
+ with conf_vars({("kubernetes_executor", "enable_tcp_keepalive"):
str(enabled)}):
+ client = get_kube_client(in_cluster=False)
+
+ configuration_options = client.api_client.configuration.socket_options
+ pool_options =
client.api_client.rest_client.pool_manager.connection_pool_kw.get("socket_options")
+ keepalive = (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+ if enabled:
+ assert keepalive in configuration_options
+ assert keepalive in pool_options
+ else:
+ assert not configuration_options or keepalive not in
configuration_options
+ assert not pool_options or keepalive not in pool_options
+
+ @pytest.mark.platform("linux")
+ @pytest.mark.parametrize("supply_configuration", [True, False])
+ @pytest.mark.parametrize("disable_verify_ssl", [True, False])
+ @pytest.mark.parametrize("enabled", [True, False])
+ def test_timeout_api_client_configuration(self, enabled,
disable_verify_ssl, supply_configuration):
+ default = Configuration.get_default_copy()
+ configuration = Configuration() if supply_configuration else None
+ expected_verify_ssl = False if disable_verify_ssl else (configuration
or default).verify_ssl
+
+ api_client = _TimeoutK8sApiClient(
+ configuration,
+ disable_verify_ssl=disable_verify_ssl,
+ enable_tcp_keepalive=enabled,
+ )
+
+ if supply_configuration:
+ assert api_client.configuration is configuration
+ assert api_client.configuration.verify_ssl is expected_verify_ssl
+ keepalive = (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+ options = api_client.configuration.socket_options or []
+ assert (keepalive in options) is enabled
+ unchanged_default = Configuration.get_default_copy()
+ assert unchanged_default.socket_options == default.socket_options
+ assert unchanged_default.verify_ssl == default.verify_ssl
+
+ @pytest.mark.platform("linux")
+ def test_deprecated_enable_tcp_keepalive_sets_default_configuration(self,
restore_default_configuration):
+ with pytest.warns(AirflowProviderDeprecationWarning):
+ _enable_tcp_keepalive()
- assert HTTPConnection.default_socket_options ==
expected_http_connection_options
- assert HTTPSConnection.default_socket_options ==
expected_https_connection_options
+ options = Configuration.get_default_copy().socket_options
+ assert (socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) in options
+ assert (socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, 120) in options
def test_disable_verify_ssl(self):
configuration = Configuration()
diff --git a/providers/google/pyproject.toml b/providers/google/pyproject.toml
index f8d6d877b6e..03c30036d61 100644
--- a/providers/google/pyproject.toml
+++ b/providers/google/pyproject.toml
@@ -162,7 +162,7 @@ dependencies = [
"apache-airflow-providers-apache-beam>=6.2.2",
]
"cncf.kubernetes" = [
- "apache-airflow-providers-cncf-kubernetes>=10.22.0",
+ "apache-airflow-providers-cncf-kubernetes>=10.22.0", # use next version
]
"fab" = [
"apache-airflow-providers-fab>=2.0.0",
diff --git
a/providers/google/src/airflow/providers/google/cloud/hooks/kubernetes_engine.py
b/providers/google/src/airflow/providers/google/cloud/hooks/kubernetes_engine.py
index a721578132e..bbbc29f61e4 100644
---
a/providers/google/src/airflow/providers/google/cloud/hooks/kubernetes_engine.py
+++
b/providers/google/src/airflow/providers/google/cloud/hooks/kubernetes_engine.py
@@ -39,7 +39,7 @@ from kubernetes_asyncio.config.kube_config import FileOrData
from airflow import version
from airflow.providers.cncf.kubernetes.hooks.kubernetes import
AsyncKubernetesHook, KubernetesHook
-from airflow.providers.cncf.kubernetes.kube_client import _enable_tcp_keepalive
+from airflow.providers.cncf.kubernetes.kube_client import
configure_tcp_keepalive
from airflow.providers.common.compat.sdk import AirflowException
from airflow.providers.google.common.consts import CLIENT_INFO
from airflow.providers.google.common.hooks.base_google import (
@@ -76,7 +76,7 @@ class GKEClusterConnection:
configuration = self._get_config()
configuration.refresh_api_key_hook = self._refresh_api_key_hook
if self.enable_tcp_keepalive:
- _enable_tcp_keepalive()
+ configure_tcp_keepalive(configuration)
return client.ApiClient(configuration)
def _refresh_api_key_hook(self, configuration:
client.configuration.Configuration):
diff --git
a/providers/google/tests/unit/google/cloud/hooks/test_kubernetes_engine.py
b/providers/google/tests/unit/google/cloud/hooks/test_kubernetes_engine.py
index f3a1a416204..e9052934ccf 100644
--- a/providers/google/tests/unit/google/cloud/hooks/test_kubernetes_engine.py
+++ b/providers/google/tests/unit/google/cloud/hooks/test_kubernetes_engine.py
@@ -675,7 +675,7 @@ class TestGKEKubernetesHookPod:
(False, True),
),
)
- @mock.patch(GKE_STRING.format("_enable_tcp_keepalive"))
+ @mock.patch(GKE_STRING.format("configure_tcp_keepalive"), autospec=True)
def test_disable_tcp_keepalive(
self,
mock_enable,
diff --git a/uv.lock b/uv.lock
index ee3b8dcc89e..0d026101c5e 100644
--- a/uv.lock
+++ b/uv.lock
@@ -4554,7 +4554,7 @@ requires-dist = [
{ name = "asgiref", marker = "python_full_version < '3.14'", specifier =
">=3.5.2" },
{ name = "asgiref", marker = "python_full_version >= '3.14'", specifier =
">=3.11.1" },
{ name = "cryptography", specifier = ">=44.0.3" },
- { name = "kubernetes", specifier = ">=35.0.0,!=36.0.0,<37.0.0" },
+ { name = "kubernetes", specifier = ">=36.0.1,<37.0.0" },
{ name = "kubernetes-asyncio", specifier = ">=32.0.0,<37.0.0" },
{ name = "urllib3", specifier = ">=2.1.0,!=2.6.0" },
]
