Using @task.kubernetes decorator in a dag causes different dag hash on every serialized dag update
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 10h
- Merged PRs (30d)
- 483
Description
### Apache Airflow Provider(s)
cncf-kubernetes
### Versions of Apache Airflow Providers
apache-airflow-providers-cncf-kubernetes==9.0.1
### Apache Airflow version
2.10.5
### Operating System
Linux
### Deployment
Other 3rd-party Helm chart
### Deployment details
**Airflow Compute:** GKE
- **scheduler replicas:** 2
**Airflow Meta Database:** MySQL 8.0.31
**Main Airflow configuration:**
_executor:_ KubernetesExecutor
_AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION:_ False
_AIRFLOW__CORE__MAX_NUM_RENDERED_TI_FIELDS_PER_TASK:_ 0
_AIRFLOW__CORE__PARALLELISM:_ 250
### What happened
Using @task.kubernetes in a DAG causes dag hash to change every time serialized dag update occurs based on AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL. The change in dag hash for every serialized dag update only occurs when multiple schedulers are configured. In my case I have 2 schedulers so I see the dag hash flip flop between two different values. After troubleshooting and going through the Airflow 2.10.5 source code, I see the root cause of the issue to be [this code statement](https://github.com/apache/airflow/blob/b93c3db6b1641b0840bd15ac7d05bc58ff2cccbf/airflow/providers/cncf/kubernetes/decorators/kubernetes.py#L60)
`template_fields: Sequence[str] = tuple(
{"op_args", "op_kwargs", *KubernetesPodOperator.template_fields} - {"cmds", "arguments"}
)`
template_fields are initialized using set which has non-deterministic order which results into different dag hash on different scheduler. The frequent change in dag hash amplifies update dag_run statements during scheduling
`UPDATE
`dag_run`
SET
`last_scheduling_decision` = ?,
`dag_hash` = ?,
`updated_at` = ?
WHERE
`dag_run` . `id` = ?`
`UPDATE
`dag_run`
SET
`last_scheduling_decision` = ?,
`updated_at` = ?
WHERE
`dag_run` . `id` = ?`
`UPDATE
`dag_run`
SET
`dag_hash` = ?,
`updated_at` = ?
WHERE
`dag_run` . `id` = ?`
This additional updates causes more lock waits in the database. This especially becomes a huge problem for the DAGs with dynamic task groups expanding into double digit tasks
### What you think should happen instead
The order of template_fields should be consistent and should not cause change in dag hash when nothing has changed in the actual DAG
### How to reproduce
1. Create a simple dag with @task.kubernetes
```"""DAG file to load test Dynamic task mapping group"""
from airflow.decorators import task, task_group
from airflow.models.dag import DAG
@task
def make_list(no_tasks: str):
"""Generate and return a list of values
Args:
no_tasks (str): Number of tasks to generate
"""
return [i for i in range(int(no_tasks if no_tasks else 100))]
@task
def consumer(arg):
"""Consume the list of values"""
print(arg)
@task
def agent(arg):
"""Agent to handle the task"""
print(arg)
@task.kubernetes(
image="python:3.8-slim-buster",
queue="kubernetes",
)
def company(arg):
"""Company to handle the task"""
print(arg)
@task_group
def supplier(arg):
"""Supply the list of values"""
consumer(arg)
agent(arg)
company(arg)
with DAG(dag_id="dynamic-map-group", schedule_interval=None) as dag:
no_of_tasks = "{{ dag_run.conf['no_of_tasks'] }}"
supplier.expand(arg=make_list(no_of_tasks))
```
2. Ensure that multiple schedulers are configured
3. Once DAG is deployed, connect to the meta database and query serialized_dag table
`select dag_hash, last_updated, data from serialized_dag where dag_id = 'dynamic-map-group'`
When you query multiple times, you will see the dag_hash changes every configured AIRFLOW__CORE__MIN_SERIALIZED_DAG_UPDATE_INTERVAL
4. Run the dag with a very high number to create large number of tasks in the dynamic task group
5. Observe the database level top queries and lock waits, you will see multiple update statements to dag_run table resulting into higher number of lock waits
### Anything else
- I checked the most recent source code for airflow 3 and in my observation the latest version may have the same issue, I do not see template fields being preserved in order during serialization
- Similar issue with template_fields order may exist in other places as well that could impact the dag hash and the issue with additional scheduling updates e.g. KubernetesJobOperator
- I was able to test with changing the template_fields logic for kubernetes decorator to keep deterministic order and confirm that dag hash does not change and does not result into extra updates to dag_run
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start in airflow/providers/cncf/kubernetes/decorators/kubernetes.py at the template_fields definition identified in the report. Reproduce with the provided @task.kubernetes DAG and multiple schedulers, then query the serialized_dag table to compare hashes across updates. Done means template field ordering is deterministic and the DAG hash remains unchanged when the DAG has not changed.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- kubernetes, python
- Domain
- backend, data-engineering
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Stale
- Clarity
- Clearly specified
- Newbie friendliness
- 55/100