AWS ECS Executor

airflow/cli/commands/task_command.py

@@ -46,7 +46,7 @@
 from airflow.models.operator import needs_expansion
 from airflow.models.param import ParamsDict
 from airflow.models.taskinstance import TaskReturnCode
-from airflow.settings import IS_K8S_EXECUTOR_POD
+from airflow.settings import IS_EXECUTOR_CONTAINER, IS_K8S_EXECUTOR_POD
 from airflow.ti_deps.dep_context import DepContext
 from airflow.ti_deps.dependencies_deps import SCHEDULER_QUEUED_DEPS
 from airflow.typing_compat import Literal
@@ -325,7 +325,7 @@ def _move_task_handlers_to_root(ti: TaskInstance) -> Generator[None, None, None]
     console_handler = next((h for h in root_logger.handlers if h.name == "console"), None)
     with LoggerMutationHelper(root_logger), LoggerMutationHelper(ti.log) as task_helper:
         task_helper.move(root_logger)
-        if IS_K8S_EXECUTOR_POD:
+        if IS_K8S_EXECUTOR_POD or IS_EXECUTOR_CONTAINER:
             if console_handler and console_handler not in root_logger.handlers:
                 root_logger.addHandler(console_handler)
         yield

airflow/providers/amazon/aws/config_templates/__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.

airflow/providers/amazon/aws/config_templates/config.yml

@@ -0,0 +1,131 @@
+# 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.
+
+---
+
+aws_ecs_executor:
+  description: |
+    This section only applies if you are using the AwsEcsExecutor in
+    Airflow's ``[core]`` configuration.
+    For more information on any of these execution parameters, see the link below:
+    https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/client/run_task.html
+    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 ECS executor to make API calls to AWS ECS.
+      version_added: "2.8"
+      type: string
+      example: "aws_default"
+      default: "aws_default"
+    region:
+      description: |
+        The name of the AWS Region where Amazon ECS is configured. Required.
+      version_added: "2.8"
+      type: string
+      example: "us-east-1"
+      default: ~
+    assign_public_ip:
+      description: |
+        Whether to assign a public IP address to the containers launched by the ECS executor.
+        For more info see url to Boto3 docs above.
+      version_added: "2.8"
+      type: boolean
+      example: "True"
+      default: "False"
+    cluster:
+      description: |
+        Name of the Amazon ECS Cluster. Required.
+      version_added: "2.8"
+      type: string
+      example: "ecs_executor_cluster"
+      default: ~
+    container_name:
+      description: |
+        Name of the container that will be used to execute Airflow tasks via the ECS executor.
+        The container should be specified in the ECS Task Definition and will receive an airflow
+        CLI command as an additional parameter to its entrypoint. For more info see url to Boto3
+        docs above. Required.
+      version_added: "2.8"
+      type: string
+      example: "ecs_executor_container"
+      default: ~
+    launch_type:
+      description: |
+        Launch type can either be 'FARGATE' OR 'EC2'. For more info see url to
+        Boto3 docs above.
+
+        If the launch type is EC2, the executor will attempt to place tasks on
+        empty EC2 instances. If there are no EC2 instances available, no task
+        is placed and this function will be called again in the next heart-beat.
+
+        If the launch type is FARGATE, this will run the tasks on new AWS Fargate
+        instances.
+      version_added: "2.8"
+      type: string
+      example: "FARGATE"
+      default: "FARGATE"
+    platform_version:
+      description: |
+        The platform version the task uses. A platform version is only specified
+        for tasks hosted on Fargate. If one isn't specified, the LATEST platform
+        version is used.
+      version_added: "2.8"
+      type: string
+      example: "1.4.0"
+      default: "LATEST"
+    security_groups:
+      description: |
+        The comma-seperated IDs of the security groups associated with the task. If you
+        don't specify a security group, the default security group for the VPC is used.
+        There's a limit of 5 security groups. For more info see url to Boto3 docs above.
+      version_added: "2.8"
+      type: string
+      example: "sg-XXXX,sg-YYYY"
+      default: ~
+    subnets:
+      description: |
+        The comma-separated IDs of the subnets associated with the task or service.
+        There's a limit of 16 subnets. For more info see url to Boto3 docs above.
+      version_added: "2.8"
+      type: string
+      example: "subnet-XXXXXXXX,subnet-YYYYYYYY"
+      default: ~
+    task_definition:
+      description: |
+        The family and revision (family:revision) or full ARN of the task definition
+        to run. If a revision isn't specified, the latest ACTIVE revision is used.
+        For more info see url to Boto3 docs above.
+      version_added: "2.8"
+      type: string
+      example: executor_task_definition:LATEST
+      default: ~
+    max_run_task_attempts:
+      description: |
+        The maximum number of times the Ecs Executor should attempt to run a task.
+      version_added: "2.8"
+      type: int
+      example: "3"
+      default: "3"
+    run_task_kwargs:
+      description: |
+        A JSON string containing arguments to provide the ECS `run_task` API (see url above).
+      version_added: "2.8"
+      type: string
+      example: '{"tags": {"key": "schema", "value": "1.0"}}'
+      default: ~

