Metadata-Version: 2.4
Name: aca-executor
Version: 0.1.0
Summary: Airflow 3.x executor that runs each task as a throwaway Azure Container Apps Job
Author: Rafael Siqueira
License: Apache-2.0
Project-URL: Homepage, https://github.com/rafagsiqueira/aca_executor
Project-URL: Repository, https://github.com/rafagsiqueira/aca_executor
Project-URL: Issues, https://github.com/rafagsiqueira/aca_executor/issues
Keywords: airflow,executor,azure,container-apps
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: System :: Distributed Computing
Requires-Python: >=3.9
Description-Content-Type: text/markdown
Requires-Dist: apache-airflow>=3.0.0
Requires-Dist: azure-identity>=1.15.0
Requires-Dist: requests>=2.28.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: requests-mock>=1.11; extra == "dev"

# aca-executor

An Airflow **3.x** executor that runs each task as a throwaway
[Azure Container Apps Job](https://learn.microsoft.com/en-us/azure/container-apps/jobs).
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`](https://learn.microsoft.com/en-us/python/api/azure-identity/azure.identity.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

```bash
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):

```ini
[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`](https://learn.microsoft.com/en-us/azure/container-apps/storage-mounts)
resource; its name goes in `logs_volume_storage_name`.

## Testing

```bash
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).
