Copy of [AIRFLOW-5071] JIRA: Thousands of Executor reports task instance X finished (success) although the task says its queued. Was the task killed externally? #10790
Comments
|
Thanks for opening your first issue here! Be sure to follow the issue template! |
|
Thanks @dmariassy for bringing this issue to Github! I think this one is quite important to fix but as long as we don't know how to replicate it we are going blind. I spent some time trying to reproduce it on 2.0 and 1.10.9 but to no effect :< |
turbaszek
added
area:Scheduler
|
Thanks for your reply @turbaszek . What did your reproduction set-up look like? If I have the time, I would like to have a go at trying to reproduce it myself in the coming weeks. |
|
As it was reported in original issue and comments this behavior should be possible to reproduce in case of fast sensors in reschedule mode. That's why I was trying to use many DAGs like this: from random import choice
from airflow.utils.dates import days_ago
from airflow.sensors.base_sensor_operator import BaseSensorOperator
from airflow.operators.bash_operator import BashOperator
from airflow import DAG
import time
class TestSensor(BaseSensorOperator):
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.mode = "RESCHEDULE"
def poke(self, context):
time.sleep(5)
return choice([True, False, False])
args = {"owner": "airflow", "start_date": days_ago(1)}
with DAG(
dag_id="%s",
is_paused_upon_creation=False,
max_active_runs=100,
default_args=args,
schedule_interval="0 * * * *",
) as dag:
start = BashOperator(task_id="start", bash_command="echo 42")
end = BashOperator(task_id="end", bash_command="echo 42")
for i in range(3):
next = TestSensor(task_id=f"next_{i}")
start >> next >> endAnd I was also playing with airflow config settings as described in comments. Although I saw failing tasks there was no issue like this one or... eventually the log was missing? I did some tests with external task sensor but also no results. |
|
Hi @turbaszek, any finding on this? We have a CeleryExecutor Redis setup with three workers (apache-airflow 1.10.12). The airflow-scheduler log has a lot of lines like this. I remember this was already a problem when we were using older versions such as 1.10.10. It's just we never paid much attention to it. Same with others in this thread, we have a lot of sensors in "reschedule" mode with I also tried to tweak these parameters. They don't seem to matter much as far as this error is concerned: The way to reproduce this issue seems to be to create a DAG with a bunch of parallel When the scheduler starts to process this DAG, we then start to see the above error happening to these sensors. And the go into import datetime
import pendulum
import time
from airflow.models.dag import DAG
from airflow.contrib.sensors.python_sensor import PythonSensor
with DAG(
dag_id="test_dag_slow",
start_date=datetime.datetime(2020, 9, 8),
schedule_interval="@daily",
) as dag:
sensors = [
PythonSensor(
task_id=f"sensor_{i}",
python_callable=lambda: False,
mode="reschedule",
retries=2,
) for i in range(20)
]
time.sleep(30) |
|
@yuqian90 thanks you so much for pointing to the DAG! I will check it and let you know. Once we can replicate the problem it will be much more easier to solve it 👍 |
Those parameters won't help you much. I was struggling to somehow workaround this issue and I believe I've found the right solution now. In my case the biggest hint while debugging was not scheduler/worker logs but the Celery Flower Web UI. We have a setup of 3 Celery workers, 4 CPU each. It often happened that Celery was running 8 or more python reschedule sensors on one worker but 0 on the others and that was the exact time when sensors started to fail. There are two Celery settings that are responsible for this behavior: I've been trying a lot to setup a local docker compose file with scheduler, webserver, flower, postgres and RabbitMQ as a Celery backend but I was not able to replicate the issue as well. I tried to start a worker container with limited CPU to somehow imitate this situation, but I failed. There are in fact tasks killed and shown as failed in Celery Flower, but not with the "killed externally" reason. |
|
@sgrzemski-ias I will setup an environment to first observe the behavior and then if it will occur I will check your suggestion! Hope that we will be able to understand what's going on here 🚀 |
|
Ok @yuqian90 @sgrzemski-ias what is you setting for core.dagbag_import_timeout ? As I'm hitting: Traceback (most recent call last): File "/usr/local/lib/airflow/airflow/models/dagbag.py", line 237, in process_file m = imp.load_source(mod_name, filepath) File "/opt/python3.6/lib/python3.6/imp.py", line 172, in load_source module = _load(spec) File "", line 684, in _load File "", line 665, in _load_unlocked File "", line 678, in exec_module File "", line 219, in _call_with_frames_removed File "/home/airflow/gcs/dags/test_dag_1.py", line 24, in time.sleep(30) File "/usr/local/lib/airflow/airflow/utils/timeout.py", line 43, in handle_timeout raise AirflowTaskTimeout(self.error_message) airflow.exceptions.AirflowTaskTimeout: Timeout, PID: 6217 |
Hi, @turbaszek in my case I have After digging further, I think the slowness that causes the error for our case is in this function: |
|
Here's another potential hint: We have increased the |
|
I can confirm that one of our customers also faced a similar issue with poke='reschedule' and increasing poke_interval had fixed the issue for them. It feels some sort of race condition. |
|
We are on Airflow 1.10.10 Besides the DAGs which have sensor tasks in them, we are even encountering this in tasks which have no sensors in them, for example a DAG which only has PythonOperator and HiveOperator in it. |
Some further investigation shows that the slow down that caused this issue for our case (Airflow 1.10.12) was in |
|
We have just introduced ExternalTaskSensor into our pipeline and faced the same issue. When initially tested on our dev instance (~200 DAGs) it worked fine, after running it on our prod environment (~400 DAGs) it was always failing after reschedule. After digging into the code, it looks that this is simply race condition in the scheduler. We have child_dag.parent_dag_completed task that waits for business process to complete calculations in parent_dag, task execution logs: Scheduler logs: From scheduler log it's visible that event from executor is processed after task is already queued for the second time. Logic related to those logs is here: def _validate_and_run_task_instances(self, simple_dag_bag):
if len(simple_dag_bag.simple_dags) > 0:
try:
self._process_and_execute_tasks(simple_dag_bag) # <-- task state is changed to queued here
except Exception as e:
self.log.error("Error queuing tasks")
self.log.exception(e)
return False
# Call heartbeats
self.log.debug("Heartbeating the executor")
self.executor.heartbeat()
self._change_state_for_tasks_failed_to_execute()
# Process events from the executor
self._process_executor_events(simple_dag_bag) # <-- notification of previous execution is processed and there is state mismatch
return TrueThis is the place where task state is changes: def _process_executor_events(self, simple_dag_bag, session=None):
# ...
if ti.try_number == try_number and ti.state == State.QUEUED:
msg = ("Executor reports task instance {} finished ({}) "
"although the task says its {}. Was the task "
"killed externally?".format(ti, state, ti.state))
Stats.incr('scheduler.tasks.killed_externally')
self.log.error(msg)
try:
simple_dag = simple_dag_bag.get_dag(dag_id)
dagbag = models.DagBag(simple_dag.full_filepath)
dag = dagbag.get_dag(dag_id)
ti.task = dag.get_task(task_id)
ti.handle_failure(msg)
except Exception:
self.log.error("Cannot load the dag bag to handle failure for %s"
". Setting task to FAILED without callbacks or "
"retries. Do you have enough resources?", ti)
ti.state = State.FAILED
session.merge(ti)
session.commit()Unfortunately I think that moving _process_executor_events before _process_and_execute_tasks would not solve the issue as event might arrive from executor while _process_and_execute_tasks is executing. Increasing poke_interval reduces chance of this race condition happening when scheduler is under a heavy load. I'm not too familiar with Airflow code base, but it seems that the root cause is the way how reschedule works and the fact that try_number is not changing. Because of that scheduler thinks that event for past execution is for the ongoing one. |
|
The cause is clear as @rafalkozik mentioned. After scheduler schedule the task at the second time(put it in queue) and then it start process the executor events of the task's first-try. It occurs when the scheduling loop time > sensor task reschedule interval. The bug can also be fixed if the rescheduled task instance use a different try number, but this will cause a lot of log files. |
I saw customers doing this (custom fork). I'm curious if this error will occur in Airflow 2.0 |
Hi @turbaszek I did not test this in Airflow 2.0 so I may be wrong. I don't see any attempts to address this in Airflow 2.0 so this is likely going to happen in 2.0 too. That said, the scheduler loop is faster in Airflow 2.0, the chance of running into this |
|
@turbaszek I am currently testing Airflow v2.0.0b3 against the same DAGS we currently run on production against 1.10.12 and I can confirm that this is still an issue. Combined with #12552 it makes the problem even worse too. |
|
To add some further context, I can consistently replicate this error on 2.0.0b3 on a very simple environment running two Docker containers - webserver and postgres - on a Python 3.7 image using LocalExecutor and with a |
|
Not sure if this is relevant but, when the task was rescheduled five minutes later, I saw this. |
I saw this also from time to time but not always so probably not related. |
|
@nathadfield @yuqian90 and others, have you been able to test 2.0? Have you observed this issue? |
|
I found that in the code of if ti.try_number == buffer_key.try_number and ti.state == State.QUEUED:
Stats.incr('scheduler.tasks.killed_externally')
msg = (
"Executor reports task instance %s finished (%s) although the "
"task says its %s. (Info: %s) Was the task killed externally?"
)
self.log.error(msg, ti, state, ti.state, info)The scheduler checks the state of the task instance. When a task instance is rescheduled (e.g: an external sensor), its state transition up_for_reschedule -> scheduled -> queued -> running. If its state is queued and not moved to the running state, the scheduler will raise an error. if ti.try_number == buffer_key.try_number and (
ti.state == State.QUEUED and not TaskReschedule.find_for_task_instance(ti, session=session)
):
Stats.incr('scheduler.tasks.killed_externally')
msg = (
"Executor reports task instance %s finished (%s) although the "
"task says its %s. (Info: %s) Was the task killed externally?"
)
self.log.error(msg, ti, state, ti.state, info)Here is my PR: #19123 |
|
we reviewed the code and found that in However, theses lines may cover a bug?
The raw task command write back the taskintance's state(like sucess) doesn't mean the child process is finished(returned)? So, in this heatbeat callback, there maybe a race condition when task state is filled back while the child process is not returned. In this senario, the local task will kill the child process by mistake. And then, the scheduler will checkout this and report "task instance X finished (success) although the task says its queued. Was the task killed externally?" this is a simple schematic diagram:
|
|
We face the same issue with tasks that stay indefinitely in a queued status, except that we don't see tasks as Example logs: Worker: Because of the MWAA autoscaling mechanism, We currently have 2 (minWorkers) to 10 (maxWorkers) mw1.medium (2 vCPU) workers. |
|
We also run into this fairly often, despite not using any sensors. We only seemed to start getting this error once we changed our Airflow database to be in the cloud (AWS RDB); our Airflow webserver & scheduler runs on desktop workstations on-premises. As others have suggested in this thread, this is a very annoying problem that requires manual intervention. @ghostbody any progress on determining if that's the correct root cause? |
|
@pbotros No, we do not solve this problem yet. 😢 |
|
The problem for us was that we had one dag that reach 32 parallelize runnable task ( 32 leaf tasks) which was the value of parameter |
|
After STRUGLING, We found a method to 100% reproduce this issue !!! tl;dr airflow/airflow/models/taskinstance.py Line 1253 in 9ac7428 Add a Then you will get this issue Conditions:
It's becasue the worker use a local task job which will spwan a child process to execute the job. The parent process set the task from related code is here: https://github.com/apache/airflow/blob/2.2.2/airflow/jobs/local_task_job.py#L89 |
|
@ghostbody do you have idea how this can be addressed? |
|
@turbaszek Let me make a PR later~ We are doing pressure tests these days and this problem had appeared often. |
@val2k Did you find a solution for this ? I am also using MWAA environment and facing the same issue. The tasks get stuck in queued state and when I look at the scheduler logs I can see the same error. "Executor reports task instance %s finished (%s) although the task says its %s. (Info: %s) Was the task killed externally?" I tried everything I can find in this thread but nothing seems to be working. |
|
We also got the same error message. In our case, it turns out that we are using the same name for different dags. |
|
airflow: 2.2.2 with mysql8、 HA scheduler、celery executor(redis backend) From logs, it show that those ti reported this error
From mysql we get that: all failed task has no external_executor_id! We use 5000 dags, each with 50 dummy task, found that, if the following two conditions are met,the probability of triggering this problem will highly increase:
We do these tests:
I read the notes below , but still don't understand this problems:
|
Hey turbaszek, Any chance to have PR submitted, we are experiencing in 2.3.0 as well. |
I think you wanted to call @ghostbody who wanted to submi the fix @vanducng . |
|
Hello here 👋, We are running the architecture with a shared NFS which is hosting our dags and logs (among other things). The speed and allowed throughput in the shared fileSystem is huge bottleneck for the scheduler (since it needs to parse the dags quite often). We noticed that the issue with the sensors and the log message [EDIT] I found the reason, see comment below.
One could argue, why am I making so much trouble if the task has been successfully run ? Because sometime, the task simply doesn't ever finish, and stay in running state indefinitely, which is hard to spot (until we have some alert because the dag has a lot of delay)
And then no log given by the worker, saying for example that the task went from
(This is a very long task basically waiting for a spark job to termiate) Here, it ended well, but sometime, there is just nothing happening. How is it possible that no logs from the first worker were generated? How is it possible that the scheduler scheduled the task a second time if it was still in running state 🤔 I am still trying to understand the complexity of Airflow to understand this issue and maybe propose a PR, but I wanted to participate with what I have been able to find so far. Some more informations about our architectures: Environment: Cloud provider or hardware configuration: AWS |
|
Sorry for the bothering 👆, I found the reason. |
|
In the case where a task is a visible again in Celery broker because I am just wondering if this is a wanted behavior from Airflow. Isn't the worker supposed to send heartbeat to the db, to tell it that it is still running? Why would we want a second worker to pick the task again after |
|
Hello, We are facing a similar issue but it looks likes for me a combination of EFS(Throughput-Provisioned(25MiB/s) Worker not sending back an exception to Scheduler that unable to read the dag file and task struck in queue state forever. Scheduler Log Worker Log
|
|
Hello @karthik-raparthi, we also did experience similar issue with EFS. EFS is definitely not suited for a big airflow deployment, and we stopped having most of its issues when we moved to FSx File System. I therefore encourage you to move to this better solution :) (we had EFS 100 Mo/s provisioned throughout and still experiencing this) |
Quite agree. There were multiple people reporting problems in huge airflow installation where EFS was used. I can also recommend (as usual) switching to Git Sync. I wrote an article about it https://medium.com/apache-airflow/shared-volumes-in-airflow-the-good-the-bad-and-the-ugly-22e9f681afca - especially when you are using Git to store your DAGs already, using shared volume is completely unnecessary and using Git Sync directly is far better solution. |
|
Thanks, @V0lantis & @potiuk for the inputs. Yes, we are in process of moving away from EFS. but trying to see any workarounds to find the issue once it struck in Queue by using some alerts. I did some research and looks like we can rely on the task_instance table on Airflow metadb to alert as soon as a Task struck in Q for more than 30 mins(this time might vary based on EFS) `select |
|
The simplest workaround is to pay EVEN MORE for OUTRAGEOUS amount of EFS IOPS. This seemed to work for many of our customers and it might be cheaper than engineering time you spend on trying to solve it (unless you are already maxing it out). |
|
This issue has been automatically marked as stale because it has been open for 365 days without any activity. There has been several Airflow releases since last activity on this issue. Kindly asking to recheck the report against latest Airflow version and let us know if the issue is reproducible. The issue will be closed in next 30 days if no further activity occurs from the issue author. |
|
This issue has been closed because it has not received response from the issue author. |
and others
Apache Airflow version: 1.10.9
Kubernetes version (if you are using kubernetes) (use
kubectl version): Server: v1.10.13, Client: v1.17.0Environment:
uname -a):Linux airflow-web-54fc4fb694-ftkp5 4.19.123-coreos #1 SMP Fri May 22 19:21:11 -00 2020 x86_64 GNU/LinuxWhat happened:
In line with the guidelines laid out in AIRFLOW-7120, I'm copying over a JIRA for a bug that has significant negative impact on our pipeline SLAs. The original ticket is AIRFLOW-5071 which has a lot of details from various users who use ExternalTaskSensors in reschedule mode and see their tasks going through the following unexpected state transitions:
running -> up_for_reschedule -> scheduled -> queued -> up_for_retry
In our case, this issue seems to affect approximately ~2000 tasks per day.
What you expected to happen:
I would expect that tasks would go through the following state transitions instead: running -> up_for_reschedule -> scheduled -> queued -> running
How to reproduce it:
Unfortunately, I don't have configuration available that could be used to easily reproduce the issue at the moment. However, based on the thread in AIRFLOW-5071, the problem seems to arise in deployments that use a large number of sensors in reschedule mode.
The text was updated successfully, but these errors were encountered: