You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "AlejandroMorgante (via GitHub)" <gi...@apache.org> on 2026/08/06 15:27:34 UTC
[PR] Add KubernetesPodExecOperator to execute commands in existing Kubernetes Pods [airflow]
AlejandroMorgante opened a new pull request, #71244:
URL: https://github.com/apache/airflow/pull/71244
Add `KubernetesPodExecOperator` to execute commands in a running container of an existing Kubernetes Pod without creating, restarting, or deleting it. This supports pre-warmed and API-visible static Pods where avoiding startup latency is important.
The operator validates the target Pod and container, streams stdout and stderr, supports optional stdout XCom, reports non-zero exit codes, and closes the exec connection when the task is killed. Provider metadata, RBAC guidance, and operator documentation are included.
Tests:
- `pytest providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod_exec.py -q` (32 passed)
- Provider mypy check
- Provider metadata and documentation validation
- Pre-commit and manual checks
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Codex (GPT-5)
Generated-by: Codex (GPT-5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889735566
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
+ def _log_output(self, output: str, *, stream_name: str) -> None:
+ log_method = self.log.warning if stream_name == "stderr" else self.log.info
Review Comment:
Done, thank you!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889720742
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod_exec.py:
##########
@@ -0,0 +1,318 @@
+# 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
+
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+from kubernetes.client.rest import ApiException
+from kubernetes.stream.ws_client import WSClient
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator
+
+MODULE = "airflow.providers.cncf.kubernetes.operators.pod_exec"
+
+
+def create_pod(
+ *,
+ container_names: tuple[str, ...] = ("main",),
+ annotations: dict[str, str] | None = None,
+ phase: str = "Running",
+ container_statuses=None,
+ with_spec: bool = True,
+ with_status: bool = True,
+):
+ if container_statuses is None:
+ container_statuses = [
+ SimpleNamespace(name=name, state=SimpleNamespace(running=object())) for name in container_names
+ ]
+ spec = SimpleNamespace(containers=[SimpleNamespace(name=name) for name in container_names])
+ status = SimpleNamespace(phase=phase, container_statuses=container_statuses)
+ return SimpleNamespace(
+ metadata=SimpleNamespace(annotations=annotations),
+ spec=spec if with_spec else None,
+ status=status if with_status else None,
+ )
+
+
+def create_exec_client(*, stdout=(), stderr=(), returncode=0):
+ exec_client = mock.MagicMock(spec=WSClient)
+ exec_client.is_open.side_effect = [True, False]
+ exec_client.peek_stdout.side_effect = [*stdout, ""]
+ exec_client.read_stdout.side_effect = stdout
+ exec_client.peek_stderr.side_effect = [*stderr, ""]
+ exec_client.read_stderr.side_effect = stderr
+ exec_client.returncode = returncode
+ return exec_client
+
+
+def create_operator(*, pod=None, hook_namespace=None, **kwargs):
+ operator = KubernetesPodExecOperator(
+ task_id="exec",
+ pod_name="existing-pod",
+ command=["echo", "hello"],
+ **kwargs,
+ )
+ hook = mock.MagicMock(spec=KubernetesHook)
+ hook.get_namespace.return_value = hook_namespace
+ hook.get_pod.return_value = pod or create_pod()
+ operator.__dict__["hook"] = hook
+ return operator, hook
+
+
+class TestKubernetesPodExecOperator:
+ def test_template_fields(self):
+ assert set(KubernetesPodExecOperator.template_fields) == {
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ }
+
+ @mock.patch(f"{MODULE}.KubernetesHook", autospec=True)
+ def test_hook_configuration(self, kubernetes_hook_mock):
+ operator = KubernetesPodExecOperator(
+ task_id="exec",
+ pod_name="existing-pod",
+ command=["date"],
+ kubernetes_conn_id="kubernetes_test",
+ in_cluster=True,
+ config_file="/tmp/kubeconfig",
+ cluster_context="test-context",
+ )
+
+ assert operator.hook is kubernetes_hook_mock.return_value
+ kubernetes_hook_mock.assert_called_once_with(
+ conn_id="kubernetes_test",
+ in_cluster=True,
+ config_file="/tmp/kubeconfig",
+ cluster_context="test-context",
+ )
+
+ @pytest.mark.parametrize(
+ ("do_xcom_push", "expected_result"),
+ [(False, None), (True, "hello\nworld\n")],
+ )
+ @mock.patch(f"{MODULE}.KubernetesPodExecOperator.log", spec=["info", "warning"])
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_execute(self, kubernetes_stream_mock, log_mock, do_xcom_push, expected_result):
+ operator, hook = create_operator(
+ namespace="test-namespace",
+ container_name="main",
+ do_xcom_push=do_xcom_push,
+ )
+ exec_client = create_exec_client(stdout=("hello\n", "world\n"), stderr=("warning\n",), returncode=0)
+ kubernetes_stream_mock.return_value = exec_client
+
+ result = operator.execute(context={})
+
+ assert result == expected_result
+ hook.get_pod.assert_called_once_with(name="existing-pod", namespace="test-namespace")
+ kubernetes_stream_mock.assert_called_once_with(
+ hook.core_v1_client.connect_get_namespaced_pod_exec,
+ name="existing-pod",
+ namespace="test-namespace",
+ container="main",
+ command=["echo", "hello"],
+ stdin=False,
+ stdout=True,
+ stderr=True,
+ tty=False,
+ _preload_content=False,
+ )
+ exec_client.update.assert_called_once_with(timeout=1)
+ assert exec_client.read_stdout.call_count == 2
+ exec_client.read_stderr.assert_called_once_with()
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+ log_mock.info.assert_has_calls(
+ [
+ mock.call("[%s] %s", "stdout", "hello"),
+ mock.call("[%s] %s", "stdout", "world"),
+ ]
+ )
+ log_mock.warning.assert_called_once_with("[%s] %s", "stderr", "warning")
+
+ @pytest.mark.parametrize(
+ ("namespace", "hook_namespace", "expected_namespace"),
+ [
+ ("task-namespace", "connection-namespace", "task-namespace"),
+ (None, "connection-namespace", "connection-namespace"),
+ (None, None, KubernetesHook.DEFAULT_NAMESPACE),
+ ],
+ )
+ def test_resolve_namespace(self, namespace, hook_namespace, expected_namespace):
+ operator, _ = create_operator(namespace=namespace, hook_namespace=hook_namespace)
+
+ assert operator._resolve_namespace() == expected_namespace
+
+ @pytest.mark.parametrize(
+ ("container_name", "annotations", "expected_container"),
+ [
+ ("secondary", {"kubectl.kubernetes.io/default-container": "main"}, "secondary"),
+ (None, {"kubectl.kubernetes.io/default-container": "secondary"}, "secondary"),
+ (None, {"kubectl.kubernetes.io/default-container": "missing"}, "main"),
+ (None, None, "main"),
+ ],
+ )
+ def test_resolve_container_name(self, container_name, annotations, expected_container):
+ pod = create_pod(container_names=("main", "secondary"), annotations=annotations)
+ operator, _ = create_operator(pod=pod, container_name=container_name)
+
+ assert operator._resolve_container_name(pod) == expected_container
+
+ @pytest.mark.parametrize("with_spec", [False, True])
+ def test_rejects_pod_without_containers(self, with_spec):
+ pod = create_pod(container_names=(), with_spec=with_spec)
+ operator, _ = create_operator(pod=pod)
+
+ with pytest.raises(PodExecException, match="does not define any containers"):
+ operator._resolve_container_name(pod)
+
+ def test_rejects_unknown_container(self):
+ pod = create_pod()
+ operator, _ = create_operator(pod=pod, container_name="missing")
+
+ with pytest.raises(PodExecException, match="does not exist"):
+ operator._resolve_container_name(pod)
+
+ @pytest.mark.parametrize(
+ ("pod", "expected_message"),
+ [
+ (create_pod(with_status=False), "phase None"),
+ (create_pod(phase="Pending"), "phase 'Pending'"),
+ (create_pod(container_statuses=[]), "is not running"),
+ (
+ create_pod(container_statuses=[SimpleNamespace(name="main", state=None)]),
+ "is not running",
+ ),
+ (
+ create_pod(
+ container_statuses=[SimpleNamespace(name="main", state=SimpleNamespace(running=None))]
+ ),
+ "is not running",
+ ),
+ ],
+ )
+ def test_rejects_unavailable_target(self, pod, expected_message):
+ operator, _ = create_operator(pod=pod)
+
+ with pytest.raises(PodExecException, match=expected_message):
+ operator._validate_container_is_running(pod, "main")
+
+ @pytest.mark.parametrize(
+ ("command", "error", "expected_message"),
+ [
+ ("echo hello", TypeError, "sequence of strings"),
+ (42, TypeError, "sequence of strings"),
+ ([], ValueError, "at least one element"),
+ (["echo", 42], TypeError, "Every element"),
+ ],
+ )
+ def test_rejects_invalid_command(self, command, error, expected_message):
+ operator, _ = create_operator()
+ operator.command = command
+
+ with pytest.raises(error, match=expected_message):
+ operator.execute(context={})
+
+ def test_rejects_empty_pod_name(self):
+ operator, _ = create_operator()
+ operator.pod_name = ""
+
+ with pytest.raises(ValueError, match="must not be empty"):
+ operator.execute(context={})
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_wraps_pod_read_error(self, kubernetes_stream_mock):
+ operator, hook = create_operator(namespace="test-namespace")
+ hook.get_pod.side_effect = ApiException(status=404, reason="Not Found")
+
+ with pytest.raises(PodExecException, match="Unable to read pod.*Not Found"):
+ operator.execute(context={})
+
+ kubernetes_stream_mock.assert_not_called()
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_wraps_exec_api_error(self, kubernetes_stream_mock):
+ operator, _ = create_operator(namespace="test-namespace")
+ kubernetes_stream_mock.side_effect = ApiException(status=403, reason="Forbidden")
+
+ with pytest.raises(PodExecException, match="Unable to execute command.*Forbidden"):
+ operator.execute(context={})
+
+ assert operator._exec_client is None
+
+ @pytest.mark.parametrize(
+ ("returncode", "expected_message"),
+ [(None, "without reporting an exit code"), (17, "failed with exit code 17")],
+ )
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_rejects_unsuccessful_command(self, kubernetes_stream_mock, returncode, expected_message):
+ operator, _ = create_operator(namespace="test-namespace")
+ exec_client = create_exec_client(returncode=returncode)
+ kubernetes_stream_mock.return_value = exec_client
+
+ with pytest.raises(PodExecException, match=expected_message):
+ operator.execute(context={})
+
+ exec_client.close.assert_called_once_with()
+
+ def test_on_kill_closes_active_connection(self):
+ operator, _ = create_operator()
+ exec_client = mock.MagicMock(spec=WSClient)
+ operator._exec_client = exec_client
+
+ operator.on_kill()
+
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_on_kill_while_consuming_output(self, kubernetes_stream_mock):
+ operator, _ = create_operator(namespace="test-namespace")
+ exec_client = create_exec_client()
+ exec_client.is_open.side_effect = lambda: (operator.on_kill(), False)[1]
+ kubernetes_stream_mock.return_value = exec_client
+
+ assert operator.execute(context={}) is None
+
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+
+ def test_on_kill_without_active_connection(self):
+ operator, _ = create_operator()
+
+ operator.on_kill()
+
+ assert operator._exec_client is None
+
+ def test_close_error_does_not_mask_task_shutdown(self):
+ operator, _ = create_operator()
+ exec_client = mock.MagicMock(spec=WSClient)
+ exec_client.close.side_effect = RuntimeError("connection already closed")
+ operator._exec_client = exec_client
+
+ operator.on_kill()
+
+ assert operator._exec_client is None
Review Comment:
Done, thanks! Added a test that makes exec_client.update() raise after the connection is assigned and verifies that the connection is closed and the internal references are cleared
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3883213054
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/exceptions.py:
##########
@@ -39,3 +39,7 @@ class KubernetesApiError(AirflowException):
class KubernetesApiPermissionError(AirflowException):
"""Raised when an error is encountered while trying access Kubernetes API."""
+
+
+class PodExecException(AirflowException):
Review Comment:
Do we need a dedicated `PodExecException` here? It appears to be functionally equivalent to `AirflowException`, which we are generally moving away from using for operator execution failures. Could these instead use the appropriate built-in exception types?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
+ def _log_output(self, output: str, *, stream_name: str) -> None:
+ log_method = self.log.warning if stream_name == "stderr" else self.log.info
+ for line in output.splitlines():
+ log_method("[%s] %s", stream_name, line)
+
+ def _consume_output(self, exec_client: WSClient) -> str:
+ stdout_chunks: list[str] = []
+ while exec_client.is_open():
+ exec_client.update(timeout=1)
+ while exec_client.peek_stdout():
+ output = exec_client.read_stdout()
+ self._log_output(output, stream_name="stdout")
+ if self.do_xcom_push:
+ stdout_chunks.append(output)
+ while exec_client.peek_stderr():
+ self._log_output(exec_client.read_stderr(), stream_name="stderr")
+ return "".join(stdout_chunks)
+
+ def _close_exec_client(self) -> None:
+ exec_client = self._exec_client
+ self._exec_client = None
+ if exec_client is None:
+ return
+ try:
+ exec_client.close()
+ except Exception:
+ self.log.exception("Failed to close Kubernetes exec connection")
Review Comment:
Could this log include the namespace, pod name, and container name? The exception traceback is retained, but `"Failed to close Kubernetes exec connection"` alone does not identify which target failed when a worker is running multiple tasks.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
+ def _log_output(self, output: str, *, stream_name: str) -> None:
+ log_method = self.log.warning if stream_name == "stderr" else self.log.info
+ for line in output.splitlines():
+ log_method("[%s] %s", stream_name, line)
+
+ def _consume_output(self, exec_client: WSClient) -> str:
+ stdout_chunks: list[str] = []
+ while exec_client.is_open():
+ exec_client.update(timeout=1)
+ while exec_client.peek_stdout():
+ output = exec_client.read_stdout()
+ self._log_output(output, stream_name="stdout")
+ if self.do_xcom_push:
+ stdout_chunks.append(output)
+ while exec_client.peek_stderr():
+ self._log_output(exec_client.read_stderr(), stream_name="stderr")
+ return "".join(stdout_chunks)
Review Comment:
When `do_xcom_push=True`, this buffers the command’s complete stdout in worker memory without any size limit, and `"".join(stdout_chunks) `then allocates the combined string as well. Since the command can emit arbitrary output, could this exhaust worker memory before we even reach the XCom size/storage constraints? Should the operator enforce a configurable limit or otherwise avoid retaining unbounded output in memory?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
Review Comment:
Could this use `PodPhase.RUNNING` from pod_manager instead of the literal `"Running"`?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
+ def _log_output(self, output: str, *, stream_name: str) -> None:
+ log_method = self.log.warning if stream_name == "stderr" else self.log.info
+ for line in output.splitlines():
Review Comment:
Are values returned by `read_stdout()` and `read_stderr(`) guaranteed to contain complete lines? If these are arbitrary WebSocket chunks, `_log_output()` can split one logical line across multiple log records. Should incomplete lines be buffered between reads?
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
Review Comment:
I think this log message is a bit too presumptive. No container status or a missing state does not necessarily mean the container is not running. You could do something like this instead:
`RuntimeError(f"Container {container_name!r} in pod {self.pod_name!r} is not running or container status cannot be retrieved.")`
##########
providers/cncf/kubernetes/tests/system/cncf/kubernetes/example_kubernetes_pod_exec.py:
##########
@@ -0,0 +1,134 @@
+# 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.
+"""Example Dag for executing a command in an existing Kubernetes Pod."""
+
+from __future__ import annotations
+
+import os
+import time
+from datetime import datetime
+
+from kubernetes.client.rest import ApiException
+
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator
+from airflow.providers.cncf.kubernetes.operators.resource import (
+ KubernetesCreateResourceOperator,
+ KubernetesDeleteResourceOperator,
+)
+from airflow.sdk import DAG, TriggerRule, task
+
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID", "default").lower().replace("_", "-")
+DAG_ID = "example_kubernetes_pod_exec_operator"
+NAMESPACE = "default"
+POD_NAME = f"airflow-pod-exec-{ENV_ID}"
+CONTAINER_NAME = "worker"
+EXPECTED_OUTPUT = "command executed in existing pod"
+
+pod_conf = f"""
+apiVersion: v1
+kind: Pod
+metadata:
+ name: {POD_NAME}
+ namespace: {NAMESPACE}
+spec:
+ restartPolicy: Never
+ containers:
+ - name: {CONTAINER_NAME}
+ image: busybox:1.38.0
+ command: ["sleep", "3600"]
+"""
+
+
+@task
+def wait_for_running_pod() -> None:
Review Comment:
Is there an existing sensor or some other abstraction you can use instead of creating your own polling loop? This looks quite heavy for an example DAG.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod_exec.py:
##########
@@ -0,0 +1,318 @@
+# 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
+
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+from kubernetes.client.rest import ApiException
+from kubernetes.stream.ws_client import WSClient
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator
+
+MODULE = "airflow.providers.cncf.kubernetes.operators.pod_exec"
+
+
+def create_pod(
+ *,
+ container_names: tuple[str, ...] = ("main",),
+ annotations: dict[str, str] | None = None,
+ phase: str = "Running",
+ container_statuses=None,
+ with_spec: bool = True,
+ with_status: bool = True,
+):
+ if container_statuses is None:
+ container_statuses = [
+ SimpleNamespace(name=name, state=SimpleNamespace(running=object())) for name in container_names
+ ]
+ spec = SimpleNamespace(containers=[SimpleNamespace(name=name) for name in container_names])
+ status = SimpleNamespace(phase=phase, container_statuses=container_statuses)
+ return SimpleNamespace(
+ metadata=SimpleNamespace(annotations=annotations),
+ spec=spec if with_spec else None,
+ status=status if with_status else None,
+ )
+
+
+def create_exec_client(*, stdout=(), stderr=(), returncode=0):
+ exec_client = mock.MagicMock(spec=WSClient)
+ exec_client.is_open.side_effect = [True, False]
+ exec_client.peek_stdout.side_effect = [*stdout, ""]
+ exec_client.read_stdout.side_effect = stdout
+ exec_client.peek_stderr.side_effect = [*stderr, ""]
+ exec_client.read_stderr.side_effect = stderr
+ exec_client.returncode = returncode
+ return exec_client
+
+
+def create_operator(*, pod=None, hook_namespace=None, **kwargs):
+ operator = KubernetesPodExecOperator(
+ task_id="exec",
+ pod_name="existing-pod",
+ command=["echo", "hello"],
+ **kwargs,
+ )
+ hook = mock.MagicMock(spec=KubernetesHook)
+ hook.get_namespace.return_value = hook_namespace
+ hook.get_pod.return_value = pod or create_pod()
+ operator.__dict__["hook"] = hook
+ return operator, hook
+
+
+class TestKubernetesPodExecOperator:
+ def test_template_fields(self):
+ assert set(KubernetesPodExecOperator.template_fields) == {
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ }
+
+ @mock.patch(f"{MODULE}.KubernetesHook", autospec=True)
+ def test_hook_configuration(self, kubernetes_hook_mock):
+ operator = KubernetesPodExecOperator(
+ task_id="exec",
+ pod_name="existing-pod",
+ command=["date"],
+ kubernetes_conn_id="kubernetes_test",
+ in_cluster=True,
+ config_file="/tmp/kubeconfig",
+ cluster_context="test-context",
+ )
+
+ assert operator.hook is kubernetes_hook_mock.return_value
+ kubernetes_hook_mock.assert_called_once_with(
+ conn_id="kubernetes_test",
+ in_cluster=True,
+ config_file="/tmp/kubeconfig",
+ cluster_context="test-context",
+ )
+
+ @pytest.mark.parametrize(
+ ("do_xcom_push", "expected_result"),
+ [(False, None), (True, "hello\nworld\n")],
+ )
+ @mock.patch(f"{MODULE}.KubernetesPodExecOperator.log", spec=["info", "warning"])
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_execute(self, kubernetes_stream_mock, log_mock, do_xcom_push, expected_result):
+ operator, hook = create_operator(
+ namespace="test-namespace",
+ container_name="main",
+ do_xcom_push=do_xcom_push,
+ )
+ exec_client = create_exec_client(stdout=("hello\n", "world\n"), stderr=("warning\n",), returncode=0)
+ kubernetes_stream_mock.return_value = exec_client
+
+ result = operator.execute(context={})
+
+ assert result == expected_result
+ hook.get_pod.assert_called_once_with(name="existing-pod", namespace="test-namespace")
+ kubernetes_stream_mock.assert_called_once_with(
+ hook.core_v1_client.connect_get_namespaced_pod_exec,
+ name="existing-pod",
+ namespace="test-namespace",
+ container="main",
+ command=["echo", "hello"],
+ stdin=False,
+ stdout=True,
+ stderr=True,
+ tty=False,
+ _preload_content=False,
+ )
+ exec_client.update.assert_called_once_with(timeout=1)
+ assert exec_client.read_stdout.call_count == 2
+ exec_client.read_stderr.assert_called_once_with()
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+ log_mock.info.assert_has_calls(
+ [
+ mock.call("[%s] %s", "stdout", "hello"),
+ mock.call("[%s] %s", "stdout", "world"),
+ ]
+ )
+ log_mock.warning.assert_called_once_with("[%s] %s", "stderr", "warning")
+
+ @pytest.mark.parametrize(
+ ("namespace", "hook_namespace", "expected_namespace"),
+ [
+ ("task-namespace", "connection-namespace", "task-namespace"),
+ (None, "connection-namespace", "connection-namespace"),
+ (None, None, KubernetesHook.DEFAULT_NAMESPACE),
+ ],
+ )
+ def test_resolve_namespace(self, namespace, hook_namespace, expected_namespace):
+ operator, _ = create_operator(namespace=namespace, hook_namespace=hook_namespace)
+
+ assert operator._resolve_namespace() == expected_namespace
+
+ @pytest.mark.parametrize(
+ ("container_name", "annotations", "expected_container"),
+ [
+ ("secondary", {"kubectl.kubernetes.io/default-container": "main"}, "secondary"),
+ (None, {"kubectl.kubernetes.io/default-container": "secondary"}, "secondary"),
+ (None, {"kubectl.kubernetes.io/default-container": "missing"}, "main"),
+ (None, None, "main"),
+ ],
+ )
+ def test_resolve_container_name(self, container_name, annotations, expected_container):
+ pod = create_pod(container_names=("main", "secondary"), annotations=annotations)
+ operator, _ = create_operator(pod=pod, container_name=container_name)
+
+ assert operator._resolve_container_name(pod) == expected_container
+
+ @pytest.mark.parametrize("with_spec", [False, True])
+ def test_rejects_pod_without_containers(self, with_spec):
+ pod = create_pod(container_names=(), with_spec=with_spec)
+ operator, _ = create_operator(pod=pod)
+
+ with pytest.raises(PodExecException, match="does not define any containers"):
+ operator._resolve_container_name(pod)
+
+ def test_rejects_unknown_container(self):
+ pod = create_pod()
+ operator, _ = create_operator(pod=pod, container_name="missing")
+
+ with pytest.raises(PodExecException, match="does not exist"):
+ operator._resolve_container_name(pod)
+
+ @pytest.mark.parametrize(
+ ("pod", "expected_message"),
+ [
+ (create_pod(with_status=False), "phase None"),
+ (create_pod(phase="Pending"), "phase 'Pending'"),
+ (create_pod(container_statuses=[]), "is not running"),
+ (
+ create_pod(container_statuses=[SimpleNamespace(name="main", state=None)]),
+ "is not running",
+ ),
+ (
+ create_pod(
+ container_statuses=[SimpleNamespace(name="main", state=SimpleNamespace(running=None))]
+ ),
+ "is not running",
+ ),
+ ],
+ )
+ def test_rejects_unavailable_target(self, pod, expected_message):
+ operator, _ = create_operator(pod=pod)
+
+ with pytest.raises(PodExecException, match=expected_message):
+ operator._validate_container_is_running(pod, "main")
+
+ @pytest.mark.parametrize(
+ ("command", "error", "expected_message"),
+ [
+ ("echo hello", TypeError, "sequence of strings"),
+ (42, TypeError, "sequence of strings"),
+ ([], ValueError, "at least one element"),
+ (["echo", 42], TypeError, "Every element"),
+ ],
+ )
+ def test_rejects_invalid_command(self, command, error, expected_message):
+ operator, _ = create_operator()
+ operator.command = command
+
+ with pytest.raises(error, match=expected_message):
+ operator.execute(context={})
+
+ def test_rejects_empty_pod_name(self):
+ operator, _ = create_operator()
+ operator.pod_name = ""
+
+ with pytest.raises(ValueError, match="must not be empty"):
+ operator.execute(context={})
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_wraps_pod_read_error(self, kubernetes_stream_mock):
+ operator, hook = create_operator(namespace="test-namespace")
+ hook.get_pod.side_effect = ApiException(status=404, reason="Not Found")
+
+ with pytest.raises(PodExecException, match="Unable to read pod.*Not Found"):
+ operator.execute(context={})
+
+ kubernetes_stream_mock.assert_not_called()
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_wraps_exec_api_error(self, kubernetes_stream_mock):
+ operator, _ = create_operator(namespace="test-namespace")
+ kubernetes_stream_mock.side_effect = ApiException(status=403, reason="Forbidden")
+
+ with pytest.raises(PodExecException, match="Unable to execute command.*Forbidden"):
+ operator.execute(context={})
+
+ assert operator._exec_client is None
+
+ @pytest.mark.parametrize(
+ ("returncode", "expected_message"),
+ [(None, "without reporting an exit code"), (17, "failed with exit code 17")],
+ )
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_rejects_unsuccessful_command(self, kubernetes_stream_mock, returncode, expected_message):
+ operator, _ = create_operator(namespace="test-namespace")
+ exec_client = create_exec_client(returncode=returncode)
+ kubernetes_stream_mock.return_value = exec_client
+
+ with pytest.raises(PodExecException, match=expected_message):
+ operator.execute(context={})
+
+ exec_client.close.assert_called_once_with()
+
+ def test_on_kill_closes_active_connection(self):
+ operator, _ = create_operator()
+ exec_client = mock.MagicMock(spec=WSClient)
+ operator._exec_client = exec_client
+
+ operator.on_kill()
+
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_on_kill_while_consuming_output(self, kubernetes_stream_mock):
+ operator, _ = create_operator(namespace="test-namespace")
+ exec_client = create_exec_client()
+ exec_client.is_open.side_effect = lambda: (operator.on_kill(), False)[1]
+ kubernetes_stream_mock.return_value = exec_client
+
+ assert operator.execute(context={}) is None
+
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+
+ def test_on_kill_without_active_connection(self):
+ operator, _ = create_operator()
+
+ operator.on_kill()
+
+ assert operator._exec_client is None
+
+ def test_close_error_does_not_mask_task_shutdown(self):
+ operator, _ = create_operator()
+ exec_client = mock.MagicMock(spec=WSClient)
+ exec_client.close.side_effect = RuntimeError("connection already closed")
+ operator._exec_client = exec_client
+
+ operator.on_kill()
+
+ assert operator._exec_client is None
Review Comment:
Could we add a test where `_consume_output()`/`exec_client.update() `raises after the connection has been assigned? The finally block is intended to guarantee cleanup during execution failures, but the current tests only cover normal completion, unsuccessful return codes, and direct `on_kill() `calls.
##########
providers/cncf/kubernetes/docs/operators.rst:
##########
@@ -19,6 +19,36 @@
.. contents:: Table of Contents
:depth: 2
+.. _howto/operator:KubernetesPodExecOperator:
+
+KubernetesPodExecOperator
+=========================
+
+The :class:`~airflow.providers.cncf.kubernetes.operators.pod_exec.KubernetesPodExecOperator`
+executes a command in a running container of an existing Kubernetes Pod. It does not create,
+restart, or delete the target Pod.
+
+.. exampleinclude:: /../tests/system/cncf/kubernetes/example_kubernetes_pod_exec.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_k8s_pod_exec]
+ :end-before: [END howto_operator_k8s_pod_exec]
+
+Commands are executed directly rather than through a shell. Include a shell explicitly when using
+pipes, redirects, variable expansion, or other shell features.
+Standard output and standard error are streamed to the task log. Set ``do_xcom_push=True`` to
+also return standard output through XCom.
+
+The target Pod and container must already be running. When ``container_name`` is omitted, the
+operator uses the ``kubectl.kubernetes.io/default-container`` annotation when present, or the
+first container otherwise. API-visible static Pods are supported through their mirror Pod name;
+components that are not exposed by the Kubernetes API cannot be targeted. The Kubernetes connection
+requires ``get`` access to ``pods`` and ``pods/exec``; see :doc:`kubernetes_rbac`.
+
+If the task or its worker stops while the command is running, Airflow closes the exec connection
+but does not modify the target Pod. Kubernetes cannot always determine whether a command completed
+before a connection failure, so configure task retries only when the command is safe to repeat.
+
Review Comment:
I think this should be moved after the section for `KubernetesPodOperator`.
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
+ def _log_output(self, output: str, *, stream_name: str) -> None:
+ log_method = self.log.warning if stream_name == "stderr" else self.log.info
Review Comment:
Should all stderr output be emitted at warning level? Many commands routinely use stderr for progress or informational messages, so this could generate misleading warning records. Logging both streams at `INFO` while preserving the [stderr] marker may be more appropriate.
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod_exec.py:
##########
@@ -0,0 +1,318 @@
+# 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
+
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+from kubernetes.client.rest import ApiException
+from kubernetes.stream.ws_client import WSClient
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator
+
+MODULE = "airflow.providers.cncf.kubernetes.operators.pod_exec"
+
+
+def create_pod(
+ *,
+ container_names: tuple[str, ...] = ("main",),
+ annotations: dict[str, str] | None = None,
+ phase: str = "Running",
+ container_statuses=None,
+ with_spec: bool = True,
+ with_status: bool = True,
+):
+ if container_statuses is None:
+ container_statuses = [
+ SimpleNamespace(name=name, state=SimpleNamespace(running=object())) for name in container_names
+ ]
+ spec = SimpleNamespace(containers=[SimpleNamespace(name=name) for name in container_names])
+ status = SimpleNamespace(phase=phase, container_statuses=container_statuses)
+ return SimpleNamespace(
+ metadata=SimpleNamespace(annotations=annotations),
+ spec=spec if with_spec else None,
+ status=status if with_status else None,
+ )
+
+
+def create_exec_client(*, stdout=(), stderr=(), returncode=0):
+ exec_client = mock.MagicMock(spec=WSClient)
+ exec_client.is_open.side_effect = [True, False]
+ exec_client.peek_stdout.side_effect = [*stdout, ""]
+ exec_client.read_stdout.side_effect = stdout
+ exec_client.peek_stderr.side_effect = [*stderr, ""]
+ exec_client.read_stderr.side_effect = stderr
+ exec_client.returncode = returncode
+ return exec_client
+
+
+def create_operator(*, pod=None, hook_namespace=None, **kwargs):
+ operator = KubernetesPodExecOperator(
+ task_id="exec",
+ pod_name="existing-pod",
+ command=["echo", "hello"],
+ **kwargs,
+ )
+ hook = mock.MagicMock(spec=KubernetesHook)
+ hook.get_namespace.return_value = hook_namespace
+ hook.get_pod.return_value = pod or create_pod()
+ operator.__dict__["hook"] = hook
+ return operator, hook
+
+
+class TestKubernetesPodExecOperator:
+ def test_template_fields(self):
+ assert set(KubernetesPodExecOperator.template_fields) == {
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ }
+
+ @mock.patch(f"{MODULE}.KubernetesHook", autospec=True)
+ def test_hook_configuration(self, kubernetes_hook_mock):
+ operator = KubernetesPodExecOperator(
+ task_id="exec",
+ pod_name="existing-pod",
+ command=["date"],
+ kubernetes_conn_id="kubernetes_test",
+ in_cluster=True,
+ config_file="/tmp/kubeconfig",
+ cluster_context="test-context",
+ )
+
+ assert operator.hook is kubernetes_hook_mock.return_value
+ kubernetes_hook_mock.assert_called_once_with(
+ conn_id="kubernetes_test",
+ in_cluster=True,
+ config_file="/tmp/kubeconfig",
+ cluster_context="test-context",
+ )
+
+ @pytest.mark.parametrize(
+ ("do_xcom_push", "expected_result"),
+ [(False, None), (True, "hello\nworld\n")],
+ )
+ @mock.patch(f"{MODULE}.KubernetesPodExecOperator.log", spec=["info", "warning"])
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_execute(self, kubernetes_stream_mock, log_mock, do_xcom_push, expected_result):
+ operator, hook = create_operator(
+ namespace="test-namespace",
+ container_name="main",
+ do_xcom_push=do_xcom_push,
+ )
+ exec_client = create_exec_client(stdout=("hello\n", "world\n"), stderr=("warning\n",), returncode=0)
+ kubernetes_stream_mock.return_value = exec_client
+
+ result = operator.execute(context={})
+
+ assert result == expected_result
+ hook.get_pod.assert_called_once_with(name="existing-pod", namespace="test-namespace")
+ kubernetes_stream_mock.assert_called_once_with(
+ hook.core_v1_client.connect_get_namespaced_pod_exec,
+ name="existing-pod",
+ namespace="test-namespace",
+ container="main",
+ command=["echo", "hello"],
+ stdin=False,
+ stdout=True,
+ stderr=True,
+ tty=False,
+ _preload_content=False,
+ )
+ exec_client.update.assert_called_once_with(timeout=1)
+ assert exec_client.read_stdout.call_count == 2
+ exec_client.read_stderr.assert_called_once_with()
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+ log_mock.info.assert_has_calls(
+ [
+ mock.call("[%s] %s", "stdout", "hello"),
+ mock.call("[%s] %s", "stdout", "world"),
+ ]
+ )
+ log_mock.warning.assert_called_once_with("[%s] %s", "stderr", "warning")
+
+ @pytest.mark.parametrize(
+ ("namespace", "hook_namespace", "expected_namespace"),
+ [
+ ("task-namespace", "connection-namespace", "task-namespace"),
+ (None, "connection-namespace", "connection-namespace"),
+ (None, None, KubernetesHook.DEFAULT_NAMESPACE),
+ ],
+ )
+ def test_resolve_namespace(self, namespace, hook_namespace, expected_namespace):
+ operator, _ = create_operator(namespace=namespace, hook_namespace=hook_namespace)
+
+ assert operator._resolve_namespace() == expected_namespace
+
+ @pytest.mark.parametrize(
+ ("container_name", "annotations", "expected_container"),
+ [
+ ("secondary", {"kubectl.kubernetes.io/default-container": "main"}, "secondary"),
+ (None, {"kubectl.kubernetes.io/default-container": "secondary"}, "secondary"),
+ (None, {"kubectl.kubernetes.io/default-container": "missing"}, "main"),
+ (None, None, "main"),
+ ],
+ )
+ def test_resolve_container_name(self, container_name, annotations, expected_container):
+ pod = create_pod(container_names=("main", "secondary"), annotations=annotations)
+ operator, _ = create_operator(pod=pod, container_name=container_name)
+
+ assert operator._resolve_container_name(pod) == expected_container
+
+ @pytest.mark.parametrize("with_spec", [False, True])
+ def test_rejects_pod_without_containers(self, with_spec):
+ pod = create_pod(container_names=(), with_spec=with_spec)
+ operator, _ = create_operator(pod=pod)
+
+ with pytest.raises(PodExecException, match="does not define any containers"):
+ operator._resolve_container_name(pod)
+
+ def test_rejects_unknown_container(self):
+ pod = create_pod()
+ operator, _ = create_operator(pod=pod, container_name="missing")
+
+ with pytest.raises(PodExecException, match="does not exist"):
+ operator._resolve_container_name(pod)
+
+ @pytest.mark.parametrize(
+ ("pod", "expected_message"),
+ [
+ (create_pod(with_status=False), "phase None"),
+ (create_pod(phase="Pending"), "phase 'Pending'"),
+ (create_pod(container_statuses=[]), "is not running"),
+ (
+ create_pod(container_statuses=[SimpleNamespace(name="main", state=None)]),
+ "is not running",
+ ),
+ (
+ create_pod(
+ container_statuses=[SimpleNamespace(name="main", state=SimpleNamespace(running=None))]
+ ),
+ "is not running",
+ ),
+ ],
+ )
+ def test_rejects_unavailable_target(self, pod, expected_message):
+ operator, _ = create_operator(pod=pod)
+
+ with pytest.raises(PodExecException, match=expected_message):
+ operator._validate_container_is_running(pod, "main")
+
+ @pytest.mark.parametrize(
+ ("command", "error", "expected_message"),
+ [
+ ("echo hello", TypeError, "sequence of strings"),
+ (42, TypeError, "sequence of strings"),
+ ([], ValueError, "at least one element"),
+ (["echo", 42], TypeError, "Every element"),
+ ],
+ )
+ def test_rejects_invalid_command(self, command, error, expected_message):
+ operator, _ = create_operator()
+ operator.command = command
+
+ with pytest.raises(error, match=expected_message):
+ operator.execute(context={})
+
+ def test_rejects_empty_pod_name(self):
+ operator, _ = create_operator()
+ operator.pod_name = ""
+
+ with pytest.raises(ValueError, match="must not be empty"):
+ operator.execute(context={})
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_wraps_pod_read_error(self, kubernetes_stream_mock):
+ operator, hook = create_operator(namespace="test-namespace")
+ hook.get_pod.side_effect = ApiException(status=404, reason="Not Found")
+
+ with pytest.raises(PodExecException, match="Unable to read pod.*Not Found"):
+ operator.execute(context={})
+
+ kubernetes_stream_mock.assert_not_called()
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_wraps_exec_api_error(self, kubernetes_stream_mock):
+ operator, _ = create_operator(namespace="test-namespace")
+ kubernetes_stream_mock.side_effect = ApiException(status=403, reason="Forbidden")
+
+ with pytest.raises(PodExecException, match="Unable to execute command.*Forbidden"):
+ operator.execute(context={})
+
+ assert operator._exec_client is None
+
+ @pytest.mark.parametrize(
+ ("returncode", "expected_message"),
+ [(None, "without reporting an exit code"), (17, "failed with exit code 17")],
+ )
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_rejects_unsuccessful_command(self, kubernetes_stream_mock, returncode, expected_message):
+ operator, _ = create_operator(namespace="test-namespace")
+ exec_client = create_exec_client(returncode=returncode)
+ kubernetes_stream_mock.return_value = exec_client
+
+ with pytest.raises(PodExecException, match=expected_message):
+ operator.execute(context={})
+
+ exec_client.close.assert_called_once_with()
+
+ def test_on_kill_closes_active_connection(self):
+ operator, _ = create_operator()
+ exec_client = mock.MagicMock(spec=WSClient)
+ operator._exec_client = exec_client
+
+ operator.on_kill()
+
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_on_kill_while_consuming_output(self, kubernetes_stream_mock):
+ operator, _ = create_operator(namespace="test-namespace")
+ exec_client = create_exec_client()
+ exec_client.is_open.side_effect = lambda: (operator.on_kill(), False)[1]
+ kubernetes_stream_mock.return_value = exec_client
+
+ assert operator.execute(context={}) is None
+
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+
+ def test_on_kill_without_active_connection(self):
+ operator, _ = create_operator()
+
+ operator.on_kill()
+
+ assert operator._exec_client is None
+
+ def test_close_error_does_not_mask_task_shutdown(self):
Review Comment:
It would be useful to assert that the suppressed exception is logged, since logging is the only resulting behaviour.
##########
providers/cncf/kubernetes/tests/system/cncf/kubernetes/example_kubernetes_pod_exec.py:
##########
@@ -0,0 +1,134 @@
+# 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.
+"""Example Dag for executing a command in an existing Kubernetes Pod."""
+
+from __future__ import annotations
+
+import os
+import time
+from datetime import datetime
+
+from kubernetes.client.rest import ApiException
+
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator
+from airflow.providers.cncf.kubernetes.operators.resource import (
+ KubernetesCreateResourceOperator,
+ KubernetesDeleteResourceOperator,
+)
+from airflow.sdk import DAG, TriggerRule, task
+
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID", "default").lower().replace("_", "-")
+DAG_ID = "example_kubernetes_pod_exec_operator"
+NAMESPACE = "default"
+POD_NAME = f"airflow-pod-exec-{ENV_ID}"
+CONTAINER_NAME = "worker"
+EXPECTED_OUTPUT = "command executed in existing pod"
+
+pod_conf = f"""
+apiVersion: v1
+kind: Pod
+metadata:
+ name: {POD_NAME}
+ namespace: {NAMESPACE}
+spec:
+ restartPolicy: Never
+ containers:
+ - name: {CONTAINER_NAME}
+ image: busybox:1.38.0
+ command: ["sleep", "3600"]
+"""
+
+
+@task
+def wait_for_running_pod() -> None:
+ hook = KubernetesHook()
+ deadline = time.monotonic() + 120
+ last_phase = None
+
+ while time.monotonic() < deadline:
+ try:
+ pod = hook.get_pod(name=POD_NAME, namespace=NAMESPACE)
+ except ApiException as error:
+ if error.status != 404:
+ raise
+ else:
+ last_phase = pod.status.phase if pod.status else None
+ container_statuses = pod.status.container_statuses if pod.status else None
+ container_status = next(
+ (status for status in container_statuses or [] if status.name == CONTAINER_NAME), None
+ )
+ if (
+ last_phase == "Running"
Review Comment:
Use `PodPhase.RUNNING` here too.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on PR #71244:
URL: https://github.com/apache/airflow/pull/71244#issuecomment-5233566514
@SameerMesiah97 Could you please review it when you have a chance? Thank you!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889732141
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
Review Comment:
Done!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889732953
##########
providers/cncf/kubernetes/docs/operators.rst:
##########
@@ -19,6 +19,36 @@
.. contents:: Table of Contents
:depth: 2
+.. _howto/operator:KubernetesPodExecOperator:
+
+KubernetesPodExecOperator
+=========================
+
+The :class:`~airflow.providers.cncf.kubernetes.operators.pod_exec.KubernetesPodExecOperator`
+executes a command in a running container of an existing Kubernetes Pod. It does not create,
+restart, or delete the target Pod.
+
+.. exampleinclude:: /../tests/system/cncf/kubernetes/example_kubernetes_pod_exec.py
+ :language: python
+ :dedent: 4
+ :start-after: [START howto_operator_k8s_pod_exec]
+ :end-before: [END howto_operator_k8s_pod_exec]
+
+Commands are executed directly rather than through a shell. Include a shell explicitly when using
+pipes, redirects, variable expansion, or other shell features.
+Standard output and standard error are streamed to the task log. Set ``do_xcom_push=True`` to
+also return standard output through XCom.
+
+The target Pod and container must already be running. When ``container_name`` is omitted, the
+operator uses the ``kubectl.kubernetes.io/default-container`` annotation when present, or the
+first container otherwise. API-visible static Pods are supported through their mirror Pod name;
+components that are not exposed by the Kubernetes API cannot be targeted. The Kubernetes connection
+requires ``get`` access to ``pods`` and ``pods/exec``; see :doc:`kubernetes_rbac`.
+
+If the task or its worker stops while the command is running, Airflow closes the exec connection
+but does not modify the target Pod. Kubernetes cannot always determine whether a command completed
+before a connection failure, so configure task retries only when the command is safe to repeat.
+
Review Comment:
Done!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889736639
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
Review Comment:
Yes, done! Thanks
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
+ def _log_output(self, output: str, *, stream_name: str) -> None:
+ log_method = self.log.warning if stream_name == "stderr" else self.log.info
+ for line in output.splitlines():
+ log_method("[%s] %s", stream_name, line)
+
+ def _consume_output(self, exec_client: WSClient) -> str:
+ stdout_chunks: list[str] = []
+ while exec_client.is_open():
+ exec_client.update(timeout=1)
+ while exec_client.peek_stdout():
+ output = exec_client.read_stdout()
+ self._log_output(output, stream_name="stdout")
+ if self.do_xcom_push:
+ stdout_chunks.append(output)
+ while exec_client.peek_stderr():
+ self._log_output(exec_client.read_stderr(), stream_name="stderr")
+ return "".join(stdout_chunks)
+
+ def _close_exec_client(self) -> None:
+ exec_client = self._exec_client
+ self._exec_client = None
+ if exec_client is None:
+ return
+ try:
+ exec_client.close()
+ except Exception:
+ self.log.exception("Failed to close Kubernetes exec connection")
Review Comment:
Done!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "SameerMesiah97 (via GitHub)" <gi...@apache.org>.
SameerMesiah97 commented on PR #71244:
URL: https://github.com/apache/airflow/pull/71244#issuecomment-5431124948
> Although this can be implemented with `BashOperator` and `kubectl exec`, that moves additional integration logic to every user: installing and maintaining `kubectl` on the workers, distributing kubeconfig and cloud authentication helpers, handling shell quoting, and mapping Kubernetes errors and output into Airflow. This operator provides that behavior through `KubernetesHook` and Airflow connections. The provider already uses the `pods/exec` API internally, so this is not a new Kubernetes mechanism.
I think this is the strongest argument here. If using this operator saves time and effort for users who want this specific functionality, then I believe it is justified. But like I mentioned above, why only `exec`? Why not other commands like `log` or `watch`? It seems premature and restrictive to narrow it to just `exec`?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #71244:
URL: https://github.com/apache/airflow/pull/71244#issuecomment-5430723989
@SameerMesiah97 what are your thoughts on that?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on PR #71244:
URL: https://github.com/apache/airflow/pull/71244#issuecomment-5432458738
Combining `exec`, `logs`, and `watch` would give one operator several different responsibilities and completion contracts.
`pods/exec` already allows this operator to start any command available inside the container, stream its stdout and stderr, and wait for its exit code. In contrast, `pods/log` only consumes a log stream, while `watch` observes resource state and requires a condition to determine when the task should finish.
This also differs from `KubernetesPodOperator`: KPO creates a Pod and configures the workload as the container's main process. It then monitors the Pod lifecycle.`KubernetesPodExecOperator` starts a secondary process through `pods/exec`, waits only for that process, and leaves the existing Pod running.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on PR #71244:
URL: https://github.com/apache/airflow/pull/71244#issuecomment-5221917597
@kaxil
> What's the User story and real-world use cases for this?
Thanks for asking. The user story is:
> As an Airflow user who already manages a long-running Kubernetes Pod, I want to execute commands in one
of its containers as Airflow tasks, without Airflow creating, restarting, or deleting that Pod.
Our concrete use case is running dbt on EKS. Today, our workflow starts a new Pod, initializes the dbt project and its dependencies, and then executes a command such as `dbt run`. A significant part of the execution time is spent scheduling the Pod and initializing the environment.
We want to manage that Pod separately and keep it warm, then use Airflow to execute ad hoc commands such as `dbt run`, `dbt test`, or `dbt build`. This avoids the repeated cold-start and initialization cost.
More generally, this is useful whenever a Pod’s lifecycle is managed outside the task and its environment is already initialized. The operator exposes the equivalent of `kubectl exec` as an Airflow task, including Airflow logging and optional XCom output, without assuming ownership of the Pod lifecycle.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck merged PR #71244:
URL: https://github.com/apache/airflow/pull/71244
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889717292
##########
providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod_exec.py:
##########
@@ -0,0 +1,318 @@
+# 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
+
+from types import SimpleNamespace
+from unittest import mock
+
+import pytest
+from kubernetes.client.rest import ApiException
+from kubernetes.stream.ws_client import WSClient
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator
+
+MODULE = "airflow.providers.cncf.kubernetes.operators.pod_exec"
+
+
+def create_pod(
+ *,
+ container_names: tuple[str, ...] = ("main",),
+ annotations: dict[str, str] | None = None,
+ phase: str = "Running",
+ container_statuses=None,
+ with_spec: bool = True,
+ with_status: bool = True,
+):
+ if container_statuses is None:
+ container_statuses = [
+ SimpleNamespace(name=name, state=SimpleNamespace(running=object())) for name in container_names
+ ]
+ spec = SimpleNamespace(containers=[SimpleNamespace(name=name) for name in container_names])
+ status = SimpleNamespace(phase=phase, container_statuses=container_statuses)
+ return SimpleNamespace(
+ metadata=SimpleNamespace(annotations=annotations),
+ spec=spec if with_spec else None,
+ status=status if with_status else None,
+ )
+
+
+def create_exec_client(*, stdout=(), stderr=(), returncode=0):
+ exec_client = mock.MagicMock(spec=WSClient)
+ exec_client.is_open.side_effect = [True, False]
+ exec_client.peek_stdout.side_effect = [*stdout, ""]
+ exec_client.read_stdout.side_effect = stdout
+ exec_client.peek_stderr.side_effect = [*stderr, ""]
+ exec_client.read_stderr.side_effect = stderr
+ exec_client.returncode = returncode
+ return exec_client
+
+
+def create_operator(*, pod=None, hook_namespace=None, **kwargs):
+ operator = KubernetesPodExecOperator(
+ task_id="exec",
+ pod_name="existing-pod",
+ command=["echo", "hello"],
+ **kwargs,
+ )
+ hook = mock.MagicMock(spec=KubernetesHook)
+ hook.get_namespace.return_value = hook_namespace
+ hook.get_pod.return_value = pod or create_pod()
+ operator.__dict__["hook"] = hook
+ return operator, hook
+
+
+class TestKubernetesPodExecOperator:
+ def test_template_fields(self):
+ assert set(KubernetesPodExecOperator.template_fields) == {
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ }
+
+ @mock.patch(f"{MODULE}.KubernetesHook", autospec=True)
+ def test_hook_configuration(self, kubernetes_hook_mock):
+ operator = KubernetesPodExecOperator(
+ task_id="exec",
+ pod_name="existing-pod",
+ command=["date"],
+ kubernetes_conn_id="kubernetes_test",
+ in_cluster=True,
+ config_file="/tmp/kubeconfig",
+ cluster_context="test-context",
+ )
+
+ assert operator.hook is kubernetes_hook_mock.return_value
+ kubernetes_hook_mock.assert_called_once_with(
+ conn_id="kubernetes_test",
+ in_cluster=True,
+ config_file="/tmp/kubeconfig",
+ cluster_context="test-context",
+ )
+
+ @pytest.mark.parametrize(
+ ("do_xcom_push", "expected_result"),
+ [(False, None), (True, "hello\nworld\n")],
+ )
+ @mock.patch(f"{MODULE}.KubernetesPodExecOperator.log", spec=["info", "warning"])
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_execute(self, kubernetes_stream_mock, log_mock, do_xcom_push, expected_result):
+ operator, hook = create_operator(
+ namespace="test-namespace",
+ container_name="main",
+ do_xcom_push=do_xcom_push,
+ )
+ exec_client = create_exec_client(stdout=("hello\n", "world\n"), stderr=("warning\n",), returncode=0)
+ kubernetes_stream_mock.return_value = exec_client
+
+ result = operator.execute(context={})
+
+ assert result == expected_result
+ hook.get_pod.assert_called_once_with(name="existing-pod", namespace="test-namespace")
+ kubernetes_stream_mock.assert_called_once_with(
+ hook.core_v1_client.connect_get_namespaced_pod_exec,
+ name="existing-pod",
+ namespace="test-namespace",
+ container="main",
+ command=["echo", "hello"],
+ stdin=False,
+ stdout=True,
+ stderr=True,
+ tty=False,
+ _preload_content=False,
+ )
+ exec_client.update.assert_called_once_with(timeout=1)
+ assert exec_client.read_stdout.call_count == 2
+ exec_client.read_stderr.assert_called_once_with()
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+ log_mock.info.assert_has_calls(
+ [
+ mock.call("[%s] %s", "stdout", "hello"),
+ mock.call("[%s] %s", "stdout", "world"),
+ ]
+ )
+ log_mock.warning.assert_called_once_with("[%s] %s", "stderr", "warning")
+
+ @pytest.mark.parametrize(
+ ("namespace", "hook_namespace", "expected_namespace"),
+ [
+ ("task-namespace", "connection-namespace", "task-namespace"),
+ (None, "connection-namespace", "connection-namespace"),
+ (None, None, KubernetesHook.DEFAULT_NAMESPACE),
+ ],
+ )
+ def test_resolve_namespace(self, namespace, hook_namespace, expected_namespace):
+ operator, _ = create_operator(namespace=namespace, hook_namespace=hook_namespace)
+
+ assert operator._resolve_namespace() == expected_namespace
+
+ @pytest.mark.parametrize(
+ ("container_name", "annotations", "expected_container"),
+ [
+ ("secondary", {"kubectl.kubernetes.io/default-container": "main"}, "secondary"),
+ (None, {"kubectl.kubernetes.io/default-container": "secondary"}, "secondary"),
+ (None, {"kubectl.kubernetes.io/default-container": "missing"}, "main"),
+ (None, None, "main"),
+ ],
+ )
+ def test_resolve_container_name(self, container_name, annotations, expected_container):
+ pod = create_pod(container_names=("main", "secondary"), annotations=annotations)
+ operator, _ = create_operator(pod=pod, container_name=container_name)
+
+ assert operator._resolve_container_name(pod) == expected_container
+
+ @pytest.mark.parametrize("with_spec", [False, True])
+ def test_rejects_pod_without_containers(self, with_spec):
+ pod = create_pod(container_names=(), with_spec=with_spec)
+ operator, _ = create_operator(pod=pod)
+
+ with pytest.raises(PodExecException, match="does not define any containers"):
+ operator._resolve_container_name(pod)
+
+ def test_rejects_unknown_container(self):
+ pod = create_pod()
+ operator, _ = create_operator(pod=pod, container_name="missing")
+
+ with pytest.raises(PodExecException, match="does not exist"):
+ operator._resolve_container_name(pod)
+
+ @pytest.mark.parametrize(
+ ("pod", "expected_message"),
+ [
+ (create_pod(with_status=False), "phase None"),
+ (create_pod(phase="Pending"), "phase 'Pending'"),
+ (create_pod(container_statuses=[]), "is not running"),
+ (
+ create_pod(container_statuses=[SimpleNamespace(name="main", state=None)]),
+ "is not running",
+ ),
+ (
+ create_pod(
+ container_statuses=[SimpleNamespace(name="main", state=SimpleNamespace(running=None))]
+ ),
+ "is not running",
+ ),
+ ],
+ )
+ def test_rejects_unavailable_target(self, pod, expected_message):
+ operator, _ = create_operator(pod=pod)
+
+ with pytest.raises(PodExecException, match=expected_message):
+ operator._validate_container_is_running(pod, "main")
+
+ @pytest.mark.parametrize(
+ ("command", "error", "expected_message"),
+ [
+ ("echo hello", TypeError, "sequence of strings"),
+ (42, TypeError, "sequence of strings"),
+ ([], ValueError, "at least one element"),
+ (["echo", 42], TypeError, "Every element"),
+ ],
+ )
+ def test_rejects_invalid_command(self, command, error, expected_message):
+ operator, _ = create_operator()
+ operator.command = command
+
+ with pytest.raises(error, match=expected_message):
+ operator.execute(context={})
+
+ def test_rejects_empty_pod_name(self):
+ operator, _ = create_operator()
+ operator.pod_name = ""
+
+ with pytest.raises(ValueError, match="must not be empty"):
+ operator.execute(context={})
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_wraps_pod_read_error(self, kubernetes_stream_mock):
+ operator, hook = create_operator(namespace="test-namespace")
+ hook.get_pod.side_effect = ApiException(status=404, reason="Not Found")
+
+ with pytest.raises(PodExecException, match="Unable to read pod.*Not Found"):
+ operator.execute(context={})
+
+ kubernetes_stream_mock.assert_not_called()
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_wraps_exec_api_error(self, kubernetes_stream_mock):
+ operator, _ = create_operator(namespace="test-namespace")
+ kubernetes_stream_mock.side_effect = ApiException(status=403, reason="Forbidden")
+
+ with pytest.raises(PodExecException, match="Unable to execute command.*Forbidden"):
+ operator.execute(context={})
+
+ assert operator._exec_client is None
+
+ @pytest.mark.parametrize(
+ ("returncode", "expected_message"),
+ [(None, "without reporting an exit code"), (17, "failed with exit code 17")],
+ )
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_rejects_unsuccessful_command(self, kubernetes_stream_mock, returncode, expected_message):
+ operator, _ = create_operator(namespace="test-namespace")
+ exec_client = create_exec_client(returncode=returncode)
+ kubernetes_stream_mock.return_value = exec_client
+
+ with pytest.raises(PodExecException, match=expected_message):
+ operator.execute(context={})
+
+ exec_client.close.assert_called_once_with()
+
+ def test_on_kill_closes_active_connection(self):
+ operator, _ = create_operator()
+ exec_client = mock.MagicMock(spec=WSClient)
+ operator._exec_client = exec_client
+
+ operator.on_kill()
+
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+
+ @mock.patch(f"{MODULE}.kubernetes_stream", autospec=True)
+ def test_on_kill_while_consuming_output(self, kubernetes_stream_mock):
+ operator, _ = create_operator(namespace="test-namespace")
+ exec_client = create_exec_client()
+ exec_client.is_open.side_effect = lambda: (operator.on_kill(), False)[1]
+ kubernetes_stream_mock.return_value = exec_client
+
+ assert operator.execute(context={}) is None
+
+ exec_client.close.assert_called_once_with()
+ assert operator._exec_client is None
+
+ def test_on_kill_without_active_connection(self):
+ operator, _ = create_operator()
+
+ operator.on_kill()
+
+ assert operator._exec_client is None
+
+ def test_close_error_does_not_mask_task_shutdown(self):
Review Comment:
Done, thanks! The test now asserts that the suppressed close exception is logged.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889723946
##########
providers/cncf/kubernetes/tests/system/cncf/kubernetes/example_kubernetes_pod_exec.py:
##########
@@ -0,0 +1,134 @@
+# 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.
+"""Example Dag for executing a command in an existing Kubernetes Pod."""
+
+from __future__ import annotations
+
+import os
+import time
+from datetime import datetime
+
+from kubernetes.client.rest import ApiException
+
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator
+from airflow.providers.cncf.kubernetes.operators.resource import (
+ KubernetesCreateResourceOperator,
+ KubernetesDeleteResourceOperator,
+)
+from airflow.sdk import DAG, TriggerRule, task
+
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID", "default").lower().replace("_", "-")
+DAG_ID = "example_kubernetes_pod_exec_operator"
+NAMESPACE = "default"
+POD_NAME = f"airflow-pod-exec-{ENV_ID}"
+CONTAINER_NAME = "worker"
+EXPECTED_OUTPUT = "command executed in existing pod"
+
+pod_conf = f"""
+apiVersion: v1
+kind: Pod
+metadata:
+ name: {POD_NAME}
+ namespace: {NAMESPACE}
+spec:
+ restartPolicy: Never
+ containers:
+ - name: {CONTAINER_NAME}
+ image: busybox:1.38.0
+ command: ["sleep", "3600"]
+"""
+
+
+@task
+def wait_for_running_pod() -> None:
+ hook = KubernetesHook()
+ deadline = time.monotonic() + 120
+ last_phase = None
+
+ while time.monotonic() < deadline:
+ try:
+ pod = hook.get_pod(name=POD_NAME, namespace=NAMESPACE)
+ except ApiException as error:
+ if error.status != 404:
+ raise
+ else:
+ last_phase = pod.status.phase if pod.status else None
+ container_statuses = pod.status.container_statuses if pod.status else None
+ container_status = next(
+ (status for status in container_statuses or [] if status.name == CONTAINER_NAME), None
+ )
+ if (
+ last_phase == "Running"
Review Comment:
Done!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889812846
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
+ def _log_output(self, output: str, *, stream_name: str) -> None:
+ log_method = self.log.warning if stream_name == "stderr" else self.log.info
+ for line in output.splitlines():
+ log_method("[%s] %s", stream_name, line)
+
+ def _consume_output(self, exec_client: WSClient) -> str:
+ stdout_chunks: list[str] = []
+ while exec_client.is_open():
+ exec_client.update(timeout=1)
+ while exec_client.peek_stdout():
+ output = exec_client.read_stdout()
+ self._log_output(output, stream_name="stdout")
+ if self.do_xcom_push:
+ stdout_chunks.append(output)
+ while exec_client.peek_stderr():
+ self._log_output(exec_client.read_stderr(), stream_name="stderr")
+ return "".join(stdout_chunks)
Review Comment:
Good point. I made XCom capture opt-in (do_xcom_push=False by default) and added a configurable max_xcom_output_size, enforced incrementally while stdout is streamed.
It defaults to 49,344 UTF-8 bytes, following the existing MAX_XCOM_SIZE precedent used by `GCSToLocalFilesystemOperator`
https://github.com/apache/airflow/blob/f760ac1c03b7277f3cb69071b8ead56d27ecbff7/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_local.py#L29-L30
and `GoogleApiToS3Operator`
https://github.com/apache/airflow/blob/f760ac1c03b7277f3cb69071b8ead56d27ecbff7/providers/amazon/src/airflow/providers/amazon/aws/transfers/google_api_to_s3.py#L36-L38
If the limit is exceeded, the task fails instead of retaining unbounded output or silently truncating it, and the exec connection is still cleaned up. I also added tests for invalid limits and multibyte output exceeding the configured byte limit.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "kaxil (via GitHub)" <gi...@apache.org>.
kaxil commented on PR #71244:
URL: https://github.com/apache/airflow/pull/71244#issuecomment-5221844368
> Add `KubernetesPodExecOperator` to execute commands in a running container of an existing Kubernetes Pod without creating, restarting, or deleting it. This supports pre-warmed and API-visible static Pods where avoiding startup latency is important.
>
> The operator validates the target Pod and container, streams stdout and stderr, supports optional stdout XCom, reports non-zero exit codes, and closes the exec connection when the task is killed. Provider metadata, RBAC guidance, and operator documentation are included.
>
> Tests:
>
> * `pytest providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod_exec.py -q` (32 passed)
> * Provider mypy check
> * Provider metadata and documentation validation
> * Pre-commit and manual checks
>
> ##### Was generative AI tooling used to co-author this PR?
> * [x] Yes — Codex (GPT-5)
>
> Generated-by: Codex (GPT-5) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
What's the User story and real-world use cases for this?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on PR #71244:
URL: https://github.com/apache/airflow/pull/71244#issuecomment-5444944521
@SameerMesiah97 Could you help me with a code review?
I’ve already tested it with the end-to-end system test and an Apache Magpie review.
Thank you! 🙏
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on PR #71244:
URL: https://github.com/apache/airflow/pull/71244#issuecomment-5430379619
Thanks for the feedback. Airflow commonly acts as a client of infrastructure it does not own: `SSHOperator` does not manage the server, SQL operators do not manage the database, and `SsmRunCommandOperator` does not manage the target instances. They own only the operation being executed. This operator follows the same model: Airflow owns the exec session, while another system owns the Pod lifecycle.
Although this can be implemented with `BashOperator` and `kubectl exec`, that moves additional integration logic to every user: installing and maintaining `kubectl` on the workers, distributing kubeconfig and cloud authentication helpers, handling shell quoting, and mapping Kubernetes errors and output into Airflow. This operator provides that behavior through `KubernetesHook` and Airflow connections. The provider already uses the `pods/exec` API internally, so this is not a new Kubernetes mechanism.
Keeping this provider-neutral also allows GKE- or EKS-specific operators to reuse it and add only their cloud authentication and cluster configuration, following the existing provider layering. I prefer keeping it separate from KPO because KPO owns the complete Pod lifecycle, while this operator deliberately treats the Pod as externally managed.
If the Pod disappears, restarts, or the connection fails, the task fails. A retry creates a new exec session; Airflow never creates, replaces, or deletes the Pod.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889801444
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod_exec.py:
##########
@@ -0,0 +1,247 @@
+# 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.
+"""Execute commands in existing Kubernetes pods."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from functools import cached_property
+from typing import TYPE_CHECKING
+
+from kubernetes.client.rest import ApiException
+from kubernetes.stream import stream as kubernetes_stream
+
+from airflow.providers.cncf.kubernetes.exceptions import PodExecException
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.version_compat import AIRFLOW_V_3_1_PLUS
+
+if AIRFLOW_V_3_1_PLUS:
+ from airflow.sdk import BaseOperator
+else:
+ from airflow.models import BaseOperator
+
+if TYPE_CHECKING:
+ from kubernetes.client import CoreV1Api, V1Pod
+ from kubernetes.stream.ws_client import WSClient
+
+ from airflow.sdk import Context
+
+__all__ = ["KubernetesPodExecOperator"]
+
+
+class KubernetesPodExecOperator(BaseOperator):
+ """
+ Execute a command in a running container of an existing Kubernetes pod.
+
+ The operator does not create, restart, or delete the target pod. Commands are executed directly,
+ without a shell; include a shell explicitly in ``command`` when shell features are required.
+
+ :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 the namespace configured in the
+ Kubernetes connection, then ``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. (templated)
+ :param kubernetes_conn_id: The :ref:`Kubernetes connection <howto/connection:kubernetes>` to use.
+ (templated)
+ :param in_cluster: Use in-cluster Kubernetes configuration.
+ :param cluster_context: Context to use from the kubeconfig. (templated)
+ :param config_file: Path to the kubeconfig file. (templated)
+ :param do_xcom_push: Return standard output for XCom when ``True``. Defaults to ``False``.
+ """
+
+ template_fields: Sequence[str] = (
+ "pod_name",
+ "command",
+ "namespace",
+ "container_name",
+ "kubernetes_conn_id",
+ "cluster_context",
+ "config_file",
+ )
+ template_fields_renderers = {"command": "py"}
+
+ def __init__(
+ self,
+ *,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str | None = None,
+ container_name: str | None = None,
+ kubernetes_conn_id: str | None = KubernetesHook.default_conn_name,
+ in_cluster: bool | None = None,
+ cluster_context: str | None = None,
+ config_file: str | None = None,
+ do_xcom_push: bool = False,
+ **kwargs,
+ ) -> None:
+ super().__init__(do_xcom_push=do_xcom_push, **kwargs)
+ self.pod_name = pod_name
+ self.command = command
+ self.namespace = namespace
+ self.container_name = container_name
+ self.kubernetes_conn_id = kubernetes_conn_id
+ self.in_cluster = in_cluster
+ self.cluster_context = cluster_context
+ self.config_file = config_file
+ self._exec_client: WSClient | None = None
+
+ @cached_property
+ def hook(self) -> KubernetesHook:
+ return KubernetesHook(
+ conn_id=self.kubernetes_conn_id,
+ in_cluster=self.in_cluster,
+ config_file=self.config_file,
+ cluster_context=self.cluster_context,
+ )
+
+ @cached_property
+ def client(self) -> CoreV1Api:
+ return self.hook.core_v1_client
+
+ def _resolve_namespace(self) -> str:
+ return self.namespace or self.hook.get_namespace() or KubernetesHook.DEFAULT_NAMESPACE
+
+ def _validate_command(self) -> list[str]:
+ if isinstance(self.command, str) or not isinstance(self.command, Sequence):
+ raise TypeError("`command` must be a sequence of strings, not a single string")
+ if not self.command:
+ raise ValueError("`command` must contain at least one element")
+ if not all(isinstance(argument, str) for argument in self.command):
+ raise TypeError("Every element of `command` must be a string")
+ return list(self.command)
+
+ def _resolve_container_name(self, pod: V1Pod) -> str:
+ containers = pod.spec.containers if pod.spec and pod.spec.containers else []
+ container_names = [container.name for container in containers]
+ if not container_names:
+ raise PodExecException(f"Pod {self.pod_name!r} does not define any containers")
+
+ if self.container_name:
+ if self.container_name not in container_names:
+ raise PodExecException(
+ f"Container {self.container_name!r} does not exist in pod {self.pod_name!r}"
+ )
+ return self.container_name
+
+ annotations = pod.metadata.annotations if pod.metadata and pod.metadata.annotations else {}
+ default_container = annotations.get("kubectl.kubernetes.io/default-container")
+ if isinstance(default_container, str) and default_container in container_names:
+ return default_container
+ return container_names[0]
+
+ def _validate_container_is_running(self, pod: V1Pod, container_name: str) -> None:
+ if not pod.status or pod.status.phase != "Running":
+ phase = pod.status.phase if pod.status else None
+ raise PodExecException(
+ f"Cannot execute a command in pod {self.pod_name!r} while it is in phase {phase!r}"
+ )
+
+ statuses = pod.status.container_statuses or []
+ container_status = next((status for status in statuses if status.name == container_name), None)
+ if (
+ container_status is None
+ or container_status.state is None
+ or container_status.state.running is None
+ ):
+ raise PodExecException(f"Container {container_name!r} in pod {self.pod_name!r} is not running")
+
+ def _log_output(self, output: str, *, stream_name: str) -> None:
+ log_method = self.log.warning if stream_name == "stderr" else self.log.info
+ for line in output.splitlines():
Review Comment:
Good point. WebSocket reads are not guaranteed to contain complete lines. I added independent buffers for stdout and stderr, so only complete newline-delimited records are logged, and any remaining partial line is flushed when the session ends. The tests now cover lines split across chunks and unterminated trailing output.
This follows the buffering approach already used by `DockerOperator.fetch_logs`
https://github.com/apache/airflow/blob/f760ac1c03b7277f3cb69071b8ead56d27ecbff7/providers/docker/src/airflow/providers/docker/operators/docker.py#L60-L77.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889723226
##########
providers/cncf/kubernetes/tests/system/cncf/kubernetes/example_kubernetes_pod_exec.py:
##########
@@ -0,0 +1,134 @@
+# 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.
+"""Example Dag for executing a command in an existing Kubernetes Pod."""
+
+from __future__ import annotations
+
+import os
+import time
+from datetime import datetime
+
+from kubernetes.client.rest import ApiException
+
+from airflow.providers.cncf.kubernetes.hooks.kubernetes import KubernetesHook
+from airflow.providers.cncf.kubernetes.operators.pod_exec import KubernetesPodExecOperator
+from airflow.providers.cncf.kubernetes.operators.resource import (
+ KubernetesCreateResourceOperator,
+ KubernetesDeleteResourceOperator,
+)
+from airflow.sdk import DAG, TriggerRule, task
+
+ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID", "default").lower().replace("_", "-")
+DAG_ID = "example_kubernetes_pod_exec_operator"
+NAMESPACE = "default"
+POD_NAME = f"airflow-pod-exec-{ENV_ID}"
+CONTAINER_NAME = "worker"
+EXPECTED_OUTPUT = "command executed in existing pod"
+
+pod_conf = f"""
+apiVersion: v1
+kind: Pod
+metadata:
+ name: {POD_NAME}
+ namespace: {NAMESPACE}
+spec:
+ restartPolicy: Never
+ containers:
+ - name: {CONTAINER_NAME}
+ image: busybox:1.38.0
+ command: ["sleep", "3600"]
+"""
+
+
+@task
+def wait_for_running_pod() -> None:
Review Comment:
Done, thanks! Replaced the manual polling loop with a rescheduling @task.sensor.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add KubernetesPodExecOperator for existing Kubernetes Pods [airflow]
Posted by "AlejandroMorgante (via GitHub)" <gi...@apache.org>.
AlejandroMorgante commented on code in PR #71244:
URL: https://github.com/apache/airflow/pull/71244#discussion_r3889739687
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/exceptions.py:
##########
@@ -39,3 +39,7 @@ class KubernetesApiError(AirflowException):
class KubernetesApiPermissionError(AirflowException):
"""Raised when an error is encountered while trying access Kubernetes API."""
+
+
+class PodExecException(AirflowException):
Review Comment:
Done, thanks! Replaced the dedicated exception with appropriate built-in exception types
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org