From 8806cc3dea8bd81735edbe906f4618f8ca397224 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Mon, 3 Oct 2022 11:22:28 -0700 Subject: [PATCH 1/3] Deprecate use of core kube_client in PodManager We are no longer using this in KPO (or anywhere else in the codebase for that matter) now that KPO uses K8s hook to generate the client. However, it seems we neglected to deprecate this call to get_kube_client as a fallback. Since this deprecation warning won't go out until provider version 5.0, we have to wait until next major to actually remove (and i add a test to remind us to do so). --- .../cncf/kubernetes/utils/pod_manager.py | 11 ++++++++++- .../cncf/kubernetes/utils/test_pod_manager.py | 18 ++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/airflow/providers/cncf/kubernetes/utils/pod_manager.py b/airflow/providers/cncf/kubernetes/utils/pod_manager.py index c65150b4181d4..a1fd511761d6e 100644 --- a/airflow/providers/cncf/kubernetes/utils/pod_manager.py +++ b/airflow/providers/cncf/kubernetes/utils/pod_manager.py @@ -121,7 +121,16 @@ def __init__( :param cluster_context: context of the cluster """ super().__init__() - self._client = kube_client or get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context) + if kube_client: + self._client = kube_client + else: + self._client = get_kube_client(in_cluster=in_cluster, cluster_context=cluster_context) + warnings.warn( + "`kube_client` not supplied to PodManager. " + "This will be a required argument in a future release. " + "Please use KubernetesHook to create the client before calling.", + DeprecationWarning, + ) self._watch = watch.Watch() def run_pod_async(self, pod: V1Pod, **kwargs) -> V1Pod: diff --git a/tests/providers/cncf/kubernetes/utils/test_pod_manager.py b/tests/providers/cncf/kubernetes/utils/test_pod_manager.py index 5299b6a4be778..851c9aa1b316d 100644 --- a/tests/providers/cncf/kubernetes/utils/test_pod_manager.py +++ b/tests/providers/cncf/kubernetes/utils/test_pod_manager.py @@ -331,6 +331,24 @@ def test_fetch_container_running_follow( assert ret.last_log_time == DateTime(2021, 1, 1, tzinfo=Timezone('UTC')) assert ret.running is exp_running + def test_pod_manager_get_client_call_deprecation(self): + """Ensure that kube_client.get_kube_client is removed from pod manager in provider 6.0.""" + try: + from airflow.providers.cncf.kubernetes.utils.pod_manager import get_kube_client # noqa + except ImportError: + raise Exception( + "You must remove this test. It only exists to remind us to remove `get_kube_client`." + ) + from airflow.providers_manager import ProvidersManager + + version = ProvidersManager().providers['apache-airflow-providers-cncf-kubernetes'].version + version_tup = tuple(map(int, version.split('.'))) + if version_tup >= (6, 0): + raise Exception( + "You must now remove `get_kube_client` from PodManager " + "and make kube_client a required argument." + ) + def params_for_test_container_is_running(): """The `container_is_running` method is designed to handle an assortment of bad objects From 1742747a4be18f0b92e476da3cd212139c63c256 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Mon, 3 Oct 2022 11:43:29 -0700 Subject: [PATCH 2/3] use semver parser --- tests/providers/cncf/kubernetes/utils/test_pod_manager.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/providers/cncf/kubernetes/utils/test_pod_manager.py b/tests/providers/cncf/kubernetes/utils/test_pod_manager.py index 851c9aa1b316d..49eebda650f22 100644 --- a/tests/providers/cncf/kubernetes/utils/test_pod_manager.py +++ b/tests/providers/cncf/kubernetes/utils/test_pod_manager.py @@ -22,6 +22,7 @@ import pendulum import pytest +import semver from kubernetes.client.rest import ApiException from pendulum import DateTime from pendulum.tz.timezone import Timezone @@ -341,9 +342,8 @@ def test_pod_manager_get_client_call_deprecation(self): ) from airflow.providers_manager import ProvidersManager - version = ProvidersManager().providers['apache-airflow-providers-cncf-kubernetes'].version - version_tup = tuple(map(int, version.split('.'))) - if version_tup >= (6, 0): + info = ProvidersManager().providers['apache-airflow-providers-cncf-kubernetes'] + if semver.VersionInfo.parse(info.version) >= (6, 0): raise Exception( "You must now remove `get_kube_client` from PodManager " "and make kube_client a required argument." From ebac19da1b9b3cc6493e3592a0bcd508f40f3289 Mon Sep 17 00:00:00 2001 From: Daniel Standish <15932138+dstandish@users.noreply.github.com> Date: Mon, 10 Oct 2022 15:26:55 -0700 Subject: [PATCH 3/3] reduce some boilerplate --- .../cncf/kubernetes/utils/test_pod_manager.py | 11 ++--- tests/test_utils/providers.py | 48 +++++++++++++++++++ 2 files changed, 52 insertions(+), 7 deletions(-) create mode 100644 tests/test_utils/providers.py diff --git a/tests/providers/cncf/kubernetes/utils/test_pod_manager.py b/tests/providers/cncf/kubernetes/utils/test_pod_manager.py index 49eebda650f22..9cb68b8897e5d 100644 --- a/tests/providers/cncf/kubernetes/utils/test_pod_manager.py +++ b/tests/providers/cncf/kubernetes/utils/test_pod_manager.py @@ -22,7 +22,6 @@ import pendulum import pytest -import semver from kubernetes.client.rest import ApiException from pendulum import DateTime from pendulum.tz.timezone import Timezone @@ -30,6 +29,7 @@ from airflow.exceptions import AirflowException from airflow.providers.cncf.kubernetes.utils.pod_manager import PodManager, PodPhase, container_is_running +from tests.test_utils.providers import get_provider_version, object_exists class TestPodManager: @@ -334,16 +334,13 @@ def test_fetch_container_running_follow( def test_pod_manager_get_client_call_deprecation(self): """Ensure that kube_client.get_kube_client is removed from pod manager in provider 6.0.""" - try: - from airflow.providers.cncf.kubernetes.utils.pod_manager import get_kube_client # noqa - except ImportError: + kube_client_path = 'airflow.providers.cncf.kubernetes.utils.pod_manager.get_kube_client' + if not object_exists(kube_client_path): raise Exception( "You must remove this test. It only exists to remind us to remove `get_kube_client`." ) - from airflow.providers_manager import ProvidersManager - info = ProvidersManager().providers['apache-airflow-providers-cncf-kubernetes'] - if semver.VersionInfo.parse(info.version) >= (6, 0): + if get_provider_version('apache-airflow-providers-cncf-kubernetes') >= (6, 0): raise Exception( "You must now remove `get_kube_client` from PodManager " "and make kube_client a required argument." diff --git a/tests/test_utils/providers.py b/tests/test_utils/providers.py new file mode 100644 index 0000000000000..84a89bef81cd8 --- /dev/null +++ b/tests/test_utils/providers.py @@ -0,0 +1,48 @@ +# 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 semver + + +def object_exists(path: str): + """Returns true if importable python object is there.""" + from airflow.utils.module_loading import import_string + + try: + import_string(path) + return True + except ImportError: + return False + + +def get_provider_version(provider_name): + """ + Returns provider version given provider package name. + + Example:: + if provider_version('apache-airflow-providers-cncf-kubernetes') >= (6, 0): + raise Exception( + "You must now remove `get_kube_client` from PodManager " + "and make kube_client a required argument." + ) + """ + from airflow.providers_manager import ProvidersManager + + info = ProvidersManager().providers[provider_name] + return semver.VersionInfo.parse(info.version)