This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 c4ab82253de Report AwsEksExecutor as the caller in its AWS user agent
(#75698)
c4ab82253de is described below
commit c4ab82253de4bb9e1d301e1bde201e84b55429c9
Author: Sean Ghaeli <[email protected]>
AuthorDate: Tue Oct 27 07:42:38 2026 -0700
Report AwsEksExecutor as the caller in its AWS user agent (#75698)
The executor's only AWS call, eks:DescribeCluster, was tagged
Caller/KubernetesExecutor,
or Caller/AirflowKubernetesScheduler from the pod watcher, so its usage
could not be told
apart from the plain KubernetesExecutor.
---
.../amazon/aws/executors/eks/_client_factory.py | 8 +++++-
.../aws/executors/eks/test__client_factory.py | 32 +++++++++++++++++++++-
.../amazon/aws/executors/eks/test_eks_executor.py | 2 +-
3 files changed, 39 insertions(+), 3 deletions(-)
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
index eb0acccf8b8..14846b26812 100644
---
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
@@ -63,6 +63,12 @@ def _get_sync_token_key(kubernetes_version: str) -> str:
_SYNC_TOKEN_KEY = _get_sync_token_key(kubernetes.__version__)
+class _EksExecutorHook(EksHook):
+ # Without this, the user agent reports KubernetesExecutor as the caller.
+ def _get_caller(self, target_function_name: str = "execute") -> str:
+ return "AwsEksExecutor"
+
+
def _get_eks_kube_client(team_name: str | None = None) -> client.CoreV1Api:
"""Build a Kubernetes client for the team's EKS cluster, in memory and
without a kubeconfig."""
from kubernetes import client
@@ -105,7 +111,7 @@ def _configure_eks_auth(
**team_kwargs,
)
- eks_hook = EksHook(aws_conn_id=conn_id, region_name=region_name)
+ eks_hook = _EksExecutorHook(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(
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
index f3b10ad6bac..454206e0061 100644
---
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
@@ -63,7 +63,7 @@ def ca_dir(tmp_path, monkeypatch):
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, "_EksExecutorHook") 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,
):
@@ -224,6 +224,36 @@ async def test_builds_client_for_the_teams_cluster(_,
mock_aws, monkeypatch, fac
await core_v1.api_client.close()
[email protected]("factory", [_get_eks_kube_client,
_get_eks_async_kube_client], ids=["sync", "async"])
[email protected]
[email protected](_client_factory, "_write_cluster_ca_file", autospec=True,
return_value=None)
[email protected](_client_factory, "fetch_access_token_for_cluster",
autospec=True, return_value="token")
[email protected](_client_factory, "StsHook", autospec=True)
[email protected]("botocore.client.BaseClient._make_api_call", autospec=True)
+async def test_describe_cluster_user_agent_names_the_executor(
+ mock_api_call, mock_sts_hook, _fetch_token, _write_ca_file, monkeypatch,
factory
+):
+ monkeypatch.setenv("AIRFLOW__AWS_EKS_EXECUTOR__CLUSTER_NAME", CLUSTER_NAME)
+ monkeypatch.setenv("AIRFLOW__AWS_EKS_EXECUTOR__REGION_NAME", "us-east-1")
+ monkeypatch.setenv("AIRFLOW_CONN_AWS_DEFAULT", "aws://")
+ mock_api_call.return_value = {
+ "cluster": {
+ "status": "ACTIVE",
+ "endpoint": CLUSTER_ENDPOINT,
+ "certificateAuthority": {"data": b64encode(CA_PEM).decode()},
+ }
+ }
+ mock_sts_hook.return_value.conn_client_meta.endpoint_url =
"https://sts.us-east-1.amazonaws.com";
+
+ core_v1 = factory()
+
+ eks_client, operation_name = mock_api_call.call_args.args[:2]
+ assert operation_name == "DescribeCluster"
+ assert "Caller/AwsEksExecutor" in
eks_client.meta.config.user_agent_extra.split(" ")
+ if inspect.iscoroutinefunction(core_v1.api_client.close):
+ await core_v1.api_client.close()
+
+
@pytest.mark.asyncio
# kubernetes_asyncio loads the CA eagerly and the fake PEM is not loadable.
@mock.patch.object(_client_factory, "_write_cluster_ca_file",
return_value=None)
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
index db78f1bab42..cb86e77e8ee 100644
--- 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
@@ -68,7 +68,7 @@ def eks_config():
@pytest.fixture
def mock_aws():
with (
- mock.patch.object(_client_factory, "EksHook") as eks_hook_cls,
+ mock.patch.object(_client_factory, "_EksExecutorHook") 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,
):
