You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by GitBox <gi...@apache.org> on 2022/05/02 22:07:38 UTC
[GitHub] [airflow] repl-chris opened a new pull request, #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
repl-chris opened a new pull request, #23432:
URL: https://github.com/apache/airflow/pull/23432
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
Celery can lose tasks on worker shutdown, causing airflow to just wait on them indefinitely (may be related to https://github.com/celery/celery/issues/7266). This PR detects these "hung" tasks and sets them back to `SCHEDULED` state so the scheduler can queue them up again.
Closes: #19699
This is basically a resurrection of [PR19769](https://github.com/apache/airflow/pull/19769) which was [reverted](https://github.com/apache/airflow/pull/21556) because no one could reproduce/test it. We can reproduce the problem reliably in an isolated test environment, so we were able to test this fairly thoroughly.
cc: @ephraimbuddy @kristoffern @kaxil
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in a newsfragement file, named `{pr_number}.significant.rst`, in [newsfragments](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r863469745
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,64 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be reschelduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
Review Comment:
Ooooh yeah we can totally do that. I thought about it but didn't think the job_id was available in the executor, and it seemed wrong to punch it through....but I see now that it is in fact already there, I will make the change š
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on PR #23432:
URL: https://github.com/apache/airflow/pull/23432#issuecomment-1120081308
> For that reason I think I'd like you to look at If we can see if the taskmeta is _actually_ lost, to separate "lost" from "just in a queue behind n long running tasks."
@ashb Yes, this would be great, I agree. I'm far from a celery expert but as far as I can tell celery does not provide any capability to do this. Any implementation would need to be broker-specific, and directly access celery internals, reading pending work items from the queue directly. This is not possible on SQS or rabbit afaik...but I've only witnessed this problem on redis, so maybe the fix only needs to work on redis. Even on redis the solution isn't pretty though...
The redis queue in celery is implemented using a redis "list" data type, which, internally is a linked-list. So, there's no great way to check if it contains a given task - we need to scan the entire list O(n) to check if a single task is in there. Redis does provide an `LPOS` command which could do this scan for us, but it's still O(n), and it's only in newer versions of redis (>=6.0.6). The list contains serialized JSON documents for the work items, so to use LPOS we'd need to synthetically generate a character-perfect copy of that JSON document, which uses an undocumented format. I think in theory it may be possible to use LUA scripting to do this scan...I'm extremely open to other ideas, but I think the only way we could realistically make this work decently is to grab our oldest known "PENDING" task_id, send a LUA script to redis which scans the list doing a `string.find(list_element, '"task_id"="our-task-id"')` sorta thing.....and then hope celery doesn't change how it works
š This whole direction sounds like a pretty bad idea IMHO...but I could go in this direction if we actually want to go there...
Is rescheduling a "just-in-a-long-queue" task really that bad? Maybe I don't fully grasp the impacts of it, but revoking and re-enqueuing a task which was healthy and waiting seems like a pretty good option compared to the alternative š
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r867255995
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,71 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be rescheduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ TaskInstance.queued_by_job_id == self.job_id,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
+ latest_states = self.bulk_state_fetcher.get_many(
+ [self.tasks[ti.key] for ti in queued_too_long]
+ )
+ queued_too_long = [
+ ti
+ for ti in queued_too_long
+ if latest_states[ti.external_executor_id][0] == celery_states.PENDING
+ ]
+ except AirflowTaskTimeout:
+ # This "latest state" check isn't super important, so if it's taking too long we'll
+ # just log it and continue on
+ self.log.debug("Timed out while loading latest celery task states for stuck queued tasks")
+
+ for task in queued_too_long:
+ self.log.info(
+ 'TaskInstance: %s found in queued state for more than %s seconds, rescheduling',
+ task,
+ self.task_adoption_timeout.total_seconds(),
Review Comment:
It actually is using this `task_adoption_timeout` setting for the "max queued time"....it only has its own setting for "run frequency"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r863469871
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,64 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be reschelduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
+ latest_states = self.bulk_state_fetcher.get_many(
+ [self.tasks[ti.key] for ti in queued_too_long]
+ )
+ queued_too_long = [
+ ti for ti in queued_too_long
+ if latest_states[ti.external_executor_id][0] == celery_states.PENDING
+ ]
+ except AirflowTaskTimeout:
+ # This "latest state" check isn't super important, so if it's taking too long we'll
+ # just log it and continue on
+ self.log.debug("Timed out while loading latest celery task states for stuck queued tasks")
+
+ for task in queued_too_long:
+ self.log.info(
+ 'TaskInstance: %s found in queued state for more than %s seconds, rescheduling',
+ task,
+ self.task_adoption_timeout.total_seconds(),
+ )
+ task.state = State.SCHEDULED
Review Comment:
Seems like a good idea, will do
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r866976840
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,71 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be rescheduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ TaskInstance.queued_by_job_id == self.job_id,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
+ latest_states = self.bulk_state_fetcher.get_many(
+ [self.tasks[ti.key] for ti in queued_too_long]
+ )
+ queued_too_long = [
+ ti
+ for ti in queued_too_long
+ if latest_states[ti.external_executor_id][0] == celery_states.PENDING
+ ]
+ except AirflowTaskTimeout:
+ # This "latest state" check isn't super important, so if it's taking too long we'll
+ # just log it and continue on
+ self.log.debug("Timed out while loading latest celery task states for stuck queued tasks")
Review Comment:
If it times out then I don't think we should continue.
##########
airflow/config_templates/config.yml:
##########
@@ -1788,6 +1788,13 @@
type: string
example: ~
default: "False"
+ - name: stuck_queued_task_check_interval
Review Comment:
Does this need a new setting, or is the existing "orphan" settings enough here?
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,71 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be rescheduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ TaskInstance.queued_by_job_id == self.job_id,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
+ latest_states = self.bulk_state_fetcher.get_many(
+ [self.tasks[ti.key] for ti in queued_too_long]
+ )
+ queued_too_long = [
+ ti
+ for ti in queued_too_long
+ if latest_states[ti.external_executor_id][0] == celery_states.PENDING
+ ]
+ except AirflowTaskTimeout:
+ # This "latest state" check isn't super important, so if it's taking too long we'll
+ # just log it and continue on
+ self.log.debug("Timed out while loading latest celery task states for stuck queued tasks")
+
+ for task in queued_too_long:
+ self.log.info(
+ 'TaskInstance: %s found in queued state for more than %s seconds, rescheduling',
+ task,
+ self.task_adoption_timeout.total_seconds(),
+ )
+ task.state = State.SCHEDULED
+ task.queued_dttm = None
+ session.merge(task)
Review Comment:
```suggestion
```
Not needed. task is already attached to the session (from when we queried it on L385)
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,71 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be rescheduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ TaskInstance.queued_by_job_id == self.job_id,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
+ latest_states = self.bulk_state_fetcher.get_many(
+ [self.tasks[ti.key] for ti in queued_too_long]
+ )
+ queued_too_long = [
+ ti
+ for ti in queued_too_long
+ if latest_states[ti.external_executor_id][0] == celery_states.PENDING
+ ]
+ except AirflowTaskTimeout:
+ # This "latest state" check isn't super important, so if it's taking too long we'll
+ # just log it and continue on
+ self.log.debug("Timed out while loading latest celery task states for stuck queued tasks")
+
+ for task in queued_too_long:
+ self.log.info(
+ 'TaskInstance: %s found in queued state for more than %s seconds, rescheduling',
+ task,
+ self.task_adoption_timeout.total_seconds(),
Review Comment:
Wrong timer.
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,71 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be rescheduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ TaskInstance.queued_by_job_id == self.job_id,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
+ latest_states = self.bulk_state_fetcher.get_many(
+ [self.tasks[ti.key] for ti in queued_too_long]
+ )
+ queued_too_long = [
+ ti
+ for ti in queued_too_long
+ if latest_states[ti.external_executor_id][0] == celery_states.PENDING
+ ]
+ except AirflowTaskTimeout:
+ # This "latest state" check isn't super important, so if it's taking too long we'll
+ # just log it and continue on
+ self.log.debug("Timed out while loading latest celery task states for stuck queued tasks")
+
+ for task in queued_too_long:
Review Comment:
```suggestion
for ti in queued_too_long:
```
etc.
Naming is important -- this is a TaskInstance, not a Task
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on PR #23432:
URL: https://github.com/apache/airflow/pull/23432#issuecomment-1121201666
@ashb As another option, I could stop using the task_adoption_timeout and instead add an explicit `stuck_queued_task_timeout` setting, and have it disabled by default. Then there isn't a behavioural change unless the user explicitly enables that setting....
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r867250315
##########
airflow/config_templates/config.yml:
##########
@@ -1788,6 +1788,13 @@
type: string
example: ~
default: "False"
+ - name: stuck_queued_task_check_interval
Review Comment:
the orphan/adoption stuff doesn't really have an appropriate setting to use instead afaik - the "adopted task timeout" runs on every heartbeat, which is fine because it's very cheap unless a task has actually timed out....in contrast, this check will (minimally) issue a database query every time it runs, which we probably don't want to do on every heartbeat.
Alternatively, instead of issuing the query to find these tasks we could internally track "sent to celery time" and "last known state" by task....then we wouldn't need to issue the query and we could run on every heartbeat instead...I kinda like this option, but if we're talking about separating "lost" from "just in a long queue" then it starts to get sketchy (see separate comment below)
In theory this feature also largely makes `_check_for_stalled_adopted_tasks()` unnecessary, as "stalled adopted tasks" and "lost tasks" are effectively the same thing. I didn't remove `_check_for_stalled_adopted_tasks()` though as I figured it added unnecessary risk, and I was trying to stay true to the design of the original PR by ephraimbuddy. A stalled adopted task, I guess, also has a slight behavioural difference - it fails the task rather than re-scheduling it (which I'm not entirely convinced is really correct)
##########
airflow/config_templates/config.yml:
##########
@@ -1788,6 +1788,13 @@
type: string
example: ~
default: "False"
+ - name: stuck_queued_task_check_interval
Review Comment:
the orphan/adoption stuff doesn't really have an appropriate setting to use instead afaik - the "adopted task timeout" runs on every heartbeat, which is fine because it's very cheap unless a task has actually timed out....in contrast, this check will (minimally) issue a database query every time it runs, which we probably don't want to do on every heartbeat.
Alternatively, instead of issuing the query to find these tasks we could internally track "sent to celery time" and "last known state" by task....then we wouldn't need to issue the query and we could run on every heartbeat instead...I kinda like this option, but if we're talking about separating "lost" from "just in a long queue" then it starts to get sketchy (see separate comment below)
In theory this feature also largely makes `_check_for_stalled_adopted_tasks()` unnecessary, as "stalled adopted tasks" and "lost tasks" are effectively the same thing. I didn't remove `_check_for_stalled_adopted_tasks()` though as I figured it added unnecessary risk, and I was trying to stay true to the design of the original PR by ephraimbuddy. A stalled adopted task, I guess, also has a slight behavioural difference - it fails the task rather than re-scheduling it (which I'm not entirely convinced is really correct)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r863502024
##########
airflow/config_templates/config.yml:
##########
@@ -1788,6 +1788,13 @@
type: string
example: ~
default: "False"
+ - name: stuck_queued_task_check_interval
+ description: |
+ How often to check for stuck queued tasks (in seconds)
+ version_added: 2.3.0
Review Comment:
```suggestion
version_added: 2.3.1
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on PR #23432:
URL: https://github.com/apache/airflow/pull/23432#issuecomment-1119073137
FYI @ephraimbuddy @tanelk @eladkal @kaxil @XD-DENG @ashb this is ready for re-review
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ephraimbuddy commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
ephraimbuddy commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r866806879
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,71 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be rescheduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ TaskInstance.queued_by_job_id == self.job_id,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
Review Comment:
One issue we had before was load on the scheduler. Is the UI complaining about the scheduler? Also, how often do you get the AirflowTaskTimeout error?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] boring-cyborg[bot] commented on pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
boring-cyborg[bot] commented on PR #23432:
URL: https://github.com/apache/airflow/pull/23432#issuecomment-1115417708
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:
- Pay attention to the quality of your code (flake8, mypy and type annotations). Our [pre-commits]( https://github.com/apache/airflow/blob/main/STATIC_CODE_CHECKS.rst#prerequisites-for-pre-commit-hooks) will help you with that.
- In case of a new feature add useful documentation (in docstrings or in `docs/` directory). Adding a new operator? Check this short [guide](https://github.com/apache/airflow/blob/main/docs/apache-airflow/howto/custom-operator.rst) Consider adding an example DAG that shows how users should use it.
- Consider using [Breeze environment](https://github.com/apache/airflow/blob/main/BREEZE.rst) for testing locally, itās a heavy docker but it ships with a working Airflow and a lot of integrations.
- Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
- Please follow [ASF Code of Conduct](https://www.apache.org/foundation/policies/conduct) for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
- Be sure to read the [Airflow Coding style]( https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#coding-style-and-best-practices).
Apache Airflow is a community-driven project and together we are making it better š.
In case of doubts contact the developers at:
Mailing List: dev@airflow.apache.org
Slack: https://s.apache.org/airflow-slack
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r863472591
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,64 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be reschelduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
Review Comment:
Come to think of it, even with that we will still need the `self.tasks` post-filter to restrict it to only tasks which have already been sent to celery....I guess it will cut down on this executor loading another instance's tasks tho, so probably still worth it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tanelk commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
tanelk commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r863441937
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,64 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be reschelduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
+ latest_states = self.bulk_state_fetcher.get_many(
+ [self.tasks[ti.key] for ti in queued_too_long]
+ )
+ queued_too_long = [
+ ti for ti in queued_too_long
+ if latest_states[ti.external_executor_id][0] == celery_states.PENDING
+ ]
+ except AirflowTaskTimeout:
+ # This "latest state" check isn't super important, so if it's taking too long we'll
+ # just log it and continue on
+ self.log.debug("Timed out while loading latest celery task states for stuck queued tasks")
+
+ for task in queued_too_long:
+ self.log.info(
+ 'TaskInstance: %s found in queued state for more than %s seconds, rescheduling',
+ task,
+ self.task_adoption_timeout.total_seconds(),
+ )
+ task.state = State.SCHEDULED
Review Comment:
Should we call celerys [AsyncResult.revoke()](https://docs.celeryq.dev/en/stable/reference/celery.result.html#celery.result.AsyncResult.revoke) here? Airflow can handle duplicate execution atempts, but it will add some logging noise.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb closed pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
ashb closed pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
URL: https://github.com/apache/airflow/pull/23432
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on PR #23432:
URL: https://github.com/apache/airflow/pull/23432#issuecomment-1123148412
> Iām either misunderstanding this or itās not accurate. At least, there is a `orphaned_tasks_check_interval` and in our experience it gets used.
Oooh yes, you're absolutely correct. I had the blinders on and was only looking at the "adopted task timeout" functionality of the celery executor - I totally forgot about the actual orphaned task adoption functionality of the scheduler š
> You might try to look at `app.control.inspect().reserved` for tasks that got reserved but didnāt make it to a worker process.
Thanks for the tip...I'll look into this a little further but I'm pretty sure they won't be there. My understanding is that reserved tasks will eventually get re-delivered to a different worker after [6 hours](https://github.com/apache/airflow/blob/main/airflow/config_templates/default_celery.py#L37), but I've had tasks stuck for far longer than that...and I did also verify that the stuck tasks did not exist in the `unacked` redis list (which I'd expect is the data source for `inspect().reserved`). Never-the-less I will dig into this a little further, thanks! :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] tanelk commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
tanelk commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r863427637
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,64 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be reschelduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
Review Comment:
Could we filter on `queued_by_job_id` in the query?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on a diff in pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on code in PR #23432:
URL: https://github.com/apache/airflow/pull/23432#discussion_r867199084
##########
airflow/executors/celery_executor.py:
##########
@@ -358,6 +369,71 @@ def _check_for_stalled_adopted_tasks(self):
for key in timedout_keys:
self.change_state(key, State.FAILED)
+ @provide_session
+ def _clear_stuck_queued_tasks(self, session: Session = NEW_SESSION) -> None:
+ """
+ Tasks can get lost by celery if the celery worker is shut down while it is picking up
+ a new work item (observed when using redis broker). When this happens the task is stuck
+ in queued state indefinitely until the scheduler is restarted (which causes the task to be
+ picked up by the adoption code and re-scheduled at that point). This function is intended
+ to detect that situation and re-schedule those tasks without requiring a scheduler
+ restart. We chose to use task_adoption_timeout to decide when a queued task is considered
+ stuck and should be rescheduled.
+ """
+ self.log.debug("Checking for stuck queued tasks")
+ max_allowed_time = utcnow() - self.task_adoption_timeout
+ queued_too_long = (
+ session.query(TaskInstance)
+ .join(TaskInstance.dag_run)
+ .filter(
+ TaskInstance.state == State.QUEUED,
+ TaskInstance.queued_by_job_id == self.job_id,
+ DagRun.state == State.RUNNING,
+ TaskInstance.queued_dttm < max_allowed_time,
+ )
+ .all()
+ )
+ # this filtering is done after the query rather than in the query because the query result
+ # set should always be quite small (a few rows at most), and self.tasks could be relatively
+ # large (hundreds or possibly thousands) which could make for a long db query in extreme cases
+ queued_too_long = [ti for ti in queued_too_long if ti.key in self.tasks]
+ if not queued_too_long:
+ return
+
+ try:
+ with timeout(seconds=15):
Review Comment:
No, the UI isn't complaining at all. As for the timeout I've never seen it - I only put that code in because you had it in the original PR here https://github.com/apache/airflow/pull/21556/commits/32d7060eb9927e14ada5f739d372a688538cf82b ....but I guess I did change the query to probably be a decent amount faster than the previous implementation, so I really doubt it's necessary anymore
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] dima-asana commented on pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
dima-asana commented on PR #23432:
URL: https://github.com/apache/airflow/pull/23432#issuecomment-1123083198
First, thank you so much for working on this. Weāve run into a very similar issue and are eagerly looking forward to a fix. We also now have an isolated environment where we can reproduce this, so Iād be happy to test some version of this patch if youād like (may have turnarounds of a week or so, sorry in advance). I went ahead and tested your current code with our setup and it works perfectly for us - the stuck task is revoked in Celery so there is no chance of it running twice and sent back to the scheduler.
> I'm far from a celery expert but as far as I can tell celery does not provide any capability to do this
Iām also not a celery expert, but I believe tasks lost due to a worker shutdown like https://github.com/celery/celery/issues/7266 have already gone from the broker queue to the celery worker consumer. This means that you would be able to find the tasks on celery rather than on the broker. You might try to look at `app.control.inspect().reserved` for tasks that got reserved but didnāt make it to a worker process. Unfortunately I donāt think itās possible to differentiate these tasks with ālegitimately reservedā tasks. That said, the default airflow configuration has a very short lifespan for legitimately reserved tasks ā with `prefetch_multiplier` as 1, `acks_late` as True, and the `fair` scheduling strategy, tasks are only reserved for workers that can āimmediatelyā consume them (in practice for us tasks are legitimately reserved for ~0.01s)
We experience this behavior due to a [different bug](https://github.com/celery/celery/issues/7515) in Celery. You can see a longer description of our experience [here](https://docs.google.com/document/d/1Peox1M7PUJQ4l1RR8DgWEDSyOMzWtbWDJnCQFpQIB54/edit?usp=sharing).
> "adopted task timeout" runs on every heartbeat
Iām either misunderstanding this or itās not accurate. At least, there is a `orphaned_tasks_check_interval` and in our experience it gets used. It could potentially be a simplification as you mentioned to unify the orphaned TI behavior - e.g. let it pick up both TIās orphaned by a dead scheduler (e.g. primarily tasks in `SCHEDULED` state) and TIās orphaned by a malfunctioning celery consumer (e.g. primarily tasks in `QUEUED` state).
As a sidenote, for tasks lost by shut down workers like https://github.com/celery/celery/issues/7266, you might give [this config](https://docs.celeryq.dev/en/stable/userguide/configuration.html#task-reject-on-worker-lost) in celery a shot.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] repl-chris commented on pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
repl-chris commented on PR #23432:
URL: https://github.com/apache/airflow/pull/23432#issuecomment-1125489780
FYI I have re-implemented this feature as an expansion of the existing "adopted task timeout" capability in https://github.com/apache/airflow/pull/23690. I believe the new implementation is superior as it is cleaner (IMHO), lighter-weight, and introduces minimal behavioural change unless it's explicitly enabled in the config.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [airflow] ashb commented on pull request #23432: Automatically reschedule stuck queued tasks in CeleryExecutor
Posted by GitBox <gi...@apache.org>.
ashb commented on PR #23432:
URL: https://github.com/apache/airflow/pull/23432#issuecomment-1129038298
Closing in favour of https://github.com/apache/airflow/pull/23690
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org