Repository navigation
task_instances stuck in "queued" and are missing corresponding celery_taskmeta entries #19699
Description
Apache Airflow version
2.2.2 (latest released)
Operating System
Linux Mint 20.2
Versions of Apache Airflow Providers
apache-airflow-providers-celery = "2.1.0"
apache-airflow-providers-papermill = "^2.1.0"
apache-airflow-providers-postgres = "^2.2.0"
apache-airflow-providers-google = "^6.1.0"
Deployment
Docker-Compose
Deployment details
Docker-compose deploys into our GCP k8s cluster
What happened
Hi,
we're running Airflow for our ETL pipelines.
Our DAGs run in parallel and we spawn a fair bit of parallel DAGs and tasks every morning for our pipelines.
We run our Airflow in a k8s cluster in GCP and we use Celery for our executors.
And we use autopilot to dynamically scale up and down the cluster as the workload increases or decreases, thereby sometimes tearing down airflow workers.
Ever since upgrading to Airflow 2.0 we've had a lot of problems with tasks getting stuck in "queued" or "running", and we've had to clean it up by manually failing the stuck tasks and re-running the DAGs.
Following the discussions here over the last months it looks we've not been alone :-)
But, after upgrading to Airflow 2.2.1 we saw a significant decrease in the number of tasks getting stuck (yay!), something we hoped for given the bug fixes for the scheduler in that release.
However, we still have a few tasks getting stuck (Stuck = "Task in queued") on most mornings that require the same manual intervention.
I've started digging in the Airflow DB trying to see where there's a discrepancy, and every time a task gets stuck it's missing a correspondning task in the table "celery_taskmeta".
This is a consistent pattern for the tasks that are stuck with us at this point. The task has rows in the tables "task_instance", "job", and "dag_run" with IDs referencing each other.
But the "external_executor_id" in "task_instance" is missing a corresponding entry in the "celery_taskmeta" table. So nothing ever gets executed and the task_instance is forever stuck in "queued" and never cleaned up by the scheduler.
I can see in "dag_run::last_scheduling_decision" that the scheduler is continuously re-evaluating this task as the timestamp is updated, so it's inspecting it at least, but it leaves everything in the "queued" state.
The other day I bumped our Airflow to 2.2.2, but we still get the same behavior.
And finally, whenever we get tasks that are stuck in "Queued" in this way they usually occur within the same few seconds timestamp-wise, and it correlates timewise to a timestamp when autopilot scaled down the number of airflow-workers.
If the tasks end up in this orphaned/queued state then they never get executed and are stuck until we fail them. Longest I've seen so far is a few days in this state until the task was discovered.
Restarting the scheduler does not resolve this issue and tasks are still stuck in "queued" afterwards.
Would it be possible (and a good idea?) to include in the scheduler a check if a "task_instance" row has a corresponding row in the "celery_taskmeta", and if its missing in "celery_taskmeta" after a given amount of time clean it up?
After reading about and watching Ash Berlin-Taylor's most excellent video on a deep dive into the Airflow's scheduler this does seem exactly like the check that we should add to the scheduler?
Also if there's any data I can dig out and provide for this, don't hesitate to let me know.
What you expected to happen
I expect orphaned tasks in the state queued and that are missing a corresponding entry in celery_taskmeta to be cleaned up and re-executed by the scheduler.
How to reproduce
Currently no deterministic way to reproduce other than a large amount of tasks and then remove a worker at just the right time.
Occurs every morning in a handful of tasks, but no deterministic way to reproduce it.
Anything else
No response
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
Activity
boring-cyborg commentedon Nov 19, 2021
Thanks for opening your first issue here! Be sure to follow the issue template!
ashb commentedon Nov 19, 2021
Hmmmm, yeah the "fix" for this might be to have a CeleryExecutor specific timeout (say 5mins? 1min even?) where if a task is still in queued state for longer than this and doesn't have an external_executor_id set it should get re-submitted to celery?
(Due to timing of things It is possible that a running task won't have an external_executor_id set, but that is okay, it's only queued we care about).
kristoffern commentedon Nov 19, 2021
@ashb Yes that sounds like the ideal solution to this problem.
I tried looking in the code base but I'm too unfamiliar with it to find where and how it would be best implemented.
Sounds on your description though like its in the spectrum of a somewhat easy fix?
Personally I would be fine with a 5min timeout to avoid risking killing tasks by mistake. But we wan't to keep the system responsive and fast which would be more in line with ~1min timeout. But then again the tasks get stuck so rarely compared to the overall amount of tasks that a 5min timeout shouldn't be too much of a problem (again just speaking of our own specific situation).
Thank you for the fast reply, and again, your deep dive into the scheduler was really really good 👍
ephraimbuddy commentedon Nov 23, 2021
@kristoffern When this task gets stuck, does the task_instance have value in the
external_executor_idcolumn or it's null?kristoffern commentedon Nov 23, 2021
@ephraimbuddy The
external_executor_idhas a value attached to it, but that ID can't be found in thecelery_taskmetatable.I did confirm that other
external_executor_idthat were successful had matching entries in thecelery_taskmetatable :-)It's almost as if (and this is just speculation) a task is created and a row in
external_executor_idis created, and somewhere before the system gets an ID back from the worker then that worker is killed by autopilot in GCP. And therefor no ID is added tocelery_taskmeta.But again this last part is just speculation as I haven't found a 100% deterministic way of reproducing it. 9 mornings out of 10 though we have this problem, and most of the time tasks that are stuck in "queued" all have roughly the same starting time within a second or so.
ashb commentedon Nov 23, 2021
The UUID value is generated by the celery client, so I guess what you suggest is possible.
ephraimbuddy commentedon Dec 1, 2021
@kristoffern Do you see logs like
ERROR: could not queue task ...in scheduler logs when this happens?kristoffern commentedon Dec 2, 2021
@ephraimbuddy No I haven't noticed those error messages, but I'll keep an eye out for them.
Usually when I come into work the tasks have been stuck a couple of times so I would have to go back thrrough the logs and look when it occurs the next time.
kristoffern commentedon Dec 3, 2021
@ephraimbuddy I had a task stuck this morning, confirmed that it was stuck as described in this bug by checking the DB.
It has been stuck for a few hours when I woke up. However docker had pruned the logs so that the earliest entry was still ~40min after the task had gotten stuck. Will continue to monitor for it if it's important?
ephraimbuddy commentedon Dec 3, 2021
No need for monitoring. Just wanted to confirm that it's not the same case as this: #13542 (comment)
waleedsamy commentedon May 13, 2022
I wanted to add that in my case, even though the task_instance's external_executor_id has no matching in the celery_taskmeta table and the ti state is queued, the task itself is running on a celery worker.