TaskInstance queue field is treated inconsistently (maybe others as well?)

[jscheffl]

Apache Airflow version

main (development)

What happened

I was noting the task_instance_mutation_hook feature and followed the example in https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/cluster-policies.html#task-instance-mutation to see how this works to "dynamically adjust the queue" of a task instance.
Reason originally was that I noticed that the queue attribute does not support being added to template_fields, but it seems using task_instance_mutation_hook we can achieve a bit of flexibility to queuing. So far this "works" but I saw a few inconsistencies:

• When using task_instance_mutation_hook the UI in DAGs->Grid->Task Details->More Details always shows the task definition value of queue and not the mutated value, which actually is stored in DB
• Moreover I realized when navigating through the tasks in Grid (w/o starting a DAG or changing completed tasks) the pure view of the Grid was overwriting values in the DB, thus completed tasks stored in DB were over-written with task standard values - so even if a task was in the past completed in a mutated queue, after clicking around the information was reset in DB. A bit strange that "just" navigating in UI writes back data into task_instance DB table.

What you think should happen instead

When I use the task_instance_mutation_hook and actually mutated task is in DB, then the UI should display the correct state as well as when navigating through the UI data should not be changed.

How to reproduce

• Implement a task_instance_mutation_hook like described in https://airflow.apache.org/docs/apache-airflow/stable/administration-and-deployment/cluster-policies.html#task-instance-mutation.
• Run one DAG that should be affected by the hook - I used example_bash_operator (Note, add a celery worker or the changed qeue name to an existing worker to get the run completed!)
• Check values in DB, e.g. SELECT task_id, queue FROM task_instance WHERE dag_id='example_bash_operator'; - see the queue is mutated
• Check task instance detail in the UI. UI shows the task default queue value
• Navigate through different DAG runs in Grid view, check status by clicking on various tasks and also the task you checked earlier
• Check values again in DB, e.g. SELECT task_id, queue FROM task_instance WHERE dag_id='example_bash_operator'; - see the queue value is reset to task standard definition.

Operating System

Ubuntu 20.04 / Breeze Dev setup in Py 3.8 Container

Versions of Apache Airflow Providers

No specific setup, execution in Breeze from latest main

Deployment

Other

Deployment details

Development setup via breeze start-airflow...
Python 3.8
Postgres
Celery worker

Anything else

I tried to debug down to understand the root cause of the inconsistentcy but was not sure if this is "by design", a "flaw" or just "a bug". Before opening a PR I need to understand how it is desired.

I nailed down the root to be airflow/models/taskinstance.py:refresh_from_task() which is called when loading.

Open question from my side:

• Is it "intended" that the queue field is volatile (and it is stored in DB just for the work distribution)? == This is a feature, don't trust task mutation. We should add a note to documentation that is is just a tweak for scheduling but not persistent.
• Is it a inconsistency how from multiple sides the task instances are instantiated and the mutation code is not called consistently across all places? --> Fix would be to call mutation code in all places consistently
• Is the queue in DB the authoritative value? Then the refresh_from_task() might not be called to update queue except in the initial creation. --> Fix would be selectively copying queue field
• How does the problem relate to the "feature" I see in the code with pool_override - would it be consistent to make queue also officially override-able?
• How about other fields stored in task_instance table in DB: pool_slots and priority_weight - do we need to apply the same fix for these as well?

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[potiuk]

I converted it into dicussion - because it feels more like discussion than issue.

I think queue is special and for a very good reason. A lot of processing and "some" cluster policy hooks happens NOT in the scheduler, but in the worker - so at the time task is being executed. But in order for the task to be executed, you need to use "queue" to determine where it is going to be executed.

So we have chicken-egg problem:

• we want to modify queue to know where to execute the task
• but the code to change the queue has to be executed at the machine/instance that is selected by the "queue"

I believe the current cluster policy page attempts to describe the general case of where each policy is being executed, but maybe some special cases like that one could be better described?

In order to do that, someone should track all the steps and execution paths where and how each modification is made, whether it can/is saved in the DB, whether changing it has any effect (like with the queue case - changing it after it has been already used to determine where to run the code to change it, has no effect).

That's likely an interesting task for someone to do. And having somoene to do it would be great.

[jscheffl]

Thanks for the feedback. Chicken-egg problem also understood. And I also understand that the DB is primarily used to persist the data and be able to share it across the nodes. And it is almost working, except in some cases the queue information is over-ridden it seems.

I'd raise my hand as I raised the ticket initially. And I made this under the assumption there might be some "guidance" needed. I expected some kind of concept but from the feedback I assume it was just implemented as it is today to make it working in general..

I'll try to trace down the usage of the queue field in the code where it is being used and (potentially by accident) modified later. I propose to file a summary and then we can leave the discussion and I can take this forward to raise a PR. But might take some time because I assume it is rather medium to low priority.

[potiuk]

Yeah.