potiuk commented on code in PR #72577:
URL: https://github.com/apache/airflow/pull/72577#discussion_r3999096240
##########
providers/google/src/airflow/providers/google/cloud/operators/kubernetes_engine.py:
##########
@@ -688,6 +689,110 @@ def execute(self, context: Context):
)
+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:
+ config_file = kwargs.pop("config_file", None)
+ if config_file is not None:
+ raise ValueError(
+ "`config_file` 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,
Review Comment:
Nit: `config_file` gets a clean `ValueError` just above, but the three
sibling parameters that are forced the same way don't — they collide with the
explicit keywords here and surface an internal-looking error:
```
kubernetes_conn_id -> TypeError: ...KubernetesPodExecOperator.__init__() got
multiple values for keyword argument 'kubernetes_conn_id'
in_cluster -> TypeError: ... 'in_cluster'
cluster_context -> TypeError: ... 'cluster_context'
config_file -> ValueError: `config_file` is not allowed for
GKEPodExecOperator because authentication is managed through `gcp_conn_id`.
```
Someone migrating a `KubernetesPodExecOperator` task over to GKE hits a
`TypeError` naming the parent class. Rejecting all four through the same path —
or just naming them in the existing `config_file` message — would be more
consistent.
---
Drafted-by: Claude Opus 5; reviewed by @potiuk before posting
##########
providers/google/tests/unit/google/cloud/operators/test_kubernetes_engine.py:
##########
@@ -825,6 +827,104 @@ def test_execute_not_scalable(self, mock_hook, mock_link,
mock_super, mock_log):
)
+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",
Review Comment:
Nit: `setup_method` always passes `namespace` explicitly, so nothing pins
the default. `GKEPodExecOperator` narrows the parent's `namespace: str | None =
None` to `namespace: str = "default"` and the docstring advertises that — worth
one assertion on an operator built without `namespace`.
Related, no action needed: because this namespace is also what
`_resolve_namespace()` returns, `test_execute_persists_link_and_returns_output`
can't distinguish resolved from raw. The assertion is a little weaker than it
looks.
---
Drafted-by: Claude Opus 5; reviewed by @potiuk before posting
##########
providers/google/tests/system/google/cloud/kubernetes_engine/example_kubernetes_engine.py:
##########
@@ -49,11 +59,60 @@
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 = "existing-pod"
+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:
Review Comment:
Nit: this sensor hand-rolls the GKE auth handshake (`GKEClusterAuthDetails`
+ `GKEKubernetesHook`) and imports `container_is_running` from
`airflow.providers.cncf.kubernetes.utils.container` — a `utils` module rather
than documented public interface.
The wait itself is clearly needed (`GKECreateCustomResourceOperator` doesn't
block until Running, and the exec operator requires a Running Pod), so this is
style only — but this file doubles as the docs source, and the sensor is the
largest single block the PR adds to it.
---
Drafted-by: Claude Opus 5; reviewed by @potiuk before posting
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
