Hi all,
I'd like to start the discussion on AIP-117, Per-run Task Scheduling Controls:
https://cwiki.apache.org/confluence/spaces/AIRFLOW/pages/446071748/WIP+AIP-117+Per-Run+Task+Scheduling+Controls
A task's pool, pool_slots, priority_weight and queue are fixed in the Dag file.
You can't change them for one run when you trigger it, and a Dag author can't
make them depend on a param or on the kind of run. Today people duplicate tasks
and branch between them, keep two copies of a Dag, or ask their Deployment
Manager for a cluster policy that reads dag_run.conf.
The proposal has two parts.
1. Lookups, for Dag authors. A task field can take its value from a Dag param,
the run type, or an Airflow Variable instead of a literal, for example
pool=Lookup.param("warehouse_pool"). A param used this way must declare enum
(or minimum and maximum for numbers), so whoever triggers the run can only pick
values the author listed, and trigger permission is enough.
2. Trigger-time overrides, for operators. An optional "overrides" field on the
trigger request sets these fields for specific tasks, TaskGroups or pools, for
Dags whose author didn't prepare anything. It needs Dag edit permission and
only works with pools and queues an administrator has nominated. Nothing is
nominated by default, so nothing changes on upgrade.
Values are resolved when the run is created and applied whenever task instances
are created, cleared or retried. A task_instance_mutation_hook still runs last
and wins. The scheduler never runs user code, and its critical section is
unchanged.
To make this concrete, here is roughly what the docs would say once it lands:
- Params: "A task's pool, pool slots, priority weight or queue can come from a
param, for example pool=Lookup.param("warehouse_pool") with
Param(default="wh_small", enum=["wh_small", "wh_large"]). The trigger form
shows it as a dropdown and notes which tasks it affects. Scheduled runs use the
default. The same value can be passed as conf from the REST API, the CLI,
TriggerDagRunOperator or a backfill."
- Dag runs: "If you can edit a Dag, you can override these fields for one run
without changing the Dag, for example {"overrides": {"tasks": {"extract":
{"pool": "heavy_pool"}}}}. Keys can be task IDs or TaskGroup IDs, and "pools":
{"etl_pool": "etl_overflow_pool"} moves every task of the run from one pool to
another. Only pools and queues your administrator has nominated can be used."
- Pools: "A pool can't be used for trigger-time overrides until it's marked as
an override target. Treat that as letting anyone who can edit a Dag send its
tasks to this pool. Removing the mark takes effect immediately, and runs
already in progress fall back to their normal pool."
- Cluster policies: "Lookups and overrides are applied before
task_instance_mutation_hook, so the hook keeps the last word. task_policy runs
at parse time and only sees defaults, so put rules that must hold for every run
in task_instance_mutation_hook."
I'd especially like feedback on:
- The security model: authors bound what the person triggering can pick, and
administrators nominate what Dag editors can override to.
- The name "Lookup", which is a working name.
Thanks,
Ramit