airflow/providers/amazon/aws/executors/__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.

airflow/providers/amazon/aws/executors/ecs/Dockerfile

@@ -0,0 +1,86 @@
+# hadolint ignore=DL3007
+FROM apache/airflow:latest
+USER root
+RUN apt-get update \
+  && apt-get install -y --no-install-recommends unzip \
+  # The below helps to keep the image size down
+  && apt-get clean \
+  && rm -rf /var/lib/apt/lists/*
+RUN    curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
+RUN    unzip awscliv2.zip && ./aws/install
+
+# Add a script to run the aws s3 sync command when the container is run
+COPY <<"EOF" /entrypoint.sh
+#!/bin/bash
+
+echo "Downloading DAGs from S3 bucket"
+aws s3 sync "$S3_URL" "$CONTAINER_DAG_PATH"
+
+exec "$@"
+EOF
+
+RUN chmod +x /entrypoint.sh
+
+USER airflow
+
+## Installing Python Dependencies
+# Python dependencies can be installed by providing a requirements.txt.
+# If the file is in a different location, use the requirements_path build argument to specify
+# the file path.
+ARG requirements_path=./requirements.txt
+ENV REQUIREMENTS_PATH=$requirements_path
+
+# Uncomment the two lines below to copy the requirements.txt file to the container, and
+# install the dependencies.
+# COPY --chown=airflow:root $REQUIREMENTS_PATH /opt/airflow/requirements.txt
+# RUN pip install --no-cache-dir -r /opt/airflow/requirements.txt
+
+
+## AWS Authentication
+# The image requires access to AWS services. This Dockerfile supports 2 ways to authenticate with AWS.
+# The first is using build arguments where you can provide the AWS credentials as arguments
+# passed when building the image. The other option is to copy the ~/.aws folder to the container,
+# and authenticate using the credentials in that folder.
+# If you would like to use an alternative method of authentication, feel free to make the
+# necessary changes to this file.
+
+# Use these arguments to provide AWS authentication information
+ARG aws_access_key_id
+ARG aws_secret_access_key
+ARG aws_default_region
+ARG aws_session_token
+
+ENV AWS_ACCESS_KEY_ID=$aws_access_key_id
+ENV AWS_SECRET_ACCESS_KEY=$aws_secret_access_key
+ENV AWS_DEFAULT_REGION=$aws_default_region
+ENV AWS_SESSION_TOKEN=$aws_session_token
+
+# Uncomment the line below to authenticate to AWS using the ~/.aws folder
+# Keep in mind the docker build context when placing .aws folder
+# COPY --chown=airflow:root ./.aws /home/airflow/.aws
+
+
+## Loading DAGs
+# This Dockerfile supports 2 ways to load DAGs onto the container.
+# One is to upload all the DAGs onto an S3 bucket, and then
+# download them onto the container. The other is to copy a local folder with
+# the DAGs onto the container.
+# If you would like to use an alternative method of loading DAGs, feel free to make the
+# necessary changes to this file.
+
+ARG host_dag_path=./dags
+ENV HOST_DAG_PATH=$host_dag_path
+# Set host_dag_path to the path of the DAGs on the host
+# COPY --chown=airflow:root $HOST_DAG_PATH $CONTAINER_DAG_PATH
+
+
+# If using S3 bucket as source of DAGs, uncommenting the next ENTRYPOINT command will overwrite this one.
+ENTRYPOINT []
+
+# Use these arguments to load DAGs onto the container from S3
+ARG s3_url
+ENV S3_URL=$s3_url
+ARG container_dag_path=/opt/airflow/dags
+ENV CONTAINER_DAG_PATH=$container_dag_path
+# Uncomment the line if using S3 bucket as the source of DAGs
+# ENTRYPOINT ["/entrypoint.sh"]