This is an automated email from the ASF dual-hosted git repository.
o-nikolas pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 3134ac232fa Add AwsEksExecutor to the Amazon provider (#74399)
3134ac232fa is described below
commit 3134ac232fa74106b746fc13fa23379bd2338408
Author: Sean Ghaeli <[email protected]>
AuthorDate: Mon Oct 19 18:02:34 2026 -0700
Add AwsEksExecutor to the Amazon provider (#74399)
Runs each task as a pod on an Amazon EKS cluster. It subclasses the
Kubernetes executor and only replaces authentication. The Kubernetes client is
built from an Airflow AWS connection with no kubeconfig, and the short-lived
EKS token is re-minted in-process shortly before it expires
---
providers/amazon/docs/executors/eks-executor.rst | 223 ++++++++++++++++++
providers/amazon/docs/executors/general.rst | 6 +-
providers/amazon/docs/executors/index.rst | 4 +
providers/amazon/provider.yaml | 41 ++++
.../providers/amazon/aws/executors/eks/__init__.py | 21 ++
.../amazon/aws/executors/eks/_client_factory.py | 170 ++++++++++++++
.../amazon/aws/executors/eks/eks_executor.py | 189 +++++++++++++++
.../providers/amazon/aws/executors/eks/utils.py | 37 +++
.../providers/amazon/aws/utils/eks_get_token.py | 17 +-
.../airflow/providers/amazon/get_provider_info.py | 34 +++
.../src/airflow/providers/amazon/version_compat.py | 2 +
.../unit/amazon/aws/executors/eks/__init__.py | 16 ++
.../aws/executors/eks/test__client_factory.py | 261 +++++++++++++++++++++
.../amazon/aws/executors/eks/test_eks_executor.py | 238 +++++++++++++++++++
.../unit/amazon/aws/executors/eks/test_utils.py | 31 +++
.../unit/amazon/aws/utils/test_eks_get_token.py | 53 ++++-
16 files changed, 1332 insertions(+), 11 deletions(-)
diff --git a/providers/amazon/docs/executors/eks-executor.rst
b/providers/amazon/docs/executors/eks-executor.rst
new file mode 100644
index 00000000000..12ca2a30ccb
--- /dev/null
+++ b/providers/amazon/docs/executors/eks-executor.rst
@@ -0,0 +1,223 @@
+ .. 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.
+
+.. |executorName| replace:: EKS
+
+.. _eks_executor:
+
+================
+AWS EKS Executor
+================
+
+The EKS executor runs each Airflow task in its own pod on an Amazon EKS
cluster.
+
+It extends the Kubernetes executor from the ``cncf.kubernetes`` provider, so
pod scheduling,
+pod templates, per-task pod overrides, log handling and requirements such as a
database
+backend other than SQLite are all as described in
+:doc:`apache-airflow-providers-cncf-kubernetes:kubernetes_executor`. What it
adds is
+authentication: it builds the Kubernetes client for your EKS cluster from an
Airflow AWS
+connection, so the scheduler needs no kubeconfig file and you do not have to
refresh cluster
+credentials yourself.
+
+For a quick start guide please see :ref:`here <eks_setup_guide>`.
+
+Requirements
+------------
+
+The executor needs ``apache-airflow-providers-cncf-kubernetes`` version
10.24.0 or
+newer, the release that added the
+:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__client_factory`
+setting the executor is built on. The executor checks this when it is imported
and raises a
+clear error if an older version is installed. Install the ``cncf.kubernetes``
extra of the
+Amazon provider together with that version:
+
+.. code-block:: bash
+
+ pip install 'apache-airflow-providers-amazon[cncf.kubernetes]' \
+ 'apache-airflow-providers-cncf-kubernetes>=10.24.0'
+
+The executor also relies on botocore using the regional STS endpoint by
default (botocore 1.40
+and newer), and the Amazon provider already requires ``botocore>=1.42.79``, so
no extra install
+step is needed.
+
+The AWS credentials the executor uses must be allowed to call
``eks:DescribeCluster``
+on the cluster, and their IAM principal must also be granted access inside the
cluster, as
+described in :ref:`eks_grant_access`. Without that in-cluster grant every
Kubernetes API call
+is rejected.
+
+How authentication works
+------------------------
+
+When the executor starts, it plugs its own Kubernetes client into the
Kubernetes
+executor. Every process that needs a client, including the separate pod
watcher process,
+builds its own from the :ref:`[aws_eks_executor] <eks_config_options>`
settings: it
+describes the cluster to find its API endpoint and certificate authority, then
gets an
+authentication token for it. An EKS token is a presigned STS URL that stays
valid for
+roughly fifteen minutes, so the executor registers a refresh hook that the
Kubernetes client
+calls on every authenticated request. The hook reuses the current token and
gets a new one
+a minute before it expires.
+
+Before it accepts any tasks, the executor checks that the cluster is
``ACTIVE`` or
+``UPDATING``, since EKS keeps the Kubernetes API available during an update
(see `Update
+existing cluster to new Kubernetes version
+<https://docs.aws.amazon.com/eks/latest/userguide/update-cluster.html>`__).
Unless
+:ref:`check_health_on_startup <eks_config_options>` is turned off, it also
checks that it
+is allowed to list pods in its namespace. It refuses to start if either check
fails.
+
+.. _eks_config_options:
+
+Config Options
+--------------
+
+The executor reads its own settings from an ``aws_eks_executor`` section in
+``airflow.cfg``, or from environment variables of the form
+``AIRFLOW__AWS_EKS_EXECUTOR__<OPTION_NAME>``, for example
+``AIRFLOW__AWS_EKS_EXECUTOR__CLUSTER_NAME=airflow-eks-cluster``. For more
information on
+how to set these options, see :doc:`apache-airflow:howto/set-config`.
+
+Required config options:
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+- CLUSTER_NAME - The name of the Amazon EKS cluster that tasks run on. The
+ executor refuses to start if this is unset.
+
+Optional config options:
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+- CONN_ID - The Airflow connection (i.e. credentials) used by the EKS
+ executor to make API calls to Amazon EKS. Defaults to ``aws_default``.
+- REGION_NAME - The AWS Region the cluster is in. When this is left empty, it
+ falls back to the :ref:`AWS connection's <howto/connection:aws>` region,
then
+ boto3's.
+- CHECK_HEALTH_ON_STARTUP - Whether to check on startup that the executor can
+ list pods in its namespace. Defaults to ``True``.
+
+Pod-level configuration
+~~~~~~~~~~~~~~~~~~~~~~~
+
+Settings for the worker pods stay in the
+:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor`
+section, exactly as for the Kubernetes executor. This includes
+:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__namespace`,
+:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__pod_template_file`,
+:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__worker_container_repository`,
+:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__worker_container_tag`,
and
+:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__delete_worker_pods`.
+See :doc:`apache-airflow-providers-cncf-kubernetes:kubernetes_executor` for
the full list, and
+for how :ref:`pod templates
<apache-airflow-providers-cncf-kubernetes:concepts:pod_template_file>` and
+:ref:`apache-airflow-providers-cncf-kubernetes:concepts:pod_override` work.
+
+Leave
:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__client_factory`
and
+:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__async_client_factory`
+unset. The executor sets both itself, and raises an error at startup if either
is set to
+something else, so that a conflicting setting cannot silently send tasks to
the wrong cluster.
+
+Set :ref:`[kubernetes_executor] pod_template_file
+<apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__pod_template_file>`
+to a YAML file describing the worker pod, since most setups need one. Without
one, the worker
+pod is built from ``worker_container_repository`` and ``worker_container_tag``
alone, which
+only works if that image already has everything the task needs. The template's
first container
+must be named ``base`` and run your Airflow worker image. The
+:ref:`apache-airflow-providers-cncf-kubernetes:concepts:pod_template_file`
section of the
+Kubernetes executor docs covers the full requirements, and its example
templates work here
+unchanged.
+
+.. _eks_logging:
+
+.. include:: general.rst
+ :start-after: .. BEGIN LOGGING
+ :end-before: .. END LOGGING
+
+- Worker pods are deleted when their task succeeds
(:ref:`[kubernetes_executor]
+ delete_worker_pods
<apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor__delete_worker_pods>`),
+ and their logs go with them, so configure remote logging to CloudWatch Logs
or S3
+ to keep task logs viewable in the Airflow UI.
+- The remote logging configuration must be set on the worker pods as well as
on
+ the scheduler and API server, and the worker pods need an IAM role, for
+ example through EKS Pod Identity, that can write to the log destination.
+
+.. _eks_setup_guide:
+
+Setting up an EKS Executor for Apache Airflow
+---------------------------------------------
+
+.. _eks_grant_access:
+
+Grant access to the cluster
+~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+Create an `EKS access entry
+<https://docs.aws.amazon.com/eks/latest/userguide/access-entries.html>`__ for
the IAM
+role or user behind the executor's AWS connection (``[aws_eks_executor]
conn_id``,
+``aws_default`` by default), and associate a policy that allows create, get,
+list, watch, patch and delete on pods, and get on pods/log, in the namespace
you plan to
+use. If you manage cluster access through the `aws-auth ConfigMap
+<https://docs.aws.amazon.com/eks/latest/userguide/auth-configmap.html>`__
instead, add
+the principal there and map it to a Kubernetes group with the same permissions.
+
+Configure Airflow
+~~~~~~~~~~~~~~~~~
+
+Select the executor and name your cluster:
+
+.. code-block:: ini
+
+ [core]
+ executor =
airflow.providers.amazon.aws.executors.eks.eks_executor.AwsEksExecutor
+
+ [aws_eks_executor]
+ cluster_name = airflow-eks-cluster
+ region_name = us-east-1
+ conn_id = aws_default
+
+ [kubernetes_executor]
+ namespace = airflow
+ pod_template_file = /opt/airflow/pod_templates/worker_template.yaml
+ worker_container_repository =
my-account.dkr.ecr.us-east-1.amazonaws.com/airflow
+ worker_container_tag = latest
+
+The worker image must contain Airflow and the Amazon provider, and it needs
access
+to your Dag files and a network path to the Airflow API server, in the same
way any
+Kubernetes executor worker does. Configure remote logging as described in the
+:ref:`logging <eks_logging>` section.
+
+Start the scheduler and trigger a small Dag. A worker pod should appear in
your namespace
+and the task should finish. If the scheduler logs an authentication or
forbidden error from
+the Kubernetes API, check the in-cluster access grant for your IAM principal
first.
+
+Multi-team deployments
+----------------------
+
+The executor supports :doc:`multi-team
<apache-airflow:core-concepts/multi-team>` mode
+the same way the Kubernetes executor does. Teams share the cluster set in the
global
+:ref:`[aws_eks_executor] <eks_config_options>` section, and each team gets its
own
+team-scoped
:ref:`apache-airflow-providers-cncf-kubernetes:config:kubernetes_executor`
settings,
+such as the namespace and pod template.
+
+The executor refuses to start next to a Kubernetes executor in the same team,
or when both are
+global, because that Kubernetes executor would also send its pods to the EKS
cluster; configure
+them for different teams instead (see `#74844
<https://github.com/apache/airflow/issues/74844>`__).
+
+Fault tolerance
+---------------
+
+Fault tolerance, including how worker pod crashes and scheduler restarts are
handled, is
+the same as for the Kubernetes executor; see
+:doc:`apache-airflow-providers-cncf-kubernetes:kubernetes_executor`. Because
tokens are
+replaced before they expire, repeated authentication failures on a
long-running scheduler
+usually mean the credentials behind your connection are no longer valid, or
the role has
+lost its access entry on the cluster.
diff --git a/providers/amazon/docs/executors/general.rst
b/providers/amazon/docs/executors/general.rst
index 19ec5febdcf..7303218d270 100644
--- a/providers/amazon/docs/executors/general.rst
+++ b/providers/amazon/docs/executors/general.rst
@@ -244,11 +244,11 @@ Configuring Remote Logging
There are many ways to configure remote logging and several supported
destinations. A general overview of Airflow Task logging can be found
-`here
<https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/logging-monitoring/logging-tasks.html>`__.
+:doc:`here
<apache-airflow:administration-and-deployment/logging-monitoring/logging-tasks>`.
Instructions for configuring S3 remote logging can be found
-`here
<https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/logging/s3-task-handler.html>`__
+:doc:`here </logging/s3-task-handler>`
and Cloudwatch remote logging
-`here
<https://airflow.apache.org/docs/apache-airflow-providers-amazon/stable/logging/cloud-watch-task-handlers.html>`__.
+:doc:`here </logging/cloud-watch-task-handlers>`.
Some important things to point out for remote logging in the context of
the |executorName| executor:
.. END LOGGING
diff --git a/providers/amazon/docs/executors/index.rst
b/providers/amazon/docs/executors/index.rst
index 466b830d70e..6e79d28864e 100644
--- a/providers/amazon/docs/executors/index.rst
+++ b/providers/amazon/docs/executors/index.rst
@@ -20,10 +20,14 @@
Amazon Executors
================
+You can run more than one Amazon executor side by side, see
+:ref:`apache-airflow:using-multiple-executors-concurrently`.
+
.. toctree::
:maxdepth: 1
ECS Executor <ecs-executor>
Batch Executor <batch-executor>
+ EKS Executor <eks-executor>
Lambda Executor (experimental) <lambda-executor>
diff --git a/providers/amazon/provider.yaml b/providers/amazon/provider.yaml
index a2536a08ee1..598f057a2bd 100644
--- a/providers/amazon/provider.yaml
+++ b/providers/amazon/provider.yaml
@@ -1524,6 +1524,46 @@ config:
type: boolean
example: "True"
default: "True"
+ aws_eks_executor:
+ description: |
+ This section only applies if you are using the AwsEksExecutor in
+ Airflow's ``[core]`` configuration. It requires the ``cncf.kubernetes``
+ extra of this provider. The executor supplies the Kubernetes client for
+ the configured Amazon EKS cluster; all pod-level settings are still read
+ from the ``[kubernetes_executor]`` section.
+ For boto3 credential management, see
+
https://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html
+ options:
+ conn_id:
+ description: |
+ The Airflow connection (i.e. credentials) used by the EKS executor to
+ make API calls to Amazon EKS.
+ version_added: 9.39.0
+ type: string
+ example: "aws_default"
+ default: "aws_default"
+ region_name:
+ description: |
+ The name of the AWS Region where the EKS cluster is located. If not
+ specified, falls back to the AWS connection's region, then boto3's.
+ version_added: 9.39.0
+ type: string
+ example: "us-east-1"
+ default: ~
+ check_health_on_startup:
+ description: |
+ Whether or not to check the EKS Executor health on startup.
+ version_added: 9.39.0
+ type: boolean
+ example: "True"
+ default: "True"
+ cluster_name:
+ description: |
+ The name of the Amazon EKS cluster to run Airflow tasks on. Required.
+ version_added: 9.39.0
+ type: string
+ example: "airflow-eks-cluster"
+ default: ~
aws_auth_manager:
description: |
This section only applies if you are using the AwsAuthManager. In other
words, if you set
@@ -1581,6 +1621,7 @@ executors:
-
airflow.providers.amazon.aws.executors.aws_lambda.lambda_executor.AwsLambdaExecutor
-
airflow.providers.amazon.aws.executors.batch.batch_executor.AwsBatchExecutor
- airflow.providers.amazon.aws.executors.ecs.ecs_executor.AwsEcsExecutor
+ - airflow.providers.amazon.aws.executors.eks.eks_executor.AwsEksExecutor
auth-managers:
- airflow.providers.amazon.aws.auth_manager.aws_auth_manager.AwsAuthManager
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/__init__.py
b/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/__init__.py
new file mode 100644
index 00000000000..831bb9e8ab7
--- /dev/null
+++
b/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/__init__.py
@@ -0,0 +1,21 @@
+# 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
+
+__all__ = ["AwsEksExecutor"]
+
+from airflow.providers.amazon.aws.executors.eks.eks_executor import
AwsEksExecutor
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/_client_factory.py
b/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/_client_factory.py
new file mode 100644
index 00000000000..4b0dfc2f001
--- /dev/null
+++
b/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/_client_factory.py
@@ -0,0 +1,170 @@
+# 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.
+"""Kubernetes client factories for the AwsEksExecutor."""
+
+# Internal to AwsEksExecutor, which writes these functions' import paths into
+# [kubernetes_executor] client_factory and async_client_factory.
+
+from __future__ import annotations
+
+import atexit
+import os
+import tempfile
+import threading
+import time
+from base64 import b64decode
+from contextlib import suppress
+from typing import TYPE_CHECKING
+
+import kubernetes
+from packaging.version import Version
+
+from airflow.providers.amazon.aws.executors.eks.utils import (
+ CONFIG_DEFAULTS,
+ CONFIG_GROUP_NAME,
+ AllEksConfigKeys,
+)
+from airflow.providers.amazon.aws.hooks.eks import EksHook
+from airflow.providers.amazon.aws.hooks.sts import StsHook
+from airflow.providers.amazon.aws.utils.eks_get_token import (
+ TOKEN_EXPIRATION_MINUTES,
+ fetch_access_token_for_cluster,
+)
+from airflow.providers.common.compat.sdk import conf
+
+if TYPE_CHECKING:
+ from kubernetes import client
+ from kubernetes_asyncio import client as async_client
+
+# UPDATING still serves the Kubernetes API, so every other status is refused.
+_USABLE_CLUSTER_STATUSES = ("ACTIVE", "UPDATING")
+
+
+def _get_sync_token_key(kubernetes_version: str) -> str:
+ # kubernetes 36 moved the bearer token from the "authorization" api_key to
"BearerToken".
+ return "BearerToken" if Version(kubernetes_version).major >= 36 else
"authorization"
+
+
+_SYNC_TOKEN_KEY = _get_sync_token_key(kubernetes.__version__)
+
+
+def _get_eks_kube_client() -> client.CoreV1Api:
+ """Build a Kubernetes client for the configured EKS cluster, in memory and
without a kubeconfig."""
+ from kubernetes import client
+
+ configuration = client.Configuration()
+ _configure_eks_auth(configuration, _SYNC_TOKEN_KEY)
+ return client.CoreV1Api(client.ApiClient(configuration=configuration))
+
+
+def _get_eks_async_kube_client() -> async_client.CoreV1Api:
+ """Build the asynchronous Kubernetes client used when
``async_pod_creation`` is enabled."""
+ from kubernetes_asyncio import client as async_client
+
+ configuration = async_client.Configuration()
+ # kubernetes_asyncio has always used "BearerToken".
+ _configure_eks_auth(configuration, "BearerToken")
+ return
async_client.CoreV1Api(async_client.ApiClient(configuration=configuration))
+
+
+def _configure_eks_auth(
+ configuration: client.Configuration | async_client.Configuration,
token_key: str
+) -> None:
+ cluster_name = conf.get(CONFIG_GROUP_NAME, AllEksConfigKeys.CLUSTER_NAME,
fallback=None)
+ if not cluster_name:
+ raise ValueError(
+ f"[{CONFIG_GROUP_NAME}] cluster_name is required to build an EKS
client. "
+ "Set it in airflow.cfg or with
AIRFLOW__AWS_EKS_EXECUTOR__CLUSTER_NAME."
+ )
+ region_name = conf.get(CONFIG_GROUP_NAME, AllEksConfigKeys.REGION_NAME,
fallback=None)
+ conn_id = conf.get(
+ CONFIG_GROUP_NAME,
+ AllEksConfigKeys.AWS_CONN_ID,
+ fallback=CONFIG_DEFAULTS[AllEksConfigKeys.AWS_CONN_ID],
+ )
+
+ eks_hook = EksHook(aws_conn_id=conn_id, region_name=region_name)
+ cluster = eks_hook.conn.describe_cluster(name=cluster_name)["cluster"]
+ if cluster["status"] not in _USABLE_CLUSTER_STATUSES:
+ raise ValueError(
+ f"EKS cluster {cluster_name} is {cluster['status']}; the executor
needs it to be ACTIVE or "
+ f"UPDATING. Wait for it to become ACTIVE, or point
[{CONFIG_GROUP_NAME}] cluster_name at a "
+ "usable cluster."
+ )
+ session = eks_hook.get_session()
+
+ # EKS needs the regional STS endpoint, which botocore uses by default
since 1.40.
+ sts_endpoint = StsHook(aws_conn_id=conn_id,
region_name=session.region_name).conn_client_meta.endpoint_url
+ sts_url = f"{sts_endpoint}/?Action=GetCallerIdentity&Version=2011-06-15"
+
+ configuration.host = cluster["endpoint"]
+ configuration.ssl_ca_cert =
_write_cluster_ca_file(cluster["certificateAuthority"]["data"])
+ configuration.api_key_prefix[token_key] = "Bearer"
+
+ # EKS tokens expire after about 15 minutes, so get a new one shortly
before the current one expires.
+ # The first call always fetches a token because token is None. The call at
the end of this function
+ # makes that happen now, so the first request already has a token.
+ token: str | None = None
+ refresh_at = 0.0
+
+ def refresh_api_key(config: client.Configuration |
async_client.Configuration) -> None:
+ nonlocal token, refresh_at
+ if token is None or time.monotonic() >= refresh_at:
+ token = fetch_access_token_for_cluster(
+ cluster_name, sts_url, region_name=session.region_name,
session=session
+ )
+ refresh_at = time.monotonic() + TOKEN_EXPIRATION_MINUTES * 60
+ config.api_key[token_key] = token
+
+ configuration.refresh_api_key_hook = refresh_api_key
+ refresh_api_key(configuration)
+
+
+# The CA file written for each cluster CA, and the process that wrote it.
+_ca_files: dict[str, tuple[str, int]] = {}
+_ca_files_lock = threading.Lock()
+
+
+def _write_cluster_ca_file(base64_ca_data: str) -> str:
+ # The kubernetes client reads the CA file lazily, so it must outlive the
client. A client is
+ # built for every log read, so write each CA once per process rather than
once per client.
+ with _ca_files_lock:
+ if base64_ca_data not in _ca_files:
+ with tempfile.NamedTemporaryFile(
+ mode="wb", prefix="airflow-eks-ca-", suffix=".pem",
delete=False
+ ) as ca_file:
+ ca_file.write(b64decode(base64_ca_data))
+ _ca_files[base64_ca_data] = (ca_file.name, os.getpid())
+ return _ca_files[base64_ca_data][0]
+
+
+def _remove_ca_files() -> None:
+ # A forked child reuses its parent's files, so each process only removes
the ones it wrote.
+ for path, pid in _ca_files.values():
+ if pid == os.getpid():
+ with suppress(OSError):
+ os.remove(path)
+
+
+def _reset_ca_files_lock() -> None:
+ # Another thread of the parent may have held the lock when it forked.
+ global _ca_files_lock
+ _ca_files_lock = threading.Lock()
+
+
+atexit.register(_remove_ca_files)
+os.register_at_fork(after_in_child=_reset_ca_files_lock)
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/eks_executor.py
b/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/eks_executor.py
new file mode 100644
index 00000000000..083b0bb9d8d
--- /dev/null
+++
b/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/eks_executor.py
@@ -0,0 +1,189 @@
+# 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.
+"""AwsEksExecutor: run Airflow tasks as pods on an Amazon EKS cluster."""
+
+from __future__ import annotations
+
+import os
+from typing import TYPE_CHECKING
+
+from airflow.executors.executor_loader import ExecutorLoader
+from airflow.providers.amazon.aws.executors.eks.utils import (
+ CONFIG_DEFAULTS,
+ CONFIG_GROUP_NAME,
+ AllEksConfigKeys,
+)
+from airflow.providers.amazon.version_compat import AIRFLOW_V_3_2_PLUS
+from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException, conf
+
+_CNCF_KUBERNETES_REQUIRED = "AwsEksExecutor requires
apache-airflow-providers-cncf-kubernetes>=10.24.0"
+
+try:
+ from kubernetes.client.rest import ApiException
+
+ from airflow.providers.cncf.kubernetes.executors.kubernetes_executor
import KubernetesExecutor
+ from airflow.providers.cncf.kubernetes.get_provider_info import (
+ get_provider_info as get_cncf_kubernetes_provider_info,
+ )
+except ImportError as e:
+ raise AirflowOptionalProviderFeatureException(_CNCF_KUBERNETES_REQUIRED)
from e
+
+# Check for the option, not the version, since unreleased trees report the
previous version.
+if "client_factory" not in (
+ get_cncf_kubernetes_provider_info().get("config",
{}).get("kubernetes_executor", {}).get("options", {})
+):
+ raise AirflowOptionalProviderFeatureException(_CNCF_KUBERNETES_REQUIRED)
+
+_FACTORY_MODULE = "airflow.providers.amazon.aws.executors.eks._client_factory"
+_CLIENT_FACTORY_PATH = f"{_FACTORY_MODULE}._get_eks_kube_client"
+_ASYNC_CLIENT_FACTORY_PATH = f"{_FACTORY_MODULE}._get_eks_async_kube_client"
+# Airflow 2 hybrid executors that wrap a KubernetesExecutor.
+_HYBRID_KUBERNETES_EXECUTORS = ("CeleryKubernetesExecutor",
"LocalKubernetesExecutor")
+
+
+class AwsEksExecutor(KubernetesExecutor):
+ """
+ A KubernetesExecutor that authenticates against an Amazon EKS cluster.
+
+ Builds the Kubernetes client from ``[aws_eks_executor]`` configuration and
keeps the
+ short-lived EKS token fresh in-process. All pod-level behaviour comes
unchanged from the
+ KubernetesExecutor and its ``[kubernetes_executor]`` configuration.
+ """
+
+ supports_multi_team: bool = True
+
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ # These need team_name, which super().__init__ sets.
+ self._check_no_team_eks_config()
+ self._validate_eks_config()
+ self._check_no_kubernetes_executor_in_scope()
+ self._ensure_client_factory()
+
+ def start(self) -> None:
+ """Call this when the Executor is run for the first time by the
scheduler."""
+ super().start()
+ check_health = conf.getboolean(
+ CONFIG_GROUP_NAME,
+ AllEksConfigKeys.CHECK_HEALTH_ON_STARTUP,
+ fallback=CONFIG_DEFAULTS[AllEksConfigKeys.CHECK_HEALTH_ON_STARTUP],
+ )
+
+ if not check_health:
+ return
+
+ self.log.info("Starting EKS Executor and determining health...")
+ try:
+ self.check_health()
+ except RuntimeError:
+ self.log.error("Stopping the Airflow Scheduler from starting until
the issue is resolved.")
+ raise
+
+ def check_health(self) -> None:
+ """
+ Make a test Kubernetes API call to check the health of the EKS
Executor.
+
+ Building the client only proves the cluster exists; without this, a
missing EKS access
+ entry or RBAC binding only shows up later as a watcher error loop.
+ """
+ if TYPE_CHECKING:
+ assert self.kube_client
+ namespace = self.kube_config.kube_namespace
+ try:
+ self.kube_client.list_namespaced_pod(namespace, limit=1)
+ except ApiException as e:
+ raise RuntimeError(
+ f"EKS Executor health check has failed because: cannot list
pods in namespace {namespace} "
+ f"({e.status} {e.reason}). The IAM principal behind the AWS
connection needs an EKS access "
+ "entry (or aws-auth mapping) on the cluster that allows
create, get, list, watch, patch and "
+ f"delete on pods and get on pods/log in namespace {namespace}.
See 'Grant access to the "
+ "cluster' in the AwsEksExecutor docs."
+ ) from e
+ self.log.info("EKS Executor health check has succeeded.")
+
+ @staticmethod
+ def _validate_eks_config() -> None:
+ if not conf.get(CONFIG_GROUP_NAME, AllEksConfigKeys.CLUSTER_NAME,
fallback=None):
+ raise ValueError(
+ f"AwsEksExecutor requires [{CONFIG_GROUP_NAME}] cluster_name
to be set. "
+ "Set it in airflow.cfg or with
AIRFLOW__AWS_EKS_EXECUTOR__CLUSTER_NAME."
+ )
+
+ def _check_no_team_eks_config(self) -> None:
+ # [aws_eks_executor] is only read globally for now, so refuse team
values rather than ignore them.
+ if not self.team_name:
+ return
+ # self.conf looks options up for this team only, falling back to the
built-in defaults.
+ team_keys = sorted(
+ key
+ for key in AllEksConfigKeys()
+ if self.conf.get(CONFIG_GROUP_NAME, key, fallback=None) not in
(None, CONFIG_DEFAULTS.get(key))
+ )
+ if team_keys:
+ raise ValueError(
+ f"AwsEksExecutor doesn't support team-specific
[{CONFIG_GROUP_NAME}] config yet, but team "
+ f"{self.team_name} sets {', '.join(team_keys)}. Set these in
the global "
+ f"[{CONFIG_GROUP_NAME}] section instead. Per-team clusters are
tracked at "
+ "https://github.com/apache/airflow/issues/74844";
+ )
+
+ def _check_no_kubernetes_executor_in_scope(self) -> None:
+ executor_names = (
+ ExecutorLoader.get_executor_names(validate_teams=False)
+ if AIRFLOW_V_3_2_PLUS
+ else ExecutorLoader.get_executor_names()
+ )
+ for executor_name in executor_names:
+ # ExecutorName has no team_name before Airflow 3.1.
+ if getattr(executor_name, "team_name", None) != self.team_name:
+ continue
+ executor_cls, _ = ExecutorLoader.import_executor_cls(executor_name)
+ if issubclass(executor_cls, AwsEksExecutor) or not (
+ issubclass(executor_cls, KubernetesExecutor)
+ or executor_cls.__name__ in _HYBRID_KUBERNETES_EXECUTORS
+ ):
+ continue
+ scope = f"team {self.team_name}" if self.team_name else "the
global scope"
+ raise ValueError(
+ f"AwsEksExecutor can't run alongside {executor_cls.__name__}
in {scope}. AwsEksExecutor "
+ f"sets the [kubernetes_executor] client factory for {scope},
so {executor_cls.__name__} "
+ "would silently send its pods to the EKS cluster. Configure
the two executors for "
+ "different teams, or don't use them together. "
+ "See https://github.com/apache/airflow/issues/74844";
+ )
+
+ def _ensure_client_factory(self) -> None:
+ # A team's factory is looked up in team config only, with no global
fallback.
+ team_name = self.team_name
+ team_kwargs = {"team_name": team_name} if team_name else {}
+ env_var_prefix = f"AIRFLOW__{team_name.upper()}___" if team_name else
"AIRFLOW__"
+ for key, path in (
+ ("client_factory", _CLIENT_FACTORY_PATH),
+ ("async_client_factory", _ASYNC_CLIENT_FACTORY_PATH),
+ ):
+ configured = conf.get("kubernetes_executor", key, fallback=None,
**team_kwargs)
+ if configured and configured != path:
+ raise ValueError(
+ f"AwsEksExecutor sets [kubernetes_executor] {key} itself,
so leave it unset; "
+ f"it is set to {configured}"
+ )
+ if not configured:
+ # An env var, not conf.set, so the spawned pod watcher process
sees it too. A plain
+ # KubernetesExecutor in the same scope reads it as well, which
+ # _check_no_kubernetes_executor_in_scope refuses; replacing
this env var is tracked at
+ # https://github.com/apache/airflow/issues/74844
+
os.environ[f"{env_var_prefix}KUBERNETES_EXECUTOR__{key.upper()}"] = path
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/utils.py
b/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/utils.py
new file mode 100644
index 00000000000..52e0064f9d8
--- /dev/null
+++ b/providers/amazon/src/airflow/providers/amazon/aws/executors/eks/utils.py
@@ -0,0 +1,37 @@
+# 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.
+"""Configuration keys and defaults used by the EKS executor."""
+
+from __future__ import annotations
+
+from airflow.providers.amazon.aws.executors.utils.base_config_keys import
BaseConfigKeys
+
+CONFIG_GROUP_NAME = "aws_eks_executor"
+
+CONFIG_DEFAULTS = {
+ "conn_id": "aws_default",
+ "check_health_on_startup": "True",
+}
+
+
+class AllEksConfigKeys(BaseConfigKeys):
+ """All keys loaded into the config which are related to the EKS
Executor."""
+
+ AWS_CONN_ID = "conn_id"
+ CHECK_HEALTH_ON_STARTUP = "check_health_on_startup"
+ CLUSTER_NAME = "cluster_name"
+ REGION_NAME = "region_name"
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/utils/eks_get_token.py
b/providers/amazon/src/airflow/providers/amazon/aws/utils/eks_get_token.py
index 0dfe77f2561..7bf09466065 100644
--- a/providers/amazon/src/airflow/providers/amazon/aws/utils/eks_get_token.py
+++ b/providers/amazon/src/airflow/providers/amazon/aws/utils/eks_get_token.py
@@ -49,14 +49,14 @@ def get_parser():
return parser
-def fetch_access_token_for_cluster(eks_cluster_name: str, sts_url: str,
region_name: str) -> str:
- # This will use the credentials from the caller set as the standard AWS
env variables
- session = boto3.Session(region_name=region_name)
+# EKS only accepts the token if sts_url is the regional STS endpoint. The URL
is signed as given.
+def fetch_access_token_for_cluster(
+ eks_cluster_name: str, sts_url: str, region_name: str, session:
boto3.Session | None = None
+) -> str:
+ if session is None:
+ # This will use the credentials from the caller set as the standard
AWS env variables
+ session = boto3.Session(region_name=region_name)
eks_client = session.client("eks")
- # This env variable is required so that we get a regionalized endpoint for
STS in regions that
- # otherwise default to global endpoints. The mechanism below to generate
the token is very picky that
- # the endpoint is regional.
- os.environ["AWS_STS_REGIONAL_ENDPOINTS"] = "regional"
credentials = session.get_credentials()
if credentials is None:
@@ -97,6 +97,9 @@ def fetch_access_token_for_cluster(eks_cluster_name: str,
sts_url: str, region_n
def main():
+ # Keeps any STS call made here for credentials regional too. This is the
kubeconfig exec plugin's
+ # own short-lived process, so setting the env var can't affect another
thread.
+ os.environ["AWS_STS_REGIONAL_ENDPOINTS"] = "regional"
parser = get_parser()
args = parser.parse_args()
access_token = fetch_access_token_for_cluster(
diff --git a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
index 0e816d7c0da..76bfd3ec6e2 100644
--- a/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
+++ b/providers/amazon/src/airflow/providers/amazon/get_provider_info.py
@@ -1570,6 +1570,39 @@ def get_provider_info():
},
},
},
+ "aws_eks_executor": {
+ "description": "This section only applies if you are using the
AwsEksExecutor in\nAirflow's ``[core]`` configuration. It requires the
``cncf.kubernetes``\nextra of this provider. The executor supplies the
Kubernetes client for\nthe configured Amazon EKS cluster; all pod-level
settings are still read\nfrom the ``[kubernetes_executor]`` section.\nFor boto3
credential management,
see\nhttps://boto3.amazonaws.com/v1/documentation/api/latest/guide/configuration.html\n";,
+ "options": {
+ "conn_id": {
+ "description": "The Airflow connection (i.e.
credentials) used by the EKS executor to\nmake API calls to Amazon EKS.\n",
+ "version_added": "9.39.0",
+ "type": "string",
+ "example": "aws_default",
+ "default": "aws_default",
+ },
+ "region_name": {
+ "description": "The name of the AWS Region where the
EKS cluster is located. If not\nspecified, falls back to the AWS connection's
region, then boto3's.\n",
+ "version_added": "9.39.0",
+ "type": "string",
+ "example": "us-east-1",
+ "default": None,
+ },
+ "check_health_on_startup": {
+ "description": "Whether or not to check the EKS
Executor health on startup.\n",
+ "version_added": "9.39.0",
+ "type": "boolean",
+ "example": "True",
+ "default": "True",
+ },
+ "cluster_name": {
+ "description": "The name of the Amazon EKS cluster to
run Airflow tasks on. Required.\n",
+ "version_added": "9.39.0",
+ "type": "string",
+ "example": "airflow-eks-cluster",
+ "default": None,
+ },
+ },
+ },
"aws_auth_manager": {
"description": "This section only applies if you are using the
AwsAuthManager. In other words, if you set\n``[core] auth_manager =
airflow.providers.amazon.aws.auth_manager.aws_auth_manager.AwsAuthManager``
in\nAirflow's configuration.\n",
"options": {
@@ -1615,6 +1648,7 @@ def get_provider_info():
"airflow.providers.amazon.aws.executors.aws_lambda.lambda_executor.AwsLambdaExecutor",
"airflow.providers.amazon.aws.executors.batch.batch_executor.AwsBatchExecutor",
"airflow.providers.amazon.aws.executors.ecs.ecs_executor.AwsEcsExecutor",
+
"airflow.providers.amazon.aws.executors.eks.eks_executor.AwsEksExecutor",
],
"auth-managers":
["airflow.providers.amazon.aws.auth_manager.aws_auth_manager.AwsAuthManager"],
"cli":
["airflow.providers.amazon.aws.cli.definition.get_aws_cli_commands"],
diff --git a/providers/amazon/src/airflow/providers/amazon/version_compat.py
b/providers/amazon/src/airflow/providers/amazon/version_compat.py
index fa6faf2ba41..92e9bc41c2c 100644
--- a/providers/amazon/src/airflow/providers/amazon/version_compat.py
+++ b/providers/amazon/src/airflow/providers/amazon/version_compat.py
@@ -40,6 +40,7 @@ AIRFLOW_V_3_0_PLUS = get_base_airflow_version_tuple() >= (3,
0, 0)
AIRFLOW_V_3_1_PLUS: bool = get_base_airflow_version_tuple() >= (3, 1, 0)
AIRFLOW_V_3_1_1_PLUS: bool = get_base_airflow_version_tuple() >= (3, 1, 1)
AIRFLOW_V_3_1_8_PLUS: bool = get_base_airflow_version_tuple() >= (3, 1, 8)
+AIRFLOW_V_3_2_PLUS: bool = get_base_airflow_version_tuple() >= (3, 2, 0)
AIRFLOW_V_3_3_PLUS: bool = get_base_airflow_version_tuple() >= (3, 3, 0)
AIRFLOW_V_3_4_PLUS: bool = get_base_airflow_version_tuple() >= (3, 4, 0)
@@ -60,6 +61,7 @@ __all__ = [
"AIRFLOW_V_3_1_PLUS",
"AIRFLOW_V_3_1_1_PLUS",
"AIRFLOW_V_3_1_8_PLUS",
+ "AIRFLOW_V_3_2_PLUS",
"AIRFLOW_V_3_3_PLUS",
"AIRFLOW_V_3_4_PLUS",
"NOTSET",
diff --git a/providers/amazon/tests/unit/amazon/aws/executors/eks/__init__.py
b/providers/amazon/tests/unit/amazon/aws/executors/eks/__init__.py
new file mode 100644
index 00000000000..13a83393a91
--- /dev/null
+++ b/providers/amazon/tests/unit/amazon/aws/executors/eks/__init__.py
@@ -0,0 +1,16 @@
+# 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.
diff --git
a/providers/amazon/tests/unit/amazon/aws/executors/eks/test__client_factory.py
b/providers/amazon/tests/unit/amazon/aws/executors/eks/test__client_factory.py
new file mode 100644
index 00000000000..29ffa632d0c
--- /dev/null
+++
b/providers/amazon/tests/unit/amazon/aws/executors/eks/test__client_factory.py
@@ -0,0 +1,261 @@
+# 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
+
+import inspect
+import os
+import signal
+import tempfile
+import threading
+from base64 import b64encode
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from unittest import mock
+
+import boto3
+import pytest
+from kubernetes import client
+from kubernetes_asyncio import client as async_client
+
+from airflow.providers.amazon.aws.executors.eks import _client_factory
+from airflow.providers.amazon.aws.executors.eks._client_factory import (
+ _get_eks_async_kube_client,
+ _get_eks_kube_client,
+)
+from airflow.providers.amazon.aws.utils.eks_get_token import (
+ TOKEN_EXPIRATION_MINUTES,
+ fetch_access_token_for_cluster,
+)
+
+from tests_common.test_utils.config import conf_vars
+
+CLUSTER_NAME = "test-eks-cluster"
+CLUSTER_ENDPOINT = "https://ABCDEF0123456789.gr7.us-east-1.eks.amazonaws.com";
+CA_PEM = b"-----BEGIN CERTIFICATE-----\nfake-ca\n-----END CERTIFICATE-----\n"
+TOKEN_REFRESH_SECONDS = TOKEN_EXPIRATION_MINUTES * 60
+
+
[email protected](autouse=True)
+def ca_dir(tmp_path, monkeypatch):
+ """Write CA files under tmp_path and start every test with no CA files
written yet."""
+ monkeypatch.setattr(tempfile, "tempdir", str(tmp_path))
+ monkeypatch.setattr(_client_factory, "_ca_files", {})
+ return tmp_path
+
+
[email protected]
+def mock_aws():
+ with (
+ conf_vars({("aws_eks_executor", "cluster_name"): CLUSTER_NAME}),
+ mock.patch.object(_client_factory, "EksHook") as eks_hook_cls,
+ mock.patch.object(_client_factory, "StsHook") as sts_hook_cls,
+ mock.patch.object(_client_factory, "fetch_access_token_for_cluster",
autospec=True) as fetch_token,
+ ):
+ eks_hook = eks_hook_cls.return_value
+ eks_hook.conn.describe_cluster.return_value = {
+ "cluster": {
+ "status": "ACTIVE",
+ "endpoint": CLUSTER_ENDPOINT,
+ "certificateAuthority": {"data": b64encode(CA_PEM).decode()},
+ }
+ }
+ eks_hook.get_session.return_value.region_name = "us-east-1"
+ sts_hook_cls.return_value.conn_client_meta.endpoint_url =
"https://sts.us-east-1.amazonaws.com";
+ fetch_token.return_value = "k8s-aws-v1.token-1"
+ yield {
+ "eks_hook_cls": eks_hook_cls,
+ "eks_hook": eks_hook,
+ "sts_hook_cls": sts_hook_cls,
+ "fetch_token": fetch_token,
+ }
+
+
+class TestGetEksKubeClient:
+ @pytest.mark.parametrize("status", ["ACTIVE", "UPDATING"])
+ @conf_vars({("aws_eks_executor", "region_name"): "eu-west-1",
("aws_eks_executor", "conn_id"): "aws_eks"})
+ def test_builds_client_from_described_cluster(self, mock_aws, status):
+
mock_aws["eks_hook"].conn.describe_cluster.return_value["cluster"]["status"] =
status
+
+ core_v1 = _get_eks_kube_client()
+
+ assert isinstance(core_v1, client.CoreV1Api)
+ configuration = core_v1.api_client.configuration
+ assert configuration.host == CLUSTER_ENDPOINT
+ assert Path(configuration.ssl_ca_cert).read_bytes() == CA_PEM
+ assert configuration.auth_settings()["BearerToken"]["value"] ==
"Bearer k8s-aws-v1.token-1"
+
mock_aws["eks_hook_cls"].assert_called_once_with(aws_conn_id="aws_eks",
region_name="eu-west-1")
+
mock_aws["eks_hook"].conn.describe_cluster.assert_called_once_with(name=CLUSTER_NAME)
+
+ @pytest.mark.parametrize(
+ ("kubernetes_version", "token_key"),
+ [("35.0.0", "authorization"), ("36.0.1", "BearerToken")],
+ )
+ def test_token_key_follows_kubernetes_version(self, kubernetes_version,
token_key):
+ assert _client_factory._get_sync_token_key(kubernetes_version) ==
token_key
+
+ @pytest.mark.parametrize(
+ ("elapsed", "expected_token", "fetch_count"),
+ [
+ pytest.param(TOKEN_REFRESH_SECONDS - 1, "k8s-aws-v1.token-1", 1,
id="within-window"),
+ pytest.param(TOKEN_REFRESH_SECONDS, "k8s-aws-v1.token-2", 2,
id="after-deadline"),
+ ],
+ )
+ @mock.patch.object(_client_factory, "time", autospec=True)
+ def test_refresh_api_key_hook_re_mints_token_before_it_expires(
+ self, mock_time, mock_aws, elapsed, expected_token, fetch_count
+ ):
+ mock_time.monotonic.return_value = 1000.0
+ configuration = _get_eks_kube_client().api_client.configuration
+
+ mock_aws["fetch_token"].return_value = "k8s-aws-v1.token-2"
+ mock_time.monotonic.return_value = 1000.0 + elapsed
+ assert configuration.auth_settings()["BearerToken"]["value"] ==
f"Bearer {expected_token}"
+
+ assert mock_aws["fetch_token"].call_count == fetch_count
+ args = mock_aws["fetch_token"].call_args
+ assert args.args[0] == CLUSTER_NAME
+ assert
args.args[1].startswith("https://sts.us-east-1.amazonaws.com/?Action=GetCallerIdentity";)
+ assert args.kwargs["session"] is
mock_aws["eks_hook"].get_session.return_value
+
+ def
test_concurrent_builds_never_change_sts_regional_endpoints_env_var(self,
mock_aws, monkeypatch):
+ monkeypatch.setenv("AWS_STS_REGIONAL_ENDPOINTS", "legacy")
+ mock_aws["eks_hook"].get_session.side_effect = lambda: boto3.Session(
+ aws_access_key_id="testing", aws_secret_access_key="testing",
region_name="us-east-1"
+ )
+ mock_aws["fetch_token"].side_effect = fetch_access_token_for_cluster
+ env_writes = []
+ environ_cls = type(os.environ)
+ for method in ("__setitem__", "__delitem__"):
+ real = getattr(environ_cls, method)
+ monkeypatch.setattr(
+ environ_cls,
+ method,
+ lambda self, key, *args, _real=real: env_writes.append(key) or
_real(self, key, *args),
+ )
+ seen = set()
+ building = threading.Event()
+
+ def watch():
+ while building.is_set():
+ seen.add(os.environ.get("AWS_STS_REGIONAL_ENDPOINTS"))
+
+ building.set()
+ watcher = threading.Thread(target=watch)
+ watcher.start()
+ try:
+ with ThreadPoolExecutor(max_workers=8) as pool:
+ list(pool.map(lambda _: _get_eks_kube_client(), range(16)))
+ finally:
+ building.clear()
+ watcher.join()
+
+ assert "AWS_STS_REGIONAL_ENDPOINTS" not in env_writes
+ assert seen == {"legacy"}
+
+ def test_builds_share_one_ca_file(self, mock_aws, ca_dir):
+ paths = {_get_eks_kube_client().api_client.configuration.ssl_ca_cert
for _ in range(5)}
+
+ assert len(paths) == 1
+ assert [str(path) for path in ca_dir.glob("airflow-eks-ca-*")] ==
list(paths)
+ assert Path(paths.pop()).read_bytes() == CA_PEM
+
+ @pytest.mark.parametrize("status", ["CREATING", "DELETING", "FAILED"])
+ def test_unusable_cluster_status_raises(self, mock_aws, status):
+
mock_aws["eks_hook"].conn.describe_cluster.return_value["cluster"]["status"] =
status
+
+ with pytest.raises(
+ ValueError,
+ match=f"{CLUSTER_NAME} is {status}; .* ACTIVE or UPDATING. Wait
for it to become ACTIVE",
+ ):
+ _get_eks_kube_client()
+
+ @conf_vars({("aws_eks_executor", "cluster_name"): None})
+ def test_missing_cluster_name_raises(self):
+ with pytest.raises(
+ ValueError,
+ match=r"\[aws_eks_executor\] cluster_name is
required.*AIRFLOW__AWS_EKS_EXECUTOR__CLUSTER_NAME",
+ ):
+ _get_eks_kube_client()
+
+
[email protected]
+# kubernetes_asyncio loads the CA eagerly and the fake PEM is not loadable.
[email protected](_client_factory, "_write_cluster_ca_file",
return_value=None)
[email protected](_client_factory, "time", autospec=True)
+async def test_async_client_refreshes_bearer_token(mock_time, _, mock_aws):
+ mock_time.monotonic.return_value = 1000.0
+ core_v1 = _get_eks_async_kube_client()
+
+ assert isinstance(core_v1, async_client.CoreV1Api)
+ configuration = core_v1.api_client.configuration
+ assert configuration.host == CLUSTER_ENDPOINT
+ mock_aws["fetch_token"].return_value = "k8s-aws-v1.token-2"
+ mock_time.monotonic.return_value = 1000.0 + TOKEN_REFRESH_SECONDS
+ auth = configuration.auth_settings()
+ # auth_settings() became a coroutine in kubernetes_asyncio 36.
+ if inspect.isawaitable(auth):
+ auth = await auth
+ assert auth["BearerToken"]["value"] == "Bearer k8s-aws-v1.token-2"
+ await core_v1.api_client.close()
+
+
+class TestWriteClusterCaFile:
+ CA_DATA = b64encode(CA_PEM).decode()
+
+ def test_concurrent_writes_of_one_ca_leave_one_file(self, ca_dir):
+ with ThreadPoolExecutor(max_workers=16) as pool:
+ paths = set(pool.map(lambda _:
_client_factory._write_cluster_ca_file(self.CA_DATA), range(64)))
+
+ assert len(paths) == 1
+ assert len(list(ca_dir.glob("airflow-eks-ca-*"))) == 1
+ assert Path(paths.pop()).read_bytes() == CA_PEM
+
+ def test_each_ca_gets_its_own_file_and_all_are_removed_at_exit(self,
ca_dir):
+ other_pem = CA_PEM.replace(b"fake-ca", b"other-ca")
+ path = _client_factory._write_cluster_ca_file(self.CA_DATA)
+ other_path =
_client_factory._write_cluster_ca_file(b64encode(other_pem).decode())
+
+ assert path != other_path
+ assert Path(other_path).read_bytes() == other_pem
+
+ _client_factory._remove_ca_files()
+
+ assert list(ca_dir.glob("airflow-eks-ca-*")) == []
+
+ @pytest.mark.skipif(not hasattr(os, "fork"), reason="needs os.fork")
+ def test_forked_child_reuses_parents_ca_file_and_leaves_it_at_exit(self,
ca_dir):
+ path = _client_factory._write_cluster_ca_file(self.CA_DATA)
+
+ # Fork while the lock is held, as another thread of the parent might
hold it.
+ with _client_factory._ca_files_lock:
+ pid = os.fork()
+ if pid == 0:
+ exit_code = 1
+ try:
+ # Die rather than hang on the inherited lock.
+ signal.signal(signal.SIGALRM, signal.SIG_DFL)
+ signal.alarm(10)
+ if _client_factory._write_cluster_ca_file(self.CA_DATA) ==
path:
+ # What the child's atexit hook would do.
+ _client_factory._remove_ca_files()
+ exit_code = 0
+ finally:
+ os._exit(exit_code)
+ _, status = os.waitpid(pid, 0)
+
+ assert os.waitstatus_to_exitcode(status) == 0
+ assert [str(p) for p in ca_dir.glob("airflow-eks-ca-*")] == [path]
diff --git
a/providers/amazon/tests/unit/amazon/aws/executors/eks/test_eks_executor.py
b/providers/amazon/tests/unit/amazon/aws/executors/eks/test_eks_executor.py
new file mode 100644
index 00000000000..096dce5c64b
--- /dev/null
+++ b/providers/amazon/tests/unit/amazon/aws/executors/eks/test_eks_executor.py
@@ -0,0 +1,238 @@
+# 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
+
+import importlib
+import os
+import sys
+from contextlib import nullcontext
+from unittest import mock
+
+import pytest
+from kubernetes.client import CoreV1Api
+from kubernetes.client.rest import ApiException
+
+from airflow.providers.amazon.aws.executors.eks import AwsEksExecutor,
_client_factory
+from airflow.providers.amazon.aws.executors.eks.eks_executor import (
+ _ASYNC_CLIENT_FACTORY_PATH,
+ _CLIENT_FACTORY_PATH,
+)
+from airflow.providers.cncf.kubernetes.executors.kubernetes_executor import
KubernetesExecutor
+from airflow.providers.common.compat.sdk import
AirflowOptionalProviderFeatureException, conf
+
+from tests_common.test_utils.config import conf_vars
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_2_PLUS
+
+EKS_EXECUTOR_MODULE = "airflow.providers.amazon.aws.executors.eks.eks_executor"
+EKS_EXECUTOR_PATH = f"{EKS_EXECUTOR_MODULE}.AwsEksExecutor"
+LOCAL_KUBERNETES_EXECUTOR_PATH = (
+
"airflow.providers.cncf.kubernetes.executors.local_kubernetes_executor.LocalKubernetesExecutor"
+)
+CLIENT_FACTORY_ENV_VAR = "AIRFLOW__KUBERNETES_EXECUTOR__CLIENT_FACTORY"
+REQUIRES_MULTI_TEAM = pytest.mark.skipif(not AIRFLOW_V_3_2_PLUS,
reason="Multi-team requires Airflow 3.2+")
+TEAMS = [
+ pytest.param(None, "AIRFLOW__", id="global"),
+ pytest.param(
+ "team_a",
+ "AIRFLOW__TEAM_A___",
+ id="team",
+ marks=REQUIRES_MULTI_TEAM,
+ ),
+]
+
+
+def _team_kwargs(team_name: str | None) -> dict:
+ return {"team_name": team_name} if team_name else {}
+
+
[email protected](autouse=True)
+def eks_config():
+ with mock.patch.dict(os.environ), conf_vars({("aws_eks_executor",
"cluster_name"): "test-eks-cluster"}):
+ for prefix in ("AIRFLOW__", "AIRFLOW__TEAM_A___"):
+ for key in ("CLIENT_FACTORY", "ASYNC_CLIENT_FACTORY"):
+ os.environ.pop(f"{prefix}KUBERNETES_EXECUTOR__{key}", None)
+ yield
+
+
+class TestAwsEksExecutor:
+ @conf_vars({("aws_eks_executor", "cluster_name"): None})
+ def test_missing_cluster_name_raises(self):
+ with pytest.raises(
+ ValueError, match=r"\[aws_eks_executor\]
cluster_name.*AIRFLOW__AWS_EKS_EXECUTOR__CLUSTER_NAME"
+ ):
+ AwsEksExecutor()
+
+ @pytest.mark.parametrize(("team_name", "env_prefix"), TEAMS)
+ def test_sets_client_factories_when_unset(self, team_name, env_prefix):
+ AwsEksExecutor(**_team_kwargs(team_name))
+
+ assert {k: v for k, v in os.environ.items() if
k.endswith("CLIENT_FACTORY")} == {
+ f"{env_prefix}KUBERNETES_EXECUTOR__CLIENT_FACTORY":
_CLIENT_FACTORY_PATH,
+ f"{env_prefix}KUBERNETES_EXECUTOR__ASYNC_CLIENT_FACTORY":
_ASYNC_CLIENT_FACTORY_PATH,
+ }
+ team_kwargs = _team_kwargs(team_name)
+ assert (
+ conf.getimport("kubernetes_executor", "client_factory",
**team_kwargs)
+ is _client_factory._get_eks_kube_client
+ )
+ assert (
+ conf.getimport("kubernetes_executor", "async_client_factory",
**team_kwargs)
+ is _client_factory._get_eks_async_kube_client
+ )
+
+ @conf_vars({("kubernetes_executor", "client_factory"):
_CLIENT_FACTORY_PATH})
+ def test_accepts_already_matching_client_factory(self):
+ AwsEksExecutor()
+
+ assert CLIENT_FACTORY_ENV_VAR not in os.environ
+
+ @pytest.mark.parametrize("key", ["client_factory", "async_client_factory"])
+ @pytest.mark.parametrize(("team_name", "env_prefix"), TEAMS)
+ def test_conflicting_client_factory_raises(self, monkeypatch, team_name,
env_prefix, key):
+ monkeypatch.setenv(f"{env_prefix}KUBERNETES_EXECUTOR__{key.upper()}",
"my_company.build_client")
+
+ with pytest.raises(ValueError, match=rf"{key}
itself.*my_company.build_client"):
+ AwsEksExecutor(**_team_kwargs(team_name))
+
+ @REQUIRES_MULTI_TEAM
+ @pytest.mark.parametrize(
+ ("env", "file_config", "key"),
+ [
+ pytest.param(
+ {"AIRFLOW__TEAM_A___AWS_EKS_EXECUTOR__CLUSTER_NAME":
"team-cluster"},
+ {("aws_eks_executor", "cluster_name"): None},
+ "cluster_name",
+ id="env-var-without-global-cluster",
+ ),
+ pytest.param(
+ {"AIRFLOW__TEAM_A___AWS_EKS_EXECUTOR__CONN_ID": "team_aws"},
+ {},
+ "conn_id",
+ id="env-var-for-option-with-default",
+ ),
+ pytest.param(
+ {},
+ {("team_a=aws_eks_executor", "region_name"): "eu-west-1"},
+ "region_name",
+ id="config-file-section",
+ ),
+ ],
+ )
+ def test_team_eks_config_raises(self, monkeypatch, env, file_config, key):
+ for name, value in env.items():
+ monkeypatch.setenv(name, value)
+
+ with (
+ conf_vars(file_config),
+ pytest.raises(
+ ValueError, match=rf"team-specific \[aws_eks_executor\].*team
team_a sets {key}\..*74844"
+ ),
+ ):
+ AwsEksExecutor(team_name="team_a")
+
+ @pytest.mark.parametrize(
+ ("executors", "team_name", "conflict"),
+ [
+ pytest.param(
+ f"KubernetesExecutor,{EKS_EXECUTOR_PATH}",
+ None,
+ "KubernetesExecutor in the global scope",
+ id="both-global",
+ ),
+ pytest.param(
+ f"{LOCAL_KUBERNETES_EXECUTOR_PATH},{EKS_EXECUTOR_PATH}",
+ None,
+ "LocalKubernetesExecutor in the global scope",
+ id="hybrid-global",
+ ),
+ pytest.param(
+ f"LocalExecutor;team_a=KubernetesExecutor,{EKS_EXECUTOR_PATH}",
+ "team_a",
+ "KubernetesExecutor in team team_a",
+ id="both-same-team",
+ marks=REQUIRES_MULTI_TEAM,
+ ),
+ pytest.param(
+ f"KubernetesExecutor;team_a={EKS_EXECUTOR_PATH}",
+ "team_a",
+ None,
+ id="global-kubernetes-team-eks",
+ marks=REQUIRES_MULTI_TEAM,
+ ),
+ pytest.param(EKS_EXECUTOR_PATH, None, None, id="eks-alone"),
+ ],
+ )
+ @pytest.mark.usefixtures("clean_executor_loader")
+ def test_refuses_kubernetes_executor_in_same_scope(self, executors,
team_name, conflict):
+ expectation = (
+ pytest.raises(ValueError, match=rf"alongside
{conflict}\..*issues/74844")
+ if conflict
+ else nullcontext()
+ )
+ with conf_vars({("core", "executor"): executors, ("core",
"multi_team"): "True"}), expectation:
+ AwsEksExecutor(**_team_kwargs(team_name))
+
+ @pytest.mark.parametrize("check_health", [True, False])
+ @mock.patch.object(KubernetesExecutor, "start", autospec=True)
+ def test_start_checks_cluster_access_unless_disabled(self, mock_start,
check_health):
+ executor = AwsEksExecutor()
+ executor.kube_client = mock.MagicMock(spec=CoreV1Api)
+
+ with conf_vars({("aws_eks_executor", "check_health_on_startup"):
str(check_health)}):
+ executor.start()
+
+ mock_start.assert_called_once_with(executor)
+ assert executor.kube_client.list_namespaced_pod.call_args_list == (
+ [mock.call(executor.kube_config.kube_namespace, limit=1)] if
check_health else []
+ )
+
+ @mock.patch.object(KubernetesExecutor, "start", autospec=True)
+ def test_start_fails_when_cluster_rejects_the_executor(self, mock_start):
+ executor = AwsEksExecutor()
+ executor.kube_client = mock.MagicMock(spec=CoreV1Api)
+ executor.kube_client.list_namespaced_pod.side_effect =
ApiException(status=401, reason="Unauthorized")
+
+ with pytest.raises(RuntimeError, match="401 Unauthorized"):
+ executor.start()
+
+
+class TestCncfKubernetesRequirement:
+ def test_import_fails_without_cncf_kubernetes(self, monkeypatch):
+ monkeypatch.delitem(sys.modules, EKS_EXECUTOR_MODULE)
+ monkeypatch.setitem(
+ sys.modules,
"airflow.providers.cncf.kubernetes.executors.kubernetes_executor", None
+ )
+
+ with pytest.raises(AirflowOptionalProviderFeatureException,
match=">=10.24.0"):
+ importlib.import_module(EKS_EXECUTOR_MODULE)
+
+ @pytest.mark.parametrize(
+ "provider_info",
+ [
+ pytest.param({"config": {"kubernetes_executor": {"options": {}}}},
id="no-client-factory"),
+ pytest.param({}, id="no-config"),
+ ],
+ )
+
@mock.patch("airflow.providers.cncf.kubernetes.get_provider_info.get_provider_info",
autospec=True)
+ def test_import_fails_with_cncf_kubernetes_without_client_factory(
+ self, mock_get_provider_info, monkeypatch, provider_info
+ ):
+ mock_get_provider_info.return_value = provider_info
+ monkeypatch.delitem(sys.modules, EKS_EXECUTOR_MODULE)
+
+ with pytest.raises(AirflowOptionalProviderFeatureException,
match=">=10.24.0"):
+ importlib.import_module(EKS_EXECUTOR_MODULE)
diff --git a/providers/amazon/tests/unit/amazon/aws/executors/eks/test_utils.py
b/providers/amazon/tests/unit/amazon/aws/executors/eks/test_utils.py
new file mode 100644
index 00000000000..413564b04e4
--- /dev/null
+++ b/providers/amazon/tests/unit/amazon/aws/executors/eks/test_utils.py
@@ -0,0 +1,31 @@
+# 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 airflow.providers.amazon.aws.executors.eks.utils import (
+ CONFIG_DEFAULTS,
+ CONFIG_GROUP_NAME,
+ AllEksConfigKeys,
+)
+from airflow.providers.amazon.get_provider_info import get_provider_info
+
+
+def test_config_keys_and_defaults_match_the_provider_config_section():
+ options = get_provider_info()["config"][CONFIG_GROUP_NAME]["options"]
+
+ assert set(AllEksConfigKeys()) == set(options)
+ assert {key: options[key]["default"] for key in CONFIG_DEFAULTS} ==
CONFIG_DEFAULTS
diff --git a/providers/amazon/tests/unit/amazon/aws/utils/test_eks_get_token.py
b/providers/amazon/tests/unit/amazon/aws/utils/test_eks_get_token.py
index bae936514ed..88202bf0ca3 100644
--- a/providers/amazon/tests/unit/amazon/aws/utils/test_eks_get_token.py
+++ b/providers/amazon/tests/unit/amazon/aws/utils/test_eks_get_token.py
@@ -17,9 +17,11 @@
from __future__ import annotations
import contextlib
+import os
from io import StringIO
from unittest import mock
+import boto3
import pytest
import time_machine
@@ -76,11 +78,17 @@ class TestGetEksToken:
) as mock_fetch_token:
mock_fetch_token.return_value =
"k8s-aws-v1.aHR0cDovL2V4YW1wbGUuY29t"
- with mock.patch("sys.argv", args),
contextlib.redirect_stdout(StringIO()) as temp_stdout:
+ with (
+ mock.patch("sys.argv", args),
+ mock.patch.dict(os.environ),
+ contextlib.redirect_stdout(StringIO()) as temp_stdout,
+ ):
+ os.environ.pop("AWS_STS_REGIONAL_ENDPOINTS", None)
# Import and directly call the main function rather than using
runpy
from airflow.providers.amazon.aws.utils.eks_get_token import
main
main()
+ assert os.environ["AWS_STS_REGIONAL_ENDPOINTS"] == "regional"
output = temp_stdout.getvalue()
token = "token: k8s-aws-v1.aHR0cDovL2V4YW1wbGUuY29t"
@@ -152,3 +160,46 @@ class TestGetEksToken:
# Verify the token format
assert result == "k8s-aws-v1.aHR0cDovL2V4YW1wbGUuY29t"
+
+
@mock.patch("airflow.providers.amazon.aws.utils.eks_get_token.RequestSigner")
+ @mock.patch("boto3.Session")
+ def test_fetch_access_token_for_cluster_uses_provided_session(self,
mock_session, mock_signer):
+ from airflow.providers.amazon.aws.utils.eks_get_token import
fetch_access_token_for_cluster
+
+ mock_signer.return_value.generate_presigned_url.return_value =
"http://example.com";
+ # boto3.Session is patched, so spec against the real class.
+ session = mock.MagicMock(spec=boto3.session.Session)
+
+ fetch_access_token_for_cluster(
+ "test-cluster", "https://sts.us-east-1.amazonaws.com";,
region_name="us-east-1", session=session
+ )
+
+ mock_session.assert_not_called()
+ session.client.assert_called_once_with("eks")
+
+
+def test_fetch_access_token_for_cluster_never_writes_env(monkeypatch):
+ from airflow.providers.amazon.aws.utils.eks_get_token import
fetch_access_token_for_cluster
+
+ monkeypatch.setenv("AWS_STS_REGIONAL_ENDPOINTS", "legacy")
+ env_writes = []
+ environ_cls = type(os.environ)
+ for method in ("__setitem__", "__delitem__"):
+ real = getattr(environ_cls, method)
+ monkeypatch.setattr(
+ environ_cls,
+ method,
+ lambda self, key, *args, _real=real: env_writes.append(key) or
_real(self, key, *args),
+ )
+ session = boto3.Session(
+ aws_access_key_id="testing", aws_secret_access_key="testing",
region_name="eu-west-1"
+ )
+
+ fetch_access_token_for_cluster(
+ "test-cluster",
+
"https://sts.eu-west-1.amazonaws.com/?Action=GetCallerIdentity&Version=2011-06-15";,
+ region_name="eu-west-1",
+ session=session,
+ )
+
+ assert env_writes == []
