Scheduler resetting Kubernetes tasks as orphaned on startup

[hterik]

Apache Airflow version

2.3.2

What happened

Scheduler was restarted.
After this it starts resetting some running tasks as orphaned.

I have seen #20982 which lists this as known issue for manually started tasks, but we also see it occasionally for scheduled tasks.

This issue appears to be a regression after 2.3 upgrade, i don't recall ever seeing it in 2.2, now we experience it almost every time scheduler restarts. Which happen almost once per day due to crashes caused by connection flakiness to the Kubernetes API or PGSQL.

What you think should happen instead

Tasks should be adopted

How to reproduce

Start some tasks running on Kubernetes with KubernetesCeleryExecutor.
Restart scheduler

Scheduler logs show following:

2022-07-13 09:59:19 {scheduler_job.py:353} INFO tasks up for execution:
		<TaskInstance: XXXXX scheduled__2022-07-12T16:00:00+00:00 [scheduled]>
	
2022-07-13 09:59:19 {scheduler_job.py:504} INFO Setting the following tasks to queued state:
	<TaskInstance: XXXXX scheduled__2022-07-12T16:00:00+00:00 [scheduled]>
2022-07-13 09:59:20  {scheduler_job.py:633} INFO Setting external_id for <TaskInstance: XXXXX scheduled__2022-07-12T16:00:00+00:00 [queued]> to 38321

...
Scheduler crashes and restarts here
...

2022-07-13 10:42:59	{scheduler_job.py:1285} Reset the following 8 orphaned TaskInstances:
    <TaskInstance: XXXXX scheduled__2022-07-12T16:00:00+00:00 [running]>
    ....
2022-07-13 10:43:00	{scheduler_job.py:353} Level=INFO Message=10 tasks up for execution:
    <TaskInstance: XXXXX scheduled__2022-07-12T16:00:00+00:00 [scheduled]>
    ....
....
2022-07-13 10:43:00	{scheduler_job.py:504} INFO Setting the following tasks to queued state:
	<TaskInstance: XXXXX scheduled__2022-07-12T16:00:00+00:00 [scheduled]>
    ...

I don't know what other logs that might be relevant.

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

apache-airflow==2.3.2
apache-airflow-client==2.1.0
apache-airflow-providers-celery==3.0.0
apache-airflow-providers-cncf-kubernetes==4.0.2
apache-airflow-providers-docker==3.0.0
apache-airflow-providers-ftp==2.1.2
apache-airflow-providers-http==2.1.2
apache-airflow-providers-imap==2.2.3
apache-airflow-providers-postgres==5.0.0
apache-airflow-providers-sqlite==2.1.3

Deployment

Other Docker-based deployment

Deployment details

Pgsql as database

Anything else

Almost every time scheduler restarts

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

[potiuk]

Is this happening for all tasks that are running when scheduler was restarted or only some of them ? Any more information /guesses of which tasks might be affected ? Also what kind of deployment you have - Kubernetes Executor? Kubernetes Celery Executor? Can you please elaborate more on that?

BTW. Why do you have such frequent restarts of PGSQL/Kubnernetes API ? this seems like something you should address in general. This is very likely that the problem is not because if you have flakiness in DB and Kubernetes APIs, there is no way Airflow implementation can cover it up. Airlfow uses PRECISELY DB and Kubernetes APIs to store information and query for it so that it is able to recover from any kind of crashes, so if your flakiness is there in either of those, it might basically mean that Airlfow will not be able to recover (because the flakiness prevented it from storing the information necessary or querying it).

This is rather unsolvable conundrum - we actually rely on stable deployment underneath. And I believe it should be handled first.

[hterik]

Using CeleryKubernetesExecutor. It appears to be random, doesn't happen to all tasks on every scheduler restart. Only happens to Kuberenetes tasks, not Celery tasks. Don't have more details sorry, will try to collect as it happens more.

---

The connection issues are unfortunately something we inherit from Azure, we use both their hosted pgsql and AKS, some times they just decide to not respond for up to half a minute. I understand at some point Airflow has to rely on the API underneath, but i think a client of a any distributed system need to have retries to suppress such minor connectivity glitches or restarts of the upstream service, there just isn't such a thing as a stable deployment. See also https://www.citusdata.com/blog/2021/02/12/reconnecting-your-app-after-a-postgres-failover/ for expansion of this philosophy.

The bottom line is that High Availability for Postgres cannot be achieved on the server-side only. Even with the best DBAs and SREs and the most robust deployment available on the planet, when a Postgres failover happens, your application code must be able to handle the ensuing “connection lost” situation.

