aca-executor

aca-executor

An Airflow 3.x executor that runs each task as a throwaway Azure Container Apps Job. It is the Azure Container Apps analogue of Airflow's Kubernetes Executor: for every task instance it creates a dedicated Manual-trigger Job, starts one execution, polls it to completion, then deletes the Job.

How it works

task queued ──► _process_workloads ──► PUT  Microsoft.App/jobs/{name}     (create Manual job)
                                          │
              sync() heartbeat loop:      ▼
   CREATING ── GET provisioningState ──► POST .../start  ──► RUNNING
   RUNNING  ── GET .../executions/{exec} ──► Succeeded → success()
                                             Failed    → fail()
                                          │
                                          ▼
                              DELETE Microsoft.App/jobs/{name}

• Each task maps to a dedicated ACA Job, named deterministically from the TaskInstanceKey (af-j-<sha1[:16]> — ACA job names are limited to 32 chars).
• The executor talks to the Azure Resource Manager REST API (management.azure.com) directly with requests.
• Authentication uses DefaultAzureCredential, so managed identity, an environment-variable service principal, or the Azure CLI login all work without extra config.
• State is polled inside sync() via a lightweight per-task state machine, so the scheduler heartbeat is never blocked waiting on Azure.

Requirements

• Apache Airflow >= 3.0 (uses the AIP-72 workload / Task Execution API interface).
• An existing Azure Container Apps managed environment.
• A worker container image that has Airflow installed and can reach the Airflow Execution API server (AIRFLOW__CORE__EXECUTION_API_SERVER_URL).
• The credential's identity needs Contributor on the resource group (to create/start/delete jobs).

Installation

pip install .            # or: uv pip install -e .

Configuration

Point Airflow at the executor and add an [aca_executor] section to airflow.cfg (or use the equivalent AIRFLOW__ACA_EXECUTOR__* env vars):

[core]
executor = aca_executor.executor.AzureContainerAppsExecutor

[aca_executor]
# --- required ---
subscription_id = 00000000-0000-0000-0000-000000000000
resource_group = my-rg
managed_environment_id = /subscriptions/.../providers/Microsoft.App/managedEnvironments/my-env
location = eastus
container_image = myregistry.azurecr.io/airflow-worker:latest

# --- optional (defaults shown) ---
cpu = 0.5
memory = 1Gi
replica_timeout = 1800
replica_retry_limit = 0
api_version = 2024-03-01
delete_job_on_completion = True
# Extra env vars injected into every task container (JSON object)
env = {"MY_VAR": "value"}

# --- logging (optional, see below) ---
logs_volume_storage_name =        # Azure Files storage registered on the env
logs_mount_path =                 # defaults to [logging] base_log_folder

Logging

This executor handles logs the same way CeleryExecutor does: it does not ship or upload logs itself. The Airflow task process running inside the ACA container uses the standard Airflow logging stack:

1. Local filesystem, standard format. The task's FileTaskHandler writes to base_log_folder using Airflow's standard path layout (dag_id/run_id/task_id/attempt=N.log), so any other component reads them exactly as it reads Celery/Kubernetes worker logs.
2. Remote logging on task conclusion. If you enable [logging] remote_logging = True with an Azure Blob (WASB) remote handler, the standard handler uploads the finished log when it closes at task exit. No executor code is involved (make sure the WASB provider is installed in the worker image).

The executor's only logging responsibility is environment passthrough: it forwards the relevant AIRFLOW__LOGGING__* options and AIRFLOW__CORE__EXECUTION_API_SERVER_URL into the container so the in-container task process is configured just like a Celery worker.

Because remote upload only happens on a clean handler close (task exit), set logs_volume_storage_name to mount an Azure Files share at base_log_folder. This preserves logs for tasks that are killed or whose Job is deleted before completion. The share must already be registered on the managed environment as a Microsoft.App/managedEnvironments/storages resource; its name goes in logs_volume_storage_name.

Testing

uv venv --python 3.12 .venv
uv pip install --python .venv/bin/python -e ".[dev]"
.venv/bin/python -m pytest

Unit tests mock the ARM REST calls (requests-mock) and the hook, so no Azure account is needed.

Not yet implemented / future work

• get_task_log is intentionally not overridden — the stock file/remote log handlers already serve task logs. A future version could query Azure Log Analytics for in-container stdout.
• get_cli_commands (executor-specific CLI helpers).