On Airflow with Kubernetes Executor, sometime the Task keeps running even though the executor pod has failed and later it is cleaned up by Scheduler as Daemon task #40979
Replies: 4 comments 16 replies
-
|
I am also seeing this issue as well, although with mine it has been mostly completed task that for some reason don't kill the pod and still show as "Running" in Kubernetes. I have also seen this on 2 Airflow versions (2.8.3 and 2.9.2). We are running on Kubernetes version 1.29. Could this be the cause? If anybody has solved this or has debugging tips that would be greatly appreciated! |
Beta Was this translation helpful? Give feedback.
-
|
Likely database connectibvity stability or possibly timeuts. It might be that the task fails to update status. Another option (and you can try it) is "schediule after task execution" configuration - you can set it to false and see if it helps. You can also take a look in your database locks for suspicious things (maybe your database runs out of resorces, or gets restarted from time to time or there is a tempoarary connectivity issue between pods and the DB. Converting it to a discussion as it is not a concrete "issue", but best things for you to do is to try to find some correlated events in other logs when it happens. |
Beta Was this translation helpful? Give feedback.
-
|
No idea actually. Any logs, anywhere correlated in time with erroneus behaviour that is happening. |
Beta Was this translation helpful? Give feedback.
-
|
Back to comment on my current situation. After looking at TONs of logs and messing with many settings (turning off remote logging, going to 1 scheduler, messing with pgbouncer metadataPoolSize and resultBackendPoolSize, messing with all the scheduler tuneable settings, messing with parallelism and parsing processes) my issues still persist.... I have turned on DEBUG for logs and have looked at logs that go into our Splunk instance, and this is what I can find: As you can see, by the logs the task looks like it marks as a success and returns with an exit code of 0. It then logs the logs to remote logging and disposes of the db connection pool: Once this has happened, the pod should complete and close, but it doesn't... Looking at the scheduler logs right after, this is what happens: And repeats like this over and over again and does not close the pod... So it seems like for some reason the "kubernetes_executor_utils" file is looking at the completed task and not seeing a "succeeded" state, even though the task logs that it is marking it as succeeded, and it also shows up as succeeded in the Airflow UI... Will try and keep digging but am really unsure where to look at this point. |
Beta Was this translation helpful? Give feedback.
-
|
Generally - look at the POD that gets into that state -> maybe there is a sidecar that keeps it running. Also try to disable "schedule_after_task_execution" - there is one extra step after task is completed - where it tries to schedule directly dependent tasks right after it finishes one. Maybe some of those tasks get into contention/wait timeout (there was a fix for that in 2.9.2 #39745 in case you are not on 2.9.2 yet - and maybe there are other reasons why it hangs - like database connectivity issues. Catching it "on the act" and possibly finding out what's going on by looking and pod status/logs in k8s (not in airlfow logs) and possibly exec-ing to the POD and seeing what processes are running still - and running pyspy on python processes to see where they hang - that what I'd likely attempt to do. |
Beta Was this translation helpful? Give feedback.
-
|
And this is why this is so tricky... Because it happens at random to random DAGs. It is never a specific DAG that happens every time. Plus after this happens, you cannot exec into the container... You get this error: So the container is "stopped" but for some reason it is still in a "Running" state. I don't have any sidecar containers running for my jobs, but I do have 2 init containers: git sync and our own custom one, which both finish and are in a completed state every time... Not really sure how I would check database connectivity. Never had a problem with it in the past but I guess it could have came up. |
Beta Was this translation helpful? Give feedback.
-
|
Also, the "AIRFLOW__SCHEDULER__SCHEDULE_AFTER_TASK_EXECUTION" to False did not work as we still saw about 5 or 6 last night |
Beta Was this translation helpful? Give feedback.
-
|
Also I have seen this issue mostly with dags which are resource intensive. For example, some dag is processing lot of data in memory and it exceeds the limit, it is killed by kubernetes and we can see OOMKilled status. Mostly this type of dags are ones which show this behaviour. Even after being killed, their status does not seem to update and the task continues to run. |
Beta Was this translation helpful? Give feedback.
-
|
Hey @potiuk , I have a DAG which is consistently generating the from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import timedelta
import logging
import sys
import time
def get_logger(logger_name = None):
formatter = logging.Formatter("%(asctime)s %(name)s %(levelname)s %(funcname)s:%(lineno)d %(message)s")
logger = logging.getLogger(logger_name) if logger_name else logging.getLogger()
logger.setLevel(logging.DEBUG)
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setFormatter(formatter)
logger.addHandler(console_handler)
return logger
class MainClass:
def __init__(self, args):
logger_name = None
for key, val in args.items():
setattr(self, key, val)
self.logger = get_logger(logger_name)
def some_class_method(self):
self.logger.info("Start testing dummy class ...")
time.sleep(100)
self.logger.info("End testing dummy class ...")
dag_args = {'retries': 0}
dag = DAG(
'cause_daemon_failure',
default_args = dag_args,
description = 'Fail dag',
schedule_interval = None,
tags = ['Daemon tester']
)
dag.catchup = False
def my_test_fnc(**kwargs):
print('Running test function')
print(kwargs)
main_obj = MainClass(args=kwargs)
main_obj.some_class_method()
test_my_fnc = PythonOperator(
task_id = 'test_my_fnc',
dag = dag,
python_callable = my_test_fnc,
op_kwargs = {
'my_var_1': 'Hello',
'send_mail': True
}
)
test_my_fncThis is a dumbed down version of an actual DAG which had lot of business logic. I was able to isolate the part which was causing failure. This dag does not fail due to OOM error. But it always ends with |
Beta Was this translation helpful? Give feedback.
-
|
Hi @potiuk , I also compared the Scheduler Job Runners in airflow In Which was replaced with |
Beta Was this translation helpful? Give feedback.
-
|
Well no OOM but you have This is caused by your fancy logger modifications. Airflow is doing dynamic modification of loggers for tasks and logging. When task is executed, it takes over stdout and stderr and redirects it back to the task logger (this is done in order to catch all the output that is printed to the stdout/stderr. By adding an extra handler in the dag you are likely messing with this mechanism and it causes infinite recursion - because whatever you write to stdout is redirected to StreamLogWriter which produces a log which is then printed to stdout handler of yours which prints it to stdout, which redirects it back to StreamLogWriter and so on. See this code: https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L381 I have no idea why you want to add your own log handler that redirects logs to stdout because we - purposefully redirect anything that is sent to stdout to the task log,. but this is definitely not something you should do. |
Beta Was this translation helpful? Give feedback.
-
|
I agree with you. Overriding the logger is a bad idea. I tried to run above DAG in Airflow 2.5.3 and it completes successfully. Though there are some warnings, but it completes.
Is it possible that |
Beta Was this translation helpful? Give feedback.
-
No idea. And no answer. And there is little value in spending others time on analysing that when we know that what you do causes problems. Change it and your problem is solved. |
Beta Was this translation helpful? Give feedback.
-
|
@ahipp13 Have you fixed the issue yet? Log in airflow-scheduler pod: |
Beta Was this translation helpful? Give feedback.
-
Apache Airflow version
Other Airflow 2 version (please specify below)
If "Other Airflow 2 version" selected, which one?
2.9.2
What happened?
Hello,
After upgrading to Airflow
2.9.2we have started seeing a strange issue with our Airflow tasks. Sometime if the task fails, the executor pod is deleted. But somehow Airflow does not get notified of the failure causing the task to continue running.After some time, scheduler finds it as a daemon task and kills the task. I don't know if this is airflow issue or Kubernetes Provider's issue. But there seem to be some drop in communication between these two.
But this has been occurring very frequently since
2.9.2upgrade.In some rare instances, I have also seen this behavior after successful completion of task. Though task complete successfully, Airflow is not notified of task state. Causing the task being killed later by Scheduler. And task status changes to failed.
What you think should happen instead?
When the task fails and pod is killed, Airflow should get notified appropriately. And task should reflect correct status.
How to reproduce
I am not sure how to re-produce it. This is not something happening everytime. It happens to some job. And on re-run the job completes correctly.
Operating System
Ubuntu Linux (on AKS)
Versions of Apache Airflow Providers
Deployment
Official Apache Airflow Helm Chart
Deployment details
We have deployed it AKS
Anything else?
No response
Are you willing to submit PR?
Code of Conduct
Beta Was this translation helpful? Give feedback.
All reactions