I filed #24748 for example of some concrete improvements that can be done in this area when it comes to Kubernetes.

---

Even if not ideal, the restarts themselves are something we could live with (if not happening all too frequently), i consider that a separate issue. If only the scheduler picked up tasks that were running earlier. Now there can be tasks that run for hours, then the scheduler restarts for a different reason (not when failing to store the status of the task in question) and then it resets that task which was 90% done and it has to rerun for hours again.

[potiuk]

Well. While agree that sometimes handling flaky connections is necessary, it is the choice of the client at which layer you do it:

1. TCP performs retrrying and keeping persistent connection even in the event of network error is built in TCP protocol (that helps with handling networking issues). This is the main reason why we are not using UDP

2. reconnecting to a server when it stops responding without disturbing the client (including failover to another server)- here various proxies help and in case of Postgres, the solution is local PgBouncer -which we recommend for every Airlfow Postgres installation. If you do not have it, I heartily recommend trying it- it is actually likely to solve many of the problems if you have server stability issues.

3. Retrying a failed transaction when it gets broken.This is something that is VERY WELL described in SQL Alchemy docs. I recommend you go and read it. There is a part about pre.pinging (but pgBouncer handles it better). But also you can read about all the possibilities you have to recycle connections (usually those are create_engine args). If you indeed are stuck with "unstable" database, then likely some of those settings might help you:

https://docs.sqlalchemy.org/en/14/core/pooling.html#dealing-with-disconnects

But this is your deployment and your connection stability and you have to apply that knowiledge and sqlalchemy parameters to handle that. Airflow nor maintainers of airlfow will not do the job. Copying the documentation from SQL Alchemy to Airlfow make no sense - so if you really have connection stability issues, this is the right level you have to do it on your own.

4. The really interesting part is the one that mention why any kind of "automated retries" are not easy and that application should really handle it because the Database state might change during such a failed transaction.

https://docs.sqlalchemy.org/en/14/faq/connections.html#how-do-i-retry-a-statement-execution-automatically

And there inded we could do something in Airflow. But it is terribly complex to handle individual DB operations. Generic solutions like decorators handling retry or reacting to engine disconnect are a bad choice (As explained in the SQL Alchemy docs). They acknowldge it can be done, but they finally conclude (and I agree with that) that the only good approach is individiually restart full transactions and implement the logic individually for each case. And this means - complicating our code a LOT. Basically you would need to analyse every single place where your transaction can break and apply correct logic on how to retry it. This is super complex. And we chose not to do it for ALL transactions. We do it in some individual cases when we see deadlocks happening (especially on MySQL), But implementing a "generic" solution means that we would have to not only review it but also carefuly test all such cases and make sure it can flawlessly retry every singlie transaction. Probably in 2 years we might have it completed. Without making any other change to Airflow.

So instead we chose another option:

