Skip to content

Add @task.eks_pod TaskFlow decorator for EksPodOperator - #72033

Open
Leondon9 wants to merge 2 commits into
apache:mainfrom
Leondon9:eks-pod-taskflow-decorator
Open

Leondon9 wants to merge 2 commits into
apache:mainfrom
Leondon9:eks-pod-taskflow-decorator

Conversation

@Leondon9

@Leondon9 Leondon9 commented Aug 24, 2026 •

Copy link
Copy Markdown
Contributor

Adds a @task.eks_pod TaskFlow decorator that runs a Python function in a pod on an Amazon EKS cluster. It wraps EksPodOperator, so the cluster kubeconfig and short-lived token come from the AWS connection (aws_conn_id) and the worker needs no aws CLI or kubeconfig.

Motivation

The usual way to run a Python function on EKS in the TaskFlow style is @task.kubernetes against a kubeconfig whose user authenticates with an aws eks get-token exec plugin. However that kubeconfig is supplied — a Kubernetes connection (kube_config), a file on the worker (kube_config_path / config_file), or the worker's default ~/.kube/config — the token is minted by an aws subprocess when the config is loaded. That has two real costs:

  • the worker (and triggerer) image must ship the aws CLI, purely for auth; and
  • AWS credentials have to be provisioned in that worker's environment (env vars, instance profile, IRSA, ...), separately from the aws_conn_id already used by every other AWS operator, so EKS auth ends up configured in two places.

(If Airflow runs inside the target EKS cluster, in-cluster auth avoids the exec plugin; this addresses the common case where it does not — a different or external cluster.)

@task.eks_pod wraps EksPodOperator, which builds the kubeconfig and a short-lived token from the AWS connection over boto3/STS. EKS then authenticates the same way as the rest of the Amazon provider, and the worker needs no aws CLI and no separately provisioned kubeconfig or credentials.

Before (@task.kubernetes; the worker authenticates via an aws eks get-token kubeconfig + the aws CLI):

@task.kubernetes(
    image="python:3.12-slim",
    kubernetes_conn_id="eks_kube_conn",  # one way to supply the exec-plugin kubeconfig
    namespace="default",
    in_cluster=False,
)
def process():
    return {"rows": 42}

After (@task.eks_pod, authenticating through the AWS connection):

@task.eks_pod(
    cluster_name="my-eks-cluster",
    image="python:3.12-slim",
    namespace="default",
    aws_conn_id="aws_default",
    region="us-east-1",
)
def process():
    return {"rows": 42}

Note

The script-injection logic is kept self-contained here rather than shared with the @task.kubernetes implementation in the cncf provider, to avoid a cross-provider private import. Factoring out a shared helper can be a follow-up.


Was generative AI tooling used to co-author this PR?
  • Yes — Claude Code (Opus 4.8)

Generated-by: Claude Code (Opus 4.8) following the guidelines

@vincbeck vincbeck left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One comment on the system test but otherwise I like it!

# Airflow 2 path
from airflow.decorators import task # type: ignore[attr-defined,no-redef]

with DAG(

@vincbeck vincbeck Aug 25, 2026 •

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please check the Amazon provider package standards to write the system test: https://github.com/apache/airflow/tree/main/providers/amazon/tests/system/amazon. You can also look at other examples from the Amazon provider package.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@vincbeck
Thanks for the review! I've reworked the system test to follow the Amazon provider conventions — it now uses SystemTestContextBuilder for the env id, creates and tears down its own cluster/nodegroup, and adds the watcher() + get_test_run() wiring, mirroring example_eks_with_nodegroups.py. Appreciate the pointer to the standards.

@Leondon9
Leondon9 force-pushed the eks-pod-taskflow-decorator branch from 233f248 to c120f31 Compare August 25, 2026 21:29
@vincbeck

Copy link
Copy Markdown
Contributor

CI is failing

@Leondon9
Leondon9 force-pushed the eks-pod-taskflow-decorator branch from c120f31 to d1be3dc Compare August 26, 2026 15:38
The TaskFlow route to EKS today is @task.kubernetes with a connection whose
kubeconfig carries an `aws eks get-token` exec plugin. The token is then minted
by an exec subprocess on the worker, so the worker must ship the AWS CLI and
credentials. This decorator wraps EksPodOperator instead, which builds the
kubeconfig and short-lived token from the AWS connection (aws_conn_id) over
boto3/STS, so the worker needs neither the CLI nor a pre-built kubeconfig.

The script-injection logic is kept self-contained rather than shared with the
@task.kubernetes implementation in the cncf provider, to avoid a cross-provider
private import; a shared helper can follow later.
@Leondon9
Leondon9 force-pushed the eks-pod-taskflow-decorator branch from d1be3dc to e4b17e3 Compare September 11, 2026 10:45
@Leondon9

Copy link
Copy Markdown
Contributor Author

This has been approved and green for a while now — just a gentle nudge in case it slipped off the radar. Happy to rebase if anything's needed; otherwise it should be ready to merge whenever it fits. Thanks!

how-to-guide: /docs/apache-airflow-providers-amazon/transfer/s3_to_dynamodb.rst
python-module: airflow.providers.amazon.aws.transfers.s3_to_dynamodb

task-decorators:

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As a new decorator is defined, should we update task-sdk/src/airflow/sdk/definitions/decorators/__init__.pyi as well?

I've opened a PR to close #48448 by adding static checks to all existed decorators to make sure the parameter list they accept match to the operator. Not sure if it would impact this.

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants