Prevent KubernetesJobWatcher getting stuck on resource too old

[ecerulm]

Currently in airflow 2.3.0 the scheduler will get into an infinite loop if the k8s API ever response with a 410 (too old resource version / resource version too old).

This currently happen in airflow deployments in separate EKS 1.21 clusters, where every hour or so the scheduler will be stuck printing the error below in a tight loop, and will require manual restart.

Below is a log except from a scheduler in such state (tight loop of too old resource version error). As you can see the now my watch begins starting at resource_version: 380528372 repeats itself at 9:58:19,950, then one second later at 09:58:21,012, the again half second later at 09:58:21,541. Always asking k8s to start a watch from the same exact resourceVersion that will always return too old resource version

[2022-05-07 09:58:19,941] {kubernetes_executor.py:288} ERROR - Error while health checking kube watcher process. Process died for unknown reasons
[2022-05-07 09:58:19,950] {kubernetes_executor.py:126} INFO - Event: and now my watch begins starting at resource_version: 380528372
[2022-05-07 09:58:19,965] {kubernetes_executor.py:111} ERROR - Unknown error in KubernetesJobWatcher. Failing
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 102, in run
    self.resource_version = self._run(
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 145, in _run
    for event in list_worker_pods():
  File "/home/airflow/.local/lib/python3.8/site-packages/kubernetes/watch/watch.py", line 182, in stream
    raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 380528372 (382477719)
Process KubernetesJobWatcher-164543:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 102, in run
    self.resource_version = self._run(
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 145, in _run
    for event in list_worker_pods():
  File "/home/airflow/.local/lib/python3.8/site-packages/kubernetes/watch/watch.py", line 182, in stream
    raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 380528372 (382477719)
[2022-05-07 09:58:21,002] {kubernetes_executor.py:288} ERROR - Error while health checking kube watcher process. Process died for unknown reasons
[2022-05-07 09:58:21,012] {kubernetes_executor.py:126} INFO - Event: and now my watch begins starting at resource_version: 380528372
[2022-05-07 09:58:21,025] {kubernetes_executor.py:111} ERROR - Unknown error in KubernetesJobWatcher. Failing
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 102, in run
    self.resource_version = self._run(
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 145, in _run
    for event in list_worker_pods():
  File "/home/airflow/.local/lib/python3.8/site-packages/kubernetes/watch/watch.py", line 182, in stream
    raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 380528372 (382477719)
Process KubernetesJobWatcher-164544:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 102, in run
    self.resource_version = self._run(
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 145, in _run
    for event in list_worker_pods():
  File "/home/airflow/.local/lib/python3.8/site-packages/kubernetes/watch/watch.py", line 182, in stream
    raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 380528372 (382477719)
[2022-05-07 09:58:21,531] {kubernetes_executor.py:288} ERROR - Error while health checking kube watcher process. Process died for unknown reasons
[2022-05-07 09:58:21,541] {kubernetes_executor.py:126} INFO - Event: and now my watch begins starting at resource_version: 380528372
[2022-05-07 09:58:21,556] {kubernetes_executor.py:111} ERROR - Unknown error in KubernetesJobWatcher. Failing
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 102, in run
    self.resource_version = self._run(
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 145, in _run
    for event in list_worker_pods():
  File "/home/airflow/.local/lib/python3.8/site-packages/kubernetes/watch/watch.py", line 182, in stream
    raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 380528372 (382477719)
Process KubernetesJobWatcher-164545:
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/multiprocessing/process.py", line 315, in _bootstrap
    self.run()
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 102, in run
    self.resource_version = self._run(
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/executors/kubernetes_executor.py", line 145, in _run
    for event in list_worker_pods():
  File "/home/airflow/.local/lib/python3.8/site-packages/kubernetes/watch/watch.py", line 182, in stream
    raise client.rest.ApiException(
kubernetes.client.exceptions.ApiException: (410)
Reason: Expired: too old resource version: 380528372 (382477719)

The situation is as follows

• KubernetesJobWatcher will start a get/list/watch cycle starting at resource_version=0 using the kubernetes-client/python library kubernetes.watch.stream()
• The client will eventually get disconnected
• KubernetesJobWatcher will try to start a new get/list/watch cycle starting at what KubernetesJobWatcher thinks it's the latest resource version 380528372.
  • KubernetesJobWatcher tries hard to keep track of the last resource_version seen but there are inherent limitations on the k8s api (like the initial get/list does not guarantee ordering)
  • even if it did a perfect job tracking resource version the watch from a specific resource version is not guaranteed to work as stated in Kubernetes documentation > efficient detection of changes

A given Kubernetes server will only preserve a historical record of changes for a limited time. Clusters using etcd 3 preserve changes in the last 5 minutes by default. When the requested watch operations fail because the historical version of that resource is not available, clients must handle the case by recognizing the status code 410 Gone, clearing their local cache, performing a new get or list operation, and starting the watch from the resourceVersion that was returned. [ecerulm: using the resourceVersion that was returned in the get/list operation, not the one that KubernetesJobWatcher saved]

• This new attempt to watch from resourceVersion 380528372 will fail with a 410 resource too old (in my scenario this happens with high probability after the scheduler has been up for 1 hour) . Again this is expected non-exceptional case, there is no guarantee that you can restart the watch for resource version 380528372.
• When the watch operation fails immediately with kubernetes.client.exceptions.ApiException: (410) Reason: Expired: too old resource version: 380528372 (382477719) KubernetesJobWatcher will die
• Then a new KubernetesJobWatcher will start that tries again to start the watch from exactly the same resource version 380528372 and that will again fail for the same reason
• From this point on the scheduler will be stuck with no progress, will continue in a tight infinite loop of starting KubernetesJobWatcher, letting it do a watch that will always fails, noticing that KubernetesJobWatcher died again, starting it again, etc

This PR solves the situation (already tested for 10 hours in one of my airflow deployments in EKS) by always starting a fresh watch (resourceVersion=0) after KubernetesJobWatcher dies.

closes: #21087
related: #15500
related: #17629
related: #22407
related: #12644

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragement file, named {pr_number}.significant.rst, in newsfragments.

[ephraimbuddy]

You may want to look at this comment #15500 (comment)

[ncapeta]

Any possibility to get a release with this ?

[potiuk]

Any possibility to get a release with this ?

I am about to prepare next release of providers very soon. If it gets merged before that it will go to that release - if not, then it will have to wait for the next one.

[eladkal]

I am about to prepare next release of providers very soon. If it gets merged before that it will go to that release - if not, then it will have to wait for the next one.

The files modified by this PR are of Airflow core so it doesn't matter for provider release :)
I'll mark it for 2.3.1

[potiuk]

The files modified by this PR are of Airflow core so it doesn't matter for provider release :)

Ah. Right. The Provider vs Core Kubernetes strikes back :)

[ncapeta]

Tks, this is making in some instances 2.3 not usable at all as the schedulers after 1h need to be deleted

[potiuk]

Tks, this is making in some instances 2.3 not usable at all as the schedulers after 1h need to be deleted

Yeah. 2.3.1 with some teething problems og 2.3.0 solved is coming soon.

[ncapeta]

any potential release date ?

[potiuk]

any potential release date ?

Whatever we tell you we might not hold to because we might not be able to find/fix all important issues by then. And if you base any "business" decisions of yours on what we tell, then you will have complaints. So better not to give you false hopes for OSS project.

Just watch how many issues are left here (some might get added though still):

https://github.com/apache/airflow/milestone/54

However being an OSS project it has an advantage, that you can just take the change and apply it yourself in your environment while waiting. It's a few line change that you can easily apply on installed Airflow - just find the installed sources and update them in the way that they will persist in your environment. This way you can handle your "business" and you cannot tell "I am stuck becasue they have not released yet"..

[ashb]

Does this change mean that we potentially miss some events?

i.e. if the version n is too old, and we reset it to 0, and the next event is m, then any pods state changes in n to m won't ever be seen?

[ecerulm]

Yes, you potentially miss old event just like before. The only difference is that the scheduler does not get into an infinite loop because of it.

If airflow gets a 410 from kubernetes api that already is a pretty good indicator that some event has been lost but there is nothing airflow can do about it at that point, right?

the question remains if kubernetes offers the right API with the right guarantees to implement this kind of scheduling reliably. Or if other approach would be better. This PR was only about preventing the scheduler from entering an infinite loop in which progress was impossible.