5. Allow the user to configure retries at task level. This is very easy to implement and it will always work (providing that Airlfow Operators stick to the "idempotency" assumption (which is the core of Airflow operator's assumptions. We KNOW we can retry every task and it's ok. We know it will have no side effects (except delays and possibly some costs involved). And we give the users a powerful tool to decide on the posible number of retries that can be accepted (and even give the opportunity to give up early with AirlfowFailureException. And we went this path. We do have retries on application level, but we deliberately chose to handle retries at task levels rather than at transaction level becuase it is the only maintainable solution in short, medium and even long term. Yes, it means that sometimes the task will be excute twice or three times. It also means that the user has the responsibility in definig the retry - but with default_args it can be set to retry by defaults. And yes - it's not perfect. But it works. And if someone wants to make the "transaction" retries work, then yeah - it is possible, but the question is who will implement and maintain that? The tradeoff here is - to either have someone (who?) to sponsor huge effort of making transactional retries or to put a bit more burden on our users to work on stability of their databases, connections, sqlalchemy engine parameters and finally task definition in case they cannot handle the former ways.

And finally even if all of that fails - and you have an occasioanl failure, then yes Airlfow is eventualle a solution that will not handle all errrors - ocassionally humans looking at it will have to manully retry something. This is the whole reason we have the UI. Otherwise - if we had a "perfect" scheduler, that would be able to handle all the connectivity and other problems we would not need the UI at all.

And to conclude. if you really have the case where connectiviity problems of your DB are problematic for you to the level that your human operators have to continuously monitor and restart things, then your connectivity is a problem not Airlfow. You need to fix it. Change the DB. Change cloud provider. Pay extra for more stable connectivity and DB. Delegating this task to Airlfow is simply unfair game.

[potiuk]

Converted it into discussion until more information is received/flakiness reasons investigated to get better understanding of what happens by the user.

[hterik]

I've been debugging this for a while now and think there is something about this query inside scheduler_job.adopt_or_reset_orphaned_tasks

airflow/airflow/jobs/scheduler_job.py

Lines 1396 to 1411 in 4261210

	query = (
	session.query(TI)
	.filter(TI.state.in_(resettable_states))
	# outerjoin is because we didn't use to have queued_by_job
	# set, so we need to pick up anything pre upgrade. This (and the
	# "or queued_by_job_id IS NONE") can go as soon as scheduler HA is
	# released.
	.outerjoin(TI.queued_by_job)
	.filter(or_(TI.queued_by_job_id.is_(None), SchedulerJob.state != State.RUNNING))
	.join(TI.dag_run)
	.filter(
	DagRun.run_type != DagRunType.BACKFILL_JOB,
	DagRun.state == State.RUNNING,
	)
	.options(load_only(TI.dag_id, TI.task_id, TI.run_id))
	)

Particularly the SchedulerJob.state != State.RUNNING part.
I think that condition should only be used on the periodic calls of adopt_or_reset_orphaned_tasks(). But first time on scheduler startup, it should pick up everything, including things that were previously in running state?

If i restart the scheduler and let it kill of some orphan tasks, then run the query above manually, with that condition commented out, it gives me exactly the same list of tasks as the scheduler just reset.

[potiuk]

Is it possible to get a reproducible test case and provide a PR for that ?

[hterik]

After further debugging i think SchedulerJob.state != State.RUNNING is probably not the issue here.

What happens is following: Using CeleryKubernetesExecutor

1. Start scheduler, this creates a job in DB of type=SchedulerJob
2. Start a DAG with tasks intended for Kubernetes. This creates a TI, with queue=kubernetes and queued_by_job_id=1
3. Kubernetes picks up the pod and runs airflow tasks run ... inside. This does 2 things:
   3a) It inserts a new job entry in the DB with job_type=LocalTaskJob
   3b). It changes the TI.queue of the existing TI instance from kubernetes to default ⚠️ .
4. Stop scheduler.
5. Wait scheduler_health_check_threshold seconds
6. Start scheduler
7. SchedulerJob.adopt_or_reset_orphaned_tasks resets the existing SchedulerJob that was running and sets it into failed state
8. Scheduler finds all TIs that have a queued_by_job that is not running (since it was reset in step 7 above it will find all of them, this is OK).
   (Side note: I am also a bit suspicious that this query should contain SchedulerJob.job_type == "SchedulerJob" but not 100% certain, it probably still doesn't solve this particular issue, similar to what was added for the query right above in Only mark SchedulerJobs as failed, not any jobs #19375 , ping @jedcunningham )
9. Inside the CeleryKubernetesExecutor.try_adopt_task_instances, it sends all TIs with queue=kubernetes to the KubernetesExectuor and the rest to the CeleryExecutor for adoption. But - Since Step 3b cleared the queue attribute to default, even kubernetes tasks get sent to the CeleryExecutor, that fails adopting them, for natural reasons. ⚠️
   

   airflow/airflow/executors/celery_kubernetes_executor.py

   Lines 172 to 177 in a2b186a

   	celery_tis = [ti for ti in tis if ti.queue != self.KUBERNETES_QUEUE]
   	kubernetes_tis = [ti for ti in tis if ti.queue == self.KUBERNETES_QUEUE]
   	return [
   	*self.celery_executor.try_adopt_task_instances(celery_tis),
   	*self.kubernetes_executor.try_adopt_task_instances(kubernetes_tis),
   	]

---

I can try to create a unit test for this, it's quite many moving parts and integrations and I'm not even entirely sure where step 3 happens or if it's correct. Solving this will be even more of a challenge for me.
Stupid question first: Are these LocalTaskJob intended to be inserted into the DB, or have i accidentally configured my deployment in a strange way?

[hterik]

On closer look i think the cause of 3b is that we had bad configuration. The pod_template_file launching the Workers in Kubernetes did not have all the airflow.cfg in sync with what the Scheduler was using. In particular the AIRFLOW__OPERATORS__DEFAULT_QUEUE was deviating, in scheduler it was set to kubernetes, whereas in workers it was unset.

With this changed my local repro of the issue is gone. Will deploy this and keep observing if it solves our real deployment. Hopefully this is case closed.