[WIP] AIP-116 Interactive DAG processor

Status

Motivation

What problem does it solve? / Why is it needed?

From the very beginning Airflow relied on DAG parsing loop continuously re-parsing and maintaining the state of DAGs in Airflow deployment. This simple approach served Airflow for a long time, but ultimately made a number of basic use cases hard or impossible to implement:

• No way to synchronously update a DAG in a fast & reliable manner. To deploy a new version of the DAG, the user needs to deliver a new version of the code (1) to components and wait until the DAG processor eventually picks up a newer version (2). For larger Airflow deployments the delay between (1) and (2) can be significant. There are workarounds (like bumping the priority),but they either don’t achieve a full sync parsing process or have huge latency overhead (like airflow dags reserialize -S <subdir>).
• A safer rollout of new DAG code is limited by the fact that any import error removes the previous non-broken version of the DAG and disrupts the execution if it was running. This came from the fact that DAG removal and absence of DAG on next parsing iteration are equivalent in the current implementation. 
• AI agents are harder to integrate with Airflow deployments, as eventually-consistent DAG ingestion requires multiple tool invocations.
• Most of the parsing iterations are throw-away CPU cycles. While dynamically generated DAGs (like the one depending on Variables) are supported, most of the DAGs do not require it and are immutable. If they are baked into the container image (a fairly common case) it would’ve been sufficient to parse them once. But to satisfy a relatively small percentage of dynamic DAGs there is no option to save up resources by avoiding extra re-parsing iterations (partially solved by AIP-66 manifest options).

Originally AIP-85 and this AIP proposed a single set of changes across the DAG processing. With multiple Airflow 3 improvements landed (AIP-72, AIP-66, DagBag refactoring into vanilla and DBDagBag), it made sense to split bigger initiative proposal(which aims to establish extendable DAG parsing controls within Airflow) into 2 relatively independent tracks: AIP-85 DAG importer and this new AIP.

Considerations

What change do you propose to make?

In addition to a regular (async) DAG processor, create an alternative "Interactive" DAG processor, which will act as an internal control-plane API for DAGs parsing.

Proposed internal API:

# Parse and return DAGs/import errors for bundle and (relative) path within a bundle (no ingestion)
- POST /bundle/parse { “bundle”, “path”, “bundle_version” } ->  { “import_errors”, “dags” }
# Parse, ingest and return ingested DAG IDs/import errors for bundle and (relative) path within a bundle.
- POST /bundle/parse_update { “bundle”, “path”, “bundle_version”, "update_options" } ->  { “import_errors”, dict[“dag_id”, “dag_version”] }
# Disable DAGs (by ID or relative path)
- DELETE /bundle {“bundle”, “path”, “dag_ids” }
# Get bundle version metadata
- GET /bundle -> { “bundle”, “bundle_version” }
# Pin bundle version for specified DAGs
- POST /bundle/set_version { “bundle”, “dag_ids”, “bundle_version” }

Implementation

A new flag `airflow dag-processor --interactive-api` will start a FastAPI described above. It can run aside a regular async dag-processor, but sets of bundles must be non-overlapping between interactive and async dag-processors and designated at such in bundle configuration (`dag_bundle_config_list` Airflow configuration).

DAG model is expanded with an optional `default_dag_version` relation to DAG version, which can be used to pin an older version for an individual DAG.

Pre-requisites

• AIP-92
• Callbacks migration (part of AIP-92) into a separate processing unit/workers.

Open questions

• Detailed authn/authz for DAG processor. One of the desired features - ability to operate a "read-only" mode DAG processor, which would require authorization to access only designated connections via Internal task SDK API and no write access to perform DAG ingestion in DAG processing API. This AIP that the question of a long-term authn/authz schema is figured out, but we anticipate that additional might be required to simplify the setup of read-only mode and prevent accidental misconfigurations (e.g. designate some connections as read-only and throw an error if read-only mode is requested, but access to non-read-only connections is granted).

Compatibility

As this is an introduction of a new component, no real impact on the existing setups is anticipated. Configuration changes should be backwards compatible.

PoC: https://github.com/mpapierowski/airflow/tree/demo/mcp 

Scope

• DagModel is expanded with a relation for setting a default DAG version
• Interactive DAG processor (API) is implemented as a standalone FastAPI
• Airflow CLI command `dag-processor` is adjusted to starting interactive processor via flags