Fix deprecated calls in cncf.kubernetes provider

airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py

@@ -32,7 +32,7 @@
 from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
     annotations_for_logging_task_metadata,
     annotations_to_key,
-    create_pod_id,
+    create_unique_id,
 )
 from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
 from airflow.utils.log.logging_mixin import LoggingMixin
@@ -413,7 +413,7 @@ def run_next(self, next_job: KubernetesJobType) -> None:
         pod = PodGenerator.construct_pod(
             namespace=self.namespace,
             scheduler_job_id=self.scheduler_job_id,
-            pod_id=create_pod_id(dag_id, task_id),
+            pod_id=create_unique_id(dag_id, task_id),
             dag_id=dag_id,
             task_id=task_id,
             kube_image=self.kube_config.kube_image,

airflow/providers/cncf/kubernetes/kubernetes_helper_functions.py

@@ -19,10 +19,10 @@
 import logging
 import secrets
 import string
-import warnings
 from typing import TYPE_CHECKING
 
 import pendulum
+from deprecated import deprecated
 from slugify import slugify
 
 from airflow.compat.functools import cache
@@ -59,6 +59,10 @@ def add_unique_suffix(*, name: str, rand_len: int = 8, max_len: int = POD_NAME_M
     return name[: max_len - len(suffix)].strip("-.") + suffix
 
 
+@deprecated(
+    reason="This function is deprecated. Please use `add_unique_suffix`",
+    category=AirflowProviderDeprecationWarning,
+)
 def add_pod_suffix(*, pod_name: str, rand_len: int = 8, max_len: int = POD_NAME_MAX_LENGTH) -> str:
     """Add random string to pod name while staying under max length.
 
@@ -67,14 +71,7 @@ def add_pod_suffix(*, pod_name: str, rand_len: int = 8, max_len: int = POD_NAME_
     :param max_len: maximum length of the pod name
     :meta private:
     """
-    warnings.warn(
-        "This function is deprecated. Please use `add_unique_suffix`.",
-        AirflowProviderDeprecationWarning,
-        stacklevel=2,
-    )
-
-    suffix = "-" + rand_str(rand_len)
-    return pod_name[: max_len - len(suffix)].strip("-.") + suffix
+    return add_unique_suffix(name=pod_name, rand_len=rand_len, max_len=max_len)
 
 
 def create_unique_id(
@@ -109,6 +106,10 @@ def create_unique_id(
         return base_name
 
 
+@deprecated(
+    reason="This function is deprecated. Please use `create_unique_id`.",
+    category=AirflowProviderDeprecationWarning,
+)
 def create_pod_id(
     dag_id: str | None = None,
     task_id: str | None = None,
@@ -125,26 +126,7 @@ def create_pod_id(
     :param unique: whether a random string suffix should be added
     :return: A valid identifier for a kubernetes pod name
     """
-    warnings.warn(
-        "This function is deprecated. Please use `create_unique_id`.",
-        AirflowProviderDeprecationWarning,
-        stacklevel=2,
-    )
-
-    if not (dag_id or task_id):
-        raise ValueError("Must supply either dag_id or task_id.")
-    name = ""
-    if dag_id:
-        name += dag_id
-    if task_id:
-        if name:
-            name += "-"
-        name += task_id
-    base_name = slugify(name, lowercase=True)[:max_length].strip(".-")
-    if unique:
-        return add_pod_suffix(pod_name=base_name, rand_len=8, max_len=max_length)
-    else:
-        return base_name
+    return create_unique_id(dag_id=dag_id, task_id=task_id, max_length=max_length, unique=unique)
 
 
 def annotations_to_key(annotations: dict[str, str]) -> TaskInstanceKey:

airflow/providers/cncf/kubernetes/pod_generator.py

@@ -45,7 +45,6 @@
 )
 from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
     POD_NAME_MAX_LENGTH,
-    add_pod_suffix,
     add_unique_suffix,
     rand_str,
 )
@@ -156,12 +155,15 @@ def __init__(
         # Attach sidecar
         self.extract_xcom = extract_xcom
 
-    @deprecated(reason="This function is deprecated.", category=AirflowProviderDeprecationWarning)
+    @deprecated(
+        reason="This method is deprecated and will be removed in the future releases",
+        category=AirflowProviderDeprecationWarning,
+    )
     def gen_pod(self) -> k8s.V1Pod:
         """Generate pod."""
         result = self.ud_pod
 
-        result.metadata.name = add_pod_suffix(pod_name=result.metadata.name)
+        result.metadata.name = add_unique_suffix(name=result.metadata.name)
 
         if self.extract_xcom:
             result = self.add_xcom_sidecar(result)
@@ -210,8 +212,8 @@ def from_obj(obj) -> dict | k8s.V1Pod | None:
             return k8s_object
         elif isinstance(k8s_legacy_object, dict):
             warnings.warn(
-                "Using a dictionary for the executor_config is deprecated and will soon be removed."
-                'please use a `kubernetes.client.models.V1Pod` class with a "pod_override" key'
+                "Using a dictionary for the executor_config is deprecated and will soon be removed. "
+                'Please use a `kubernetes.client.models.V1Pod` class with a "pod_override" key'
                 " instead. ",
                 category=AirflowProviderDeprecationWarning,
                 stacklevel=2,
@@ -575,7 +577,7 @@ def deserialize_model_dict(pod_dict: dict | None) -> k8s.V1Pod:
 
     @staticmethod
     @deprecated(
-        reason="This function is deprecated. Use `add_pod_suffix` in `kubernetes_helper_functions`.",
+        reason="This method is deprecated. Use `add_pod_suffix` in `kubernetes_helper_functions`.",
         category=AirflowProviderDeprecationWarning,
     )
     def make_unique_pod_id(pod_id: str) -> str | None:
@@ -595,12 +597,6 @@ def make_unique_pod_id(pod_id: str) -> str | None:
         :param pod_id: requested pod name
         :return: ``str`` valid Pod name of appropriate length
         """
-        warnings.warn(
-            "This function is deprecated. Use `add_pod_suffix` in `kubernetes_helper_functions`.",
-            AirflowProviderDeprecationWarning,
-            stacklevel=2,
-        )
-
         if not pod_id:
             return None
 

airflow/providers/cncf/kubernetes/template_rendering.py

@@ -24,9 +24,7 @@
 
 from airflow.exceptions import AirflowException
 from airflow.providers.cncf.kubernetes.kube_config import KubeConfig
-from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
-    create_pod_id,
-)
+from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import create_unique_id
 from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
 from airflow.utils.session import NEW_SESSION, provide_session
 
@@ -43,7 +41,7 @@ def render_k8s_pod_yaml(task_instance: TaskInstance) -> dict | None:
         task_id=task_instance.task_id,
         map_index=task_instance.map_index,
         date=None,
-        pod_id=create_pod_id(task_instance.dag_id, task_instance.task_id),
+        pod_id=create_unique_id(task_instance.dag_id, task_instance.task_id),
         try_number=task_instance.try_number,
         kube_image=kube_config.kube_image,
         args=task_instance.command_as_list(),

tests/deprecations_ignore.yml

@@ -54,6 +54,7 @@
 
 
 # CLI
+# https://github.com/apache/airflow/issues/39199
 - tests/cli/commands/test_kubernetes_command.py::TestGenerateDagYamlCommand::test_generate_dag_yaml
 
 
@@ -456,34 +457,6 @@
 - tests/providers/atlassian/jira/operators/test_jira.py::TestJiraOperator::test_project_issue_count
 - tests/providers/atlassian/jira/operators/test_jira.py::TestJiraOperator::test_update_issue
 - tests/providers/atlassian/jira/sensors/test_jira.py::TestJiraSensor::test_issue_label_set
-- tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py::TestAirflowKubernetesScheduler::test_create_pod_id
-- tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py::TestKubernetesExecutor::test_pod_template_file_override_in_executor_config
-- tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py::TestKubernetesExecutor::test_run_next_exception_requeue
-- tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py::TestKubernetesExecutor::test_run_next_pmh_error
-- tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py::TestKubernetesExecutor::test_run_next_pod_reconciliation_error
-- tests/providers/cncf/kubernetes/operators/test_pod.py::TestKubernetesPodOperator::test_execute_async_callbacks
-- tests/providers/cncf/kubernetes/operators/test_pod.py::TestKubernetesPodOperator::test_mark_checked_if_not_deleted
-- tests/providers/cncf/kubernetes/operators/test_pod.py::TestKubernetesPodOperator::test_pod_delete_after_await_container_error
-- tests/providers/cncf/kubernetes/operators/test_pod.py::TestKubernetesPodOperatorAsync::test_async_create_pod_should_throw_exception
-- tests/providers/cncf/kubernetes/operators/test_pod.py::TestKubernetesPodOperatorAsync::test_async_create_pod_with_skip_on_exit_code_should_skip
-- tests/providers/cncf/kubernetes/operators/test_pod.py::TestKubernetesPodOperatorAsync::test_async_create_pod_xcom_push_should_execute_successfully
-- tests/providers/cncf/kubernetes/operators/test_pod.py::TestKubernetesPodOperatorAsync::test_async_get_logs_should_execute_successfully
-- tests/providers/cncf/kubernetes/operators/test_pod.py::TestKubernetesPodOperatorAsync::test_async_write_logs_should_execute_successfully
-- tests/providers/cncf/kubernetes/operators/test_pod.py::test_async_kpo_wait_termination_before_cleanup_on_failure
-- tests/providers/cncf/kubernetes/operators/test_pod.py::test_async_kpo_wait_termination_before_cleanup_on_success
-- tests/providers/cncf/kubernetes/operators/test_pod.py::test_async_skip_kpo_wait_termination_with_timeout_event
-- tests/providers/cncf/kubernetes/operators/test_spark_kubernetes.py::TestSparkKubernetesOperator::test_create_application_from_yaml_json
-- tests/providers/cncf/kubernetes/test_kubernetes_helper_functions.py::TestCreatePodId::test_create_pod_id
-- tests/providers/cncf/kubernetes/test_kubernetes_helper_functions.py::TestCreatePodId::test_create_pod_id_dag_and_task
-- tests/providers/cncf/kubernetes/test_kubernetes_helper_functions.py::TestCreatePodId::test_create_pod_id_dag_only
-- tests/providers/cncf/kubernetes/test_kubernetes_helper_functions.py::TestCreatePodId::test_create_pod_id_dag_too_long_non_unique
-- tests/providers/cncf/kubernetes/test_kubernetes_helper_functions.py::TestCreatePodId::test_create_pod_id_dag_too_long_with_suffix
-- tests/providers/cncf/kubernetes/test_kubernetes_helper_functions.py::TestCreatePodId::test_create_pod_id_task_only
-- tests/providers/cncf/kubernetes/test_pod_generator.py::TestPodGenerator::test_from_obj
-- tests/providers/cncf/kubernetes/test_pod_generator.py::TestPodGenerator::test_gen_pod_extract_xcom
-- tests/providers/cncf/kubernetes/test_pod_generator.py::TestPodGenerator::test_pod_name_confirm_to_max_length
-- tests/providers/cncf/kubernetes/test_pod_generator.py::TestPodGenerator::test_pod_name_is_valid
-- tests/providers/cncf/kubernetes/test_template_rendering.py::test_render_k8s_pod_yaml
 - tests/providers/common/sql/hooks/test_dbapi.py::TestDbApiHook::test_insert_rows_executemany
 - tests/providers/common/sql/hooks/test_dbapi.py::TestDbApiHook::test_insert_rows_replace_executemany_hana_dialect
 - tests/providers/common/sql/hooks/test_dbapi.py::TestDbApiHook::test_instance_check_works_for_legacy_db_api_hook

tests/providers/cncf/kubernetes/executors/test_kubernetes_executor.py

@@ -28,40 +28,36 @@
 from kubernetes.client.rest import ApiException
 from urllib3 import HTTPResponse
 
-from airflow.exceptions import AirflowException
+from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.operators.bash import BashOperator
 from airflow.operators.empty import EmptyOperator
+from airflow.providers.cncf.kubernetes import pod_generator
+from airflow.providers.cncf.kubernetes.executors.kubernetes_executor import (
+    KubernetesExecutor,
+    PodReconciliationError,
+)
+from airflow.providers.cncf.kubernetes.executors.kubernetes_executor_types import (
+    ADOPTED,
+    POD_EXECUTOR_DONE_KEY,
+)
+from airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils import (
+    AirflowKubernetesScheduler,
+    KubernetesJobWatcher,
+    ResourceVersion,
+    get_base_pod_from_template,
+)
+from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
+    annotations_for_logging_task_metadata,
+    annotations_to_key,
+    create_unique_id,
+    get_logs_task_metadata,
+)
+from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
 from airflow.utils import timezone
 from airflow.utils.state import State, TaskInstanceState
 from tests.test_utils.config import conf_vars
 
-try:
-    from airflow.providers.cncf.kubernetes import pod_generator
-    from airflow.providers.cncf.kubernetes.executors.kubernetes_executor import (
-        KubernetesExecutor,
-        PodReconciliationError,
-    )
-    from airflow.providers.cncf.kubernetes.executors.kubernetes_executor_types import (
-        ADOPTED,
-        POD_EXECUTOR_DONE_KEY,
-    )
-    from airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils import (
-        AirflowKubernetesScheduler,
-        KubernetesJobWatcher,
-        ResourceVersion,
-        create_pod_id,
-        get_base_pod_from_template,
-    )
-    from airflow.providers.cncf.kubernetes.kubernetes_helper_functions import (
-        annotations_for_logging_task_metadata,
-        annotations_to_key,
-        get_logs_task_metadata,
-    )
-    from airflow.providers.cncf.kubernetes.pod_generator import PodGenerator
-except ImportError:
-    AirflowKubernetesScheduler = None  # type: ignore
-
 
 class TestAirflowKubernetesScheduler:
     @staticmethod
@@ -100,17 +96,12 @@ def _is_safe_label_value(value):
         regex = r"^[^a-z0-9A-Z]*|[^a-zA-Z0-9_\-\.]|[^a-z0-9A-Z]*$"
         return len(value) <= 63 and re.match(regex, value)
 
-    @pytest.mark.skipif(
-        AirflowKubernetesScheduler is None, reason="kubernetes python package is not installed"
-    )
     def test_create_pod_id(self):
         for dag_id, task_id in self._cases():
-            pod_name = PodGenerator.make_unique_pod_id(create_pod_id(dag_id, task_id))
-            assert self._is_valid_pod_id(pod_name)
+            with pytest.warns(AirflowProviderDeprecationWarning, match=r"deprecated\. Use `add_pod_suffix`"):
+                pod_name = PodGenerator.make_unique_pod_id(create_unique_id(dag_id, task_id))
+            assert self._is_valid_pod_id(pod_name), f"dag_id={dag_id!r}, task_id={task_id!r}"
 
-    @pytest.mark.skipif(
-        AirflowKubernetesScheduler is None, reason="kubernetes python package is not installed"
-    )
     @mock.patch("airflow.providers.cncf.kubernetes.pod_generator.PodGenerator")
     @mock.patch("airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubeConfig")
     def test_get_base_pod_from_template(self, mock_kubeconfig, mock_generator, data_file):
@@ -144,14 +135,16 @@ def test_get_base_pod_from_template(self, mock_kubeconfig, mock_generator, data_
 
     def test_make_safe_label_value(self):
         for dag_id, task_id in self._cases():
+            case = f"dag_id={dag_id!r}, task_id={task_id!r}"
             safe_dag_id = pod_generator.make_safe_label_value(dag_id)
-            assert self._is_safe_label_value(safe_dag_id)
+            assert self._is_safe_label_value(safe_dag_id), case
             safe_task_id = pod_generator.make_safe_label_value(task_id)
-            assert self._is_safe_label_value(safe_task_id)
-            dag_id = "my_dag_id"
-            assert dag_id == pod_generator.make_safe_label_value(dag_id)
-            dag_id = "my_dag_id_" + "a" * 64
-            assert "my_dag_id_" + "a" * 43 + "-0ce114c45" == pod_generator.make_safe_label_value(dag_id)
+            assert self._is_safe_label_value(safe_task_id), case
+
+        dag_id = "my_dag_id"
+        assert dag_id == pod_generator.make_safe_label_value(dag_id)
+        dag_id = "my_dag_id_" + "a" * 64
+        assert "my_dag_id_" + "a" * 43 + "-0ce114c45" == pod_generator.make_safe_label_value(dag_id)
 
     def test_execution_date_serialize_deserialize(self):
         datetime_obj = datetime.now()