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 0ec26723858 Add GKEPodExecOperator for existing Pods (#72577)
0ec26723858 is described below
commit 0ec2672385898b4e60af516f7fb94da4b14bf877
Author: Alejandro Morgante <[email protected]>
AuthorDate: Thu Oct 1 18:52:44 2026 -0300
Add GKEPodExecOperator for existing Pods (#72577)
---
providers/google/docs/index.rst | 8 +-
.../docs/operators/cloud/kubernetes_engine.rst | 20 +++
providers/google/pyproject.toml | 2 +-
.../google/cloud/operators/kubernetes_engine.py | 114 ++++++++++++++++
.../kubernetes_engine/example_kubernetes_engine.py | 105 ++++++++++++++-
.../cloud/operators/test_kubernetes_engine.py | 146 ++++++++++++++++++++-
6 files changed, 388 insertions(+), 7 deletions(-)
diff --git a/providers/google/docs/index.rst b/providers/google/docs/index.rst
index 2bfceb0ed7f..0ac1470c55f 100644
--- a/providers/google/docs/index.rst
+++ b/providers/google/docs/index.rst
@@ -243,11 +243,11 @@ Install them when installing from PyPI. For example:
pip install apache-airflow-providers-google[apache.beam]
-==================== ====================================================
+==================== =====================================================
Extra Dependencies
-==================== ====================================================
+==================== =====================================================
``apache.beam`` ``apache-airflow-providers-apache-beam>=6.2.2``
-``cncf.kubernetes`` ``apache-airflow-providers-cncf-kubernetes>=10.1.0``
+``cncf.kubernetes`` ``apache-airflow-providers-cncf-kubernetes>=10.22.0``
``fab`` ``apache-airflow-providers-fab>=2.0.0``
``leveldb`` ``plyvel>=1.5.1; python_version < '3.13'``
``oracle`` ``apache-airflow-providers-oracle>=3.1.0``
@@ -269,7 +269,7 @@ Extra Dependencies
``standard`` ``apache-airflow-providers-standard``
``common.ai`` ``apache-airflow-providers-common-ai>=0.10.0``
``common.messaging`` ``apache-airflow-providers-common-messaging>=2.0.0``
-==================== ====================================================
+==================== =====================================================
Downloading official packages
-----------------------------
diff --git a/providers/google/docs/operators/cloud/kubernetes_engine.rst
b/providers/google/docs/operators/cloud/kubernetes_engine.rst
index 9baaa0ee6ad..35c26e68d50 100644
--- a/providers/google/docs/operators/cloud/kubernetes_engine.rst
+++ b/providers/google/docs/operators/cloud/kubernetes_engine.rst
@@ -210,6 +210,26 @@ lot less resources wasted on idle Operators or Sensors:
:end-before: [END howto_operator_gke_start_pod_xcom_async]
+.. _howto/operator:GKEPodExecOperator:
+
+Execute a command in an existing Pod
+""""""""""""""""""""""""""""""""""""
+
+Use
:class:`~airflow.providers.google.cloud.operators.kubernetes_engine.GKEPodExecOperator`
to execute a
+command in a running container of an existing Pod using Google Cloud
credentials. The operator discovers the
+GKE cluster endpoint, so a Kubernetes connection or ``kube_config`` file is
not required.
+
+The Pod and container must already exist and be running. The operator streams
the command output and waits for
+its exit code, but it does not create, restart, or delete the Pod. For more
information about command execution,
+output, and XCom behavior, see :ref:`howto/operator:KubernetesPodExecOperator`.
+
+.. exampleinclude::
/../../google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_gke_pod_exec]
+ :end-before: [END howto_operator_gke_pod_exec]
+
+
.. _howto/operator:GKEStartJobOperator:
Run a Job on a GKE cluster
diff --git a/providers/google/pyproject.toml b/providers/google/pyproject.toml
index 451fdbd2f83..363c1a6f25e 100644
--- a/providers/google/pyproject.toml
+++ b/providers/google/pyproject.toml
@@ -163,7 +163,7 @@ dependencies = [
"apache-airflow-providers-apache-beam>=6.2.2",
]
"cncf.kubernetes" = [
- "apache-airflow-providers-cncf-kubernetes>=10.1.0",
+ "apache-airflow-providers-cncf-kubernetes>=10.22.0",
]
"fab" = [
"apache-airflow-providers-fab>=2.0.0",
diff --git
a/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
b/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
index ebcf080d34d..cfdb01d28d6 100644
---
a/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
+++
b/providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py
@@ -61,6 +61,16 @@ from
airflow.providers.google.cloud.triggers.kubernetes_engine import (
from airflow.providers.google.common.hooks.base_google import
PROVIDE_PROJECT_ID
from airflow.providers_manager import ProvidersManager
+try:
+ from airflow.providers.cncf.kubernetes.operators.pod_exec import
KubernetesPodExecOperator
+except ImportError:
+ from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException
+
+ raise AirflowOptionalProviderFeatureException(
+ "Failed to import KubernetesPodExecOperator. This operator is only
available in cncf-kubernetes "
+ "provider version >=10.22.0"
+ )
+
try:
from airflow.providers.cncf.kubernetes.operators.job import
KubernetesDeleteJobOperator
except ImportError:
@@ -689,6 +699,110 @@ class
GKEStartKueueInsideClusterOperator(GKEOperatorMixin, KubernetesInstallKueu
)
+class GKEPodExecOperator(GKEOperatorMixin, KubernetesPodExecOperator):
+ """
+ Execute a command in a running container of an existing Pod on Google
Kubernetes Engine.
+
+ The operator authenticates with Google Cloud and delegates command
execution to
+
:class:`~airflow.providers.cncf.kubernetes.operators.pod_exec.KubernetesPodExecOperator`.
+ It does not create, restart, or delete the target Pod.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:GKEPodExecOperator`
+
+ :param location: The Google Kubernetes Engine zone or region in which the
cluster resides.
+ (templated)
+ :param cluster_name: The name of the Google Kubernetes Engine cluster.
(templated)
+ :param pod_name: Name of the existing Kubernetes Pod. (templated)
+ :param command: Command and arguments to execute in the container.
(templated)
+ :param namespace: Namespace containing the Pod. Defaults to ``default``.
(templated)
+ :param container_name: Name of the container in which to execute the
command. When omitted, the
+ ``kubectl.kubernetes.io/default-container`` annotation or the first
container is used.
+ Defaults to ``None``. (templated)
+ :param use_internal_ip: Use the internal IP address as the endpoint.
Defaults to ``False``.
+ (templated)
+ :param use_dns_endpoint: Use the DNS address as the endpoint. Defaults to
``False``. This must be
+ set to ``True`` for Sovereign Cloud from Google. (templated)
+ :param project_id: The Google Cloud project ID. Defaults to the project
inferred from the Google
+ Cloud connection. (templated)
+ :param gcp_conn_id: The Google Cloud connection ID to use. Defaults to
``google_cloud_default``.
+ (templated)
+ :param impersonation_chain: Optional service account to impersonate using
short-term credentials,
+ or a sequence of accounts required to impersonate the final account.
Defaults to ``None``.
+ (templated)
+ :param do_xcom_push: Return standard output through XCom when ``True``.
Defaults to ``False``.
+ :param max_xcom_output_size: Maximum UTF-8 byte size retained for XCom.
Defaults to 49,344 bytes.
+ """
+
+ template_fields: Sequence[str] = tuple(
+ set(GKEOperatorMixin.template_fields)
+ | (
+ set(KubernetesPodExecOperator.template_fields)
+ - {"cluster_context", "config_file", "kubernetes_conn_id"}
+ )
+ )
+ operator_extra_links = (KubernetesEnginePodLink(),)
+
+ def __init__(
+ self,
+ *,
+ location: str,
+ cluster_name: str,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str = "default",
+ container_name: str | None = None,
+ use_internal_ip: bool = False,
+ use_dns_endpoint: bool = False,
+ project_id: str = PROVIDE_PROJECT_ID,
+ gcp_conn_id: str = "google_cloud_default",
+ impersonation_chain: str | Sequence[str] | None = None,
+ **kwargs,
+ ) -> None:
+ for parameter in ("config_file", "kubernetes_conn_id", "in_cluster",
"cluster_context"):
+ if parameter in kwargs:
+ raise ValueError(
+ f"`{parameter}` is not allowed for GKEPodExecOperator
because authentication is managed "
+ "through `gcp_conn_id`."
+ )
+ if gcp_conn_id is None:
+ raise ValueError(
+ "`gcp_conn_id` must not be None. To use Application Default
Credentials, configure an "
+ "empty `google_cloud_default` connection."
+ )
+ super().__init__(
+ pod_name=pod_name,
+ command=command,
+ namespace=namespace,
+ container_name=container_name,
+ kubernetes_conn_id=None,
+ in_cluster=False,
+ cluster_context=None,
+ config_file=None,
+ **kwargs,
+ )
+ self.project_id = project_id
+ self.location = location
+ self.cluster_name = cluster_name
+ self.gcp_conn_id = gcp_conn_id
+ self.use_internal_ip = use_internal_ip
+ self.use_dns_endpoint = use_dns_endpoint
+ self.impersonation_chain = impersonation_chain
+
+ def execute(self, context: Context) -> str | None:
+ namespace = self._resolve_namespace()
+ KubernetesEnginePodLink.persist(
+ context=context,
+ project_id=self.project_id,
+ location=self.location,
+ cluster_name=self.cluster_name,
+ namespace=namespace,
+ pod_name=self.pod_name,
+ )
+ return super().execute(context)
+
+
class GKEStartPodOperator(GKEOperatorMixin, KubernetesPodOperator):
"""
Executes a task in a Kubernetes pod in the specified Google Kubernetes
Engine cluster.
diff --git
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
index 54de4a904ba..77b04def7ad 100644
---
a/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
+++
b/providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py
@@ -24,18 +24,28 @@ from __future__ import annotations
import os
from datetime import datetime
+from kubernetes.client.rest import ApiException
+
from airflow.models.dag import DAG
+from airflow.providers.cncf.kubernetes.utils.container import
container_is_running
+from airflow.providers.cncf.kubernetes.utils.pod_manager import PodPhase
+from airflow.providers.google.cloud.hooks.kubernetes_engine import GKEHook,
GKEKubernetesHook
from airflow.providers.google.cloud.operators.kubernetes_engine import (
+ GKEClusterAuthDetails,
GKECreateClusterOperator,
+ GKECreateCustomResourceOperator,
GKEDeleteClusterOperator,
+ GKEDeleteCustomResourceOperator,
+ GKEPodExecOperator,
GKEStartPodOperator,
)
from airflow.providers.standard.operators.bash import BashOperator
try:
- from airflow.sdk import TriggerRule
+ from airflow.sdk import TriggerRule, task
except ImportError:
# Compatibility for Airflow < 3.1
+ from airflow.decorators import task # type: ignore[attr-defined,no-redef]
from airflow.utils.trigger_rule import TriggerRule # type:
ignore[no-redef,attr-defined]
from system.google import DEFAULT_GCP_SYSTEM_TEST_PROJECT_ID
@@ -49,11 +59,60 @@ GCP_LOCATION = "europe-west1"
CLUSTER_NAME_BASE = f"cluster-{DAG_ID}".replace("_", "-")
CLUSTER_NAME_FULL = CLUSTER_NAME_BASE + f"-{ENV_ID}".replace("_", "-")
CLUSTER_NAME = CLUSTER_NAME_BASE if len(CLUSTER_NAME_FULL) >= 33 else
CLUSTER_NAME_FULL
+EXEC_POD_NAME = f"existing-pod-{ENV_ID}".replace("_", "-")
+EXEC_CONTAINER_NAME = "main"
+EXPECTED_EXEC_OUTPUT = "command executed in existing GKE Pod"
# [START howto_operator_gcp_gke_create_cluster_definition]
CLUSTER = {"name": CLUSTER_NAME, "initial_node_count": 1, "autopilot":
{"enabled": True}}
# [END howto_operator_gcp_gke_create_cluster_definition]
+EXEC_POD = f"""
+apiVersion: v1
+kind: Pod
+metadata:
+ name: {EXEC_POD_NAME}
+ namespace: default
+spec:
+ restartPolicy: Never
+ containers:
+ - name: {EXEC_CONTAINER_NAME}
+ image: busybox:1.38.0
+ command: ["sleep", "3600"]
+"""
+
+
[email protected](poke_interval=10, timeout=300, mode="reschedule")
+def wait_for_running_exec_pod() -> bool:
+ cluster_hook = GKEHook(location=GCP_LOCATION)
+ cluster_url, ssl_ca_cert = GKEClusterAuthDetails(
+ cluster_name=CLUSTER_NAME,
+ project_id=GCP_PROJECT_ID,
+ use_internal_ip=False,
+ use_dns_endpoint=False,
+ cluster_hook=cluster_hook,
+ ).fetch_cluster_info()
+ hook = GKEKubernetesHook(
+ gcp_conn_id="google_cloud_default",
+ cluster_url=cluster_url,
+ ssl_ca_cert=ssl_ca_cert,
+ )
+ try:
+ pod = hook.get_pod(name=EXEC_POD_NAME, namespace="default")
+ except ApiException as error:
+ if error.status == 404:
+ return False
+ raise
+ return bool(
+ pod.status and pod.status.phase == PodPhase.RUNNING and
container_is_running(pod, EXEC_CONTAINER_NAME)
+ )
+
+
+@task
+def verify_exec_output(output: str) -> None:
+ if output != EXPECTED_EXEC_OUTPUT:
+ raise ValueError(f"Unexpected command output: {output!r}")
+
with DAG(
DAG_ID,
@@ -99,6 +158,41 @@ with DAG(
)
# [END howto_operator_gke_start_pod_xcom]
+ create_exec_pod = GKECreateCustomResourceOperator(
+ task_id="create_exec_pod",
+ project_id=GCP_PROJECT_ID,
+ location=GCP_LOCATION,
+ cluster_name=CLUSTER_NAME,
+ yaml_conf=EXEC_POD,
+ )
+
+ exec_pod_is_running = wait_for_running_exec_pod()
+
+ # [START howto_operator_gke_pod_exec]
+ exec_in_existing_pod = GKEPodExecOperator(
+ task_id="exec_in_existing_pod",
+ project_id=GCP_PROJECT_ID,
+ location=GCP_LOCATION,
+ cluster_name=CLUSTER_NAME,
+ pod_name=EXEC_POD_NAME,
+ namespace="default",
+ container_name=EXEC_CONTAINER_NAME,
+ command=["sh", "-c", f"printf '{EXPECTED_EXEC_OUTPUT}'"],
+ do_xcom_push=True,
+ )
+ # [END howto_operator_gke_pod_exec]
+
+ exec_output_is_valid = verify_exec_output(exec_in_existing_pod.output)
+
+ delete_exec_pod = GKEDeleteCustomResourceOperator(
+ task_id="delete_exec_pod",
+ project_id=GCP_PROJECT_ID,
+ location=GCP_LOCATION,
+ cluster_name=CLUSTER_NAME,
+ yaml_conf=EXEC_POD,
+ trigger_rule=TriggerRule.ALL_DONE,
+ )
+
# [START howto_operator_gke_xcom_result]
pod_task_xcom_result = BashOperator(
task_id="pod_task_xcom_result",
@@ -124,6 +218,15 @@ with DAG(
delete_cluster.trigger_rule = TriggerRule.ALL_DONE
create_cluster >> [pod_task, pod_task_xcom] >> delete_cluster
+ (
+ create_cluster
+ >> create_exec_pod
+ >> exec_pod_is_running
+ >> exec_in_existing_pod
+ >> exec_output_is_valid
+ >> delete_exec_pod
+ >> delete_cluster
+ )
pod_task_xcom >> pod_task_xcom_result
from tests_common.test_utils.watcher import watcher
diff --git
a/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
b/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
index 4c92ab65a79..0008fc31c37 100644
---
a/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
+++
b/providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py
@@ -17,6 +17,7 @@
# under the License.
from __future__ import annotations
+import runpy
from copy import deepcopy
from unittest import mock
from unittest.mock import PropertyMock, call
@@ -36,12 +37,13 @@ from airflow.providers.cncf.kubernetes.operators.kueue
import (
KubernetesStartKueueJobOperator,
)
from airflow.providers.cncf.kubernetes.operators.pod import
KubernetesPodOperator
+from airflow.providers.cncf.kubernetes.operators.pod_exec import
KubernetesPodExecOperator
from airflow.providers.cncf.kubernetes.operators.resource import (
KubernetesCreateResourceOperator,
KubernetesDeleteResourceOperator,
)
from airflow.providers.cncf.kubernetes.utils.pod_manager import OnFinishAction
-from airflow.providers.common.compat.sdk import AirflowException
+from airflow.providers.common.compat.sdk import AirflowException,
AirflowOptionalProviderFeatureException
from airflow.providers.google.cloud.links.base import BASE_LINK
from airflow.providers.google.cloud.links.kubernetes_engine import (
KUBERNETES_JOB_LINK,
@@ -49,6 +51,7 @@ from airflow.providers.google.cloud.links.kubernetes_engine
import (
KubernetesEngineJobLink,
KubernetesEnginePodLink,
)
+from airflow.providers.google.cloud.operators import kubernetes_engine
from airflow.providers.google.cloud.operators.kubernetes_engine import (
GKEClusterAuthDetails,
GKECreateClusterOperator,
@@ -59,6 +62,7 @@ from
airflow.providers.google.cloud.operators.kubernetes_engine import (
GKEDescribeJobOperator,
GKEListJobsOperator,
GKEOperatorMixin,
+ GKEPodExecOperator,
GKEResumeJobOperator,
GKEStartJobOperator,
GKEStartKueueInsideClusterOperator,
@@ -825,6 +829,146 @@ class TestGKEStartKueueInsideClusterOperator:
)
[email protected]("sys.modules",
{"airflow.providers.cncf.kubernetes.operators.pod_exec": None})
+def
test_missing_kubernetes_pod_exec_operator_reports_required_provider_version():
+ with pytest.raises(
+ AirflowOptionalProviderFeatureException, match="cncf-kubernetes
provider version >=10.22.0"
+ ):
+ runpy.run_path(kubernetes_engine.__file__)
+
+
+class TestGKEPodExecOperator:
+ def setup_method(self):
+ self.operator = GKEPodExecOperator(
+ task_id=TEST_TASK_ID,
+ project_id=TEST_PROJECT_ID,
+ location=TEST_LOCATION,
+ cluster_name=GKE_CLUSTER_NAME,
+ pod_name=K8S_POD_NAME,
+ namespace=K8S_NAMESPACE,
+ container_name="worker",
+ command=["dbt", "run"],
+ gcp_conn_id=TEST_CONN_ID,
+ impersonation_chain=TEST_IMPERSONATION_CHAIN,
+ do_xcom_push=True,
+ max_xcom_output_size=1024,
+ )
+
+ def test_constructor_forces_gke_auth_and_forwards_exec_options(self):
+ assert self.operator.kubernetes_conn_id is None
+ assert self.operator.in_cluster is False
+ assert self.operator.cluster_context is None
+ assert self.operator.config_file is None
+ assert self.operator.do_xcom_push is True
+ assert self.operator.max_xcom_output_size == 1024
+
+ def test_default_namespace_when_omitted(self):
+ operator = GKEPodExecOperator(
+ task_id=TEST_TASK_ID,
+ location=TEST_LOCATION,
+ cluster_name=GKE_CLUSTER_NAME,
+ pod_name=K8S_POD_NAME,
+ command=["true"],
+ )
+
+ assert operator.namespace == "default"
+
+ @pytest.mark.parametrize(
+ ("kwargs", "expected_message"),
+ [
+ (
+ {"config_file": "/path/to/kubeconfig"},
+ "`config_file` is not allowed for GKEPodExecOperator",
+ ),
+ (
+ {"config_file": None},
+ "`config_file` is not allowed for GKEPodExecOperator",
+ ),
+ (
+ {"kubernetes_conn_id": "kubernetes_default"},
+ "`kubernetes_conn_id` is not allowed for GKEPodExecOperator",
+ ),
+ (
+ {"in_cluster": True},
+ "`in_cluster` is not allowed for GKEPodExecOperator",
+ ),
+ (
+ {"cluster_context": "test-context"},
+ "`cluster_context` is not allowed for GKEPodExecOperator",
+ ),
+ (
+ {"gcp_conn_id": None},
+ "`gcp_conn_id` must not be None",
+ ),
+ ],
+ ids=[
+ "config-file",
+ "config-file-none",
+ "kubernetes-connection",
+ "in-cluster",
+ "cluster-context",
+ "missing-gcp-connection",
+ ],
+ )
+ def test_invalid_auth_parameters(self, kwargs, expected_message):
+ with pytest.raises(ValueError, match=expected_message):
+ GKEPodExecOperator(
+ task_id=TEST_TASK_ID,
+ location=TEST_LOCATION,
+ cluster_name=GKE_CLUSTER_NAME,
+ pod_name=K8S_POD_NAME,
+ command=["true"],
+ **kwargs,
+ )
+
+ def test_template_fields(self):
+ expected_template_fields = set(GKEOperatorMixin.template_fields) | (
+ set(KubernetesPodExecOperator.template_fields)
+ - {"cluster_context", "config_file", "kubernetes_conn_id"}
+ )
+
+ assert set(GKEPodExecOperator.template_fields) ==
expected_template_fields
+
+ @mock.patch(GKE_OPERATORS_PATH.format("GKEKubernetesHook"), autospec=True)
+ @mock.patch(GKE_OPERATORS_PATH.format("GKEOperatorMixin.cluster_info"),
new_callable=PropertyMock)
+ def test_hook_uses_gke_authentication(self, mock_cluster_info, mock_hook):
+ mock_cluster_info.return_value = (GKE_CLUSTER_URL, GKE_SSL_CA_CERT)
+
+ result = self.operator.hook
+
+ assert result == mock_hook.return_value
+ mock_hook.assert_called_once_with(
+ gcp_conn_id=TEST_CONN_ID,
+ impersonation_chain=TEST_IMPERSONATION_CHAIN,
+ cluster_url=GKE_CLUSTER_URL,
+ ssl_ca_cert=GKE_SSL_CA_CERT,
+ enable_tcp_keepalive=False,
+ use_dns_endpoint=False,
+ )
+
+ @mock.patch(GKE_OPERATORS_PATH.format("KubernetesEnginePodLink.persist"),
autospec=True)
+ @mock.patch(
+ GKE_OPERATORS_PATH.format("KubernetesPodExecOperator.execute"),
+ autospec=True,
+ return_value="command output",
+ )
+ def test_execute_persists_link_and_returns_output(self, mock_execute,
mock_persist_link):
+ context = {}
+
+ result = self.operator.execute(context)
+
+ assert result == "command output"
+ mock_persist_link.assert_called_once_with(
+ context=context,
+ project_id=TEST_PROJECT_ID,
+ location=TEST_LOCATION,
+ cluster_name=GKE_CLUSTER_NAME,
+ namespace=K8S_NAMESPACE,
+ pod_name=K8S_POD_NAME,
+ )
+ mock_execute.assert_called_once_with(self.operator, context)
+
+
class TestGKEStartPodOperator:
def setup_method(self):
self.operator = GKEStartPodOperator(
