ramitkataria commented on code in PR #72542:
URL: https://github.com/apache/airflow/pull/72542#discussion_r4051713888
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -1348,3 +1349,110 @@ def _refresh_cached_properties(self) -> None:
self.log.exception("Failed to refresh AWS credentials.")
raise
super()._refresh_cached_properties()
+
+
+class EksPodExecOperator(KubernetesPodExecOperator):
+ """
+ Execute a command in a running container of an existing Pod on Amazon EKS.
+
+ The operator authenticates with Amazon EKS and delegates command execution
to
+
:class:`~airflow.providers.cncf.kubernetes.operators.pod_exec.KubernetesPodExecOperator`.
+ It does not create, restart, or delete the target Pod.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:EksPodExecOperator`
+
+ :param cluster_name: The name of the Amazon EKS Cluster containing the
Pod. (templated)
+ :param pod_name: Name of the existing Kubernetes Pod. (templated)
+ :param command: Command and arguments to execute in the container.
(templated)
+ :param namespace: Namespace containing the Pod. Defaults to ``default``.
(templated)
+ :param container_name: Name of the container in which to execute the
command. When omitted, the
+ ``kubectl.kubernetes.io/default-container`` annotation or the first
container is used.
+ Defaults to ``None``. (templated)
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
(templated)
+ Defaults to ``aws_default``. If this is ``None`` or empty, the default
boto3 credential
+ strategy is used without an Airflow connection lookup.
+ :param region_name: AWS region containing the Amazon EKS Cluster.
(templated)
+ Defaults to ``None``, which uses the region from the AWS connection
when available and
+ otherwise falls back to the default boto3 region strategy.
+ :param verify: Whether to verify SSL certificates, or the path to a CA
bundle. Defaults to
+ ``None``, which uses the value from the AWS connection when available.
(templated)
+ :param botocore_config: Configuration dictionary for the botocore client.
Defaults to ``None``,
+ which uses ``config_kwargs`` from the AWS connection when available.
+ :param do_xcom_push: Return standard output through XCom when ``True``.
Defaults to ``False``.
+ :param max_xcom_output_size: Maximum UTF-8 byte size retained for XCom.
Defaults to 49,344 bytes.
+ """
+
+ template_fields: Sequence[str] = aws_template_fields(
+ "cluster_name",
+ *(
+ field
+ for field in KubernetesPodExecOperator.template_fields
+ if field not in {"cluster_context", "config_file",
"kubernetes_conn_id"}
+ ),
+ )
+
+ def __init__(
+ self,
+ *,
+ cluster_name: str,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str = DEFAULT_NAMESPACE_NAME,
+ container_name: str | None = None,
+ aws_conn_id: str | None = DEFAULT_CONN_ID,
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ) -> None:
+ hook_params = AwsHookParams.from_constructor(
+ aws_conn_id, region_name, verify, botocore_config,
additional_params=kwargs
+ )
+ super().__init__(
+ pod_name=pod_name,
+ command=command,
+ namespace=namespace,
+ container_name=container_name,
+ kubernetes_conn_id=None,
Review Comment:
Does `kubernetes_conn_id=None` actually skip the connection lookup here?
`KubernetesHook.__init__` does `self.conn_id = conn_id or kubernetes_conn_id`,
so I think `None` falls back to `kubernetes_default`. If that connection has
`kube_config` or `cluster_context` set, `get_conn` would fail against the
generated kubeconfig, and since the parameter is hardcoded there is no way to
work around it. Would it make sense to either override `hook` to pass
`conn_id=None` explicitly, or keep `kubernetes_conn_id` user-settable like
`EksPodOperator` does?
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/eks.py:
##########
@@ -1348,3 +1349,110 @@ def _refresh_cached_properties(self) -> None:
self.log.exception("Failed to refresh AWS credentials.")
raise
super()._refresh_cached_properties()
+
+
+class EksPodExecOperator(KubernetesPodExecOperator):
+ """
+ Execute a command in a running container of an existing Pod on Amazon EKS.
+
+ The operator authenticates with Amazon EKS and delegates command execution
to
+
:class:`~airflow.providers.cncf.kubernetes.operators.pod_exec.KubernetesPodExecOperator`.
+ It does not create, restart, or delete the target Pod.
+
+ .. seealso::
+ For more information on how to use this operator, take a look at the
guide:
+ :ref:`howto/operator:EksPodExecOperator`
+
+ :param cluster_name: The name of the Amazon EKS Cluster containing the
Pod. (templated)
+ :param pod_name: Name of the existing Kubernetes Pod. (templated)
+ :param command: Command and arguments to execute in the container.
(templated)
+ :param namespace: Namespace containing the Pod. Defaults to ``default``.
(templated)
+ :param container_name: Name of the container in which to execute the
command. When omitted, the
+ ``kubectl.kubernetes.io/default-container`` annotation or the first
container is used.
+ Defaults to ``None``. (templated)
+ :param aws_conn_id: The Airflow connection used for AWS credentials.
(templated)
+ Defaults to ``aws_default``. If this is ``None`` or empty, the default
boto3 credential
+ strategy is used without an Airflow connection lookup.
+ :param region_name: AWS region containing the Amazon EKS Cluster.
(templated)
+ Defaults to ``None``, which uses the region from the AWS connection
when available and
+ otherwise falls back to the default boto3 region strategy.
+ :param verify: Whether to verify SSL certificates, or the path to a CA
bundle. Defaults to
+ ``None``, which uses the value from the AWS connection when available.
(templated)
+ :param botocore_config: Configuration dictionary for the botocore client.
Defaults to ``None``,
+ which uses ``config_kwargs`` from the AWS connection when available.
+ :param do_xcom_push: Return standard output through XCom when ``True``.
Defaults to ``False``.
+ :param max_xcom_output_size: Maximum UTF-8 byte size retained for XCom.
Defaults to 49,344 bytes.
+ """
+
+ template_fields: Sequence[str] = aws_template_fields(
+ "cluster_name",
+ *(
+ field
+ for field in KubernetesPodExecOperator.template_fields
+ if field not in {"cluster_context", "config_file",
"kubernetes_conn_id"}
+ ),
+ )
+
+ def __init__(
+ self,
+ *,
+ cluster_name: str,
+ pod_name: str,
+ command: Sequence[str],
+ namespace: str = DEFAULT_NAMESPACE_NAME,
+ container_name: str | None = None,
+ aws_conn_id: str | None = DEFAULT_CONN_ID,
+ region_name: str | None = None,
+ verify: bool | str | None = None,
+ botocore_config: dict | None = None,
+ **kwargs,
+ ) -> None:
+ hook_params = AwsHookParams.from_constructor(
+ aws_conn_id, region_name, verify, botocore_config,
additional_params=kwargs
+ )
+ super().__init__(
+ pod_name=pod_name,
+ command=command,
+ namespace=namespace,
+ container_name=container_name,
+ kubernetes_conn_id=None,
+ in_cluster=False,
+ cluster_context=None,
+ config_file=None,
+ **kwargs,
+ )
+ self.cluster_name = cluster_name
+ self.aws_conn_id = hook_params.aws_conn_id
+ self.region_name = hook_params.region_name
+ self.verify = hook_params.verify
+ self.botocore_config = hook_params.botocore_config
+
+ def execute(self, context: Context) -> str | None:
+ eks_hook = EksHook(
+ aws_conn_id=self.aws_conn_id,
+ region_name=self.region_name,
+ verify=self.verify,
+ config=self.botocore_config,
+ )
+ credentials = eks_hook.get_session().get_credentials()
+ if credentials is None:
+ raise RuntimeError(
+ "Unable to retrieve AWS credentials. Credentials may have
expired or not been configured. "
+ "Please check your AWS connection configuration."
+ )
+ frozen_credentials = credentials.get_frozen_credentials()
+ with eks_hook._secure_credential_context(
+ frozen_credentials.access_key,
+ frozen_credentials.secret_key,
+ frozen_credentials.token,
+ ) as credentials_file:
+ with eks_hook.generate_config_file(
+ eks_cluster_name=self.cluster_name,
+ pod_namespace=self.namespace,
+ credentials_file=credentials_file,
+ ) as config_file:
+ self.config_file = config_file
+ try:
+ return super().execute(context)
+ finally:
+ self.config_file = None
Review Comment:
nit: `hook` and `client` are `cached_property` on the parent, so resetting
`config_file` here does not release the hook that still points at the deleted
temp file. Only matters if the same instance runs `execute` twice, and
`EksPodOperator` has the same shape
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
