Handle kubernetes watcher stream disconnection

[ephraimbuddy]

Currently, when Kubernetes watch stream times out and we get error 410,
we just return resource version '0' which is not the latest version.

From the documentation, timing out is expected and we should handle it
by performing a list>watch>relist operation so we can continue watching
from the latest resource version. See https://kubernetes.io/docs/reference/using-api/api-concepts/#efficient-detection-of-changes

This PR follows the list>watch>relist pattern

Closes: #15418
This would likely fix some issues #14175, #13916, and #12644 (comment)

---

^ 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 UPDATING.md.

[github-actions]

The Workflow run is cancelling this PR. It has some failed jobs matching ^Pylint$,^Static checks,^Build docs$,^Spell check docs$,^Provider packages,^Checks: Helm tests$,^Test OpenAPI*.

[jedcunningham]

A few minor things, but otherwise LGTM.

[github-actions]

The Workflow run is cancelling this PR. Building images for the PR has failed. Follow the workflow link to check the reason.

[ashb]

So this change introduces a problem.

If we are asking to watch at resource version A (which is too old), and we do a list and get back version A+5 (say) then we have "lost" the events A+1..A+4, i.e. we would never call process_event() for the events in this range.

So we need to think if that is a problem or not

[jedcunningham]

we do a list and get back version A+5 (say) then we have "lost" the events A+1..A+4, i.e. we would never call process_event() for the events in this range.

Unfortunately, right now we have that same problem as we (attempt to) rewatch from "0", a.k.a any point in history. Per the docs, the latest is preferred so the behavior is likely pretty similar:
https://kubernetes.io/docs/reference/using-api/api-concepts/#the-resourceversion-parameter

I think the piece we are missing is a step to reconcile our state against what the list returns, e.g. detect we missed A+1..A+4 and action them. More work needs to be done to harden this area and this is the first step (and this alone doesn't buy us much other than setting us up for being more complete later).

@ephraimbuddy and I will spend some time trying to reproduce this scenario on demand to make sure we are starting down the right path.

[kaxil]

@ephraimbuddy Conflicts here

[ephraimbuddy]

@ephraimbuddy Conflicts here

Resolved.

[ephraimbuddy]

I came across using allow_watch_bookmarks https://kubernetes.io/docs/reference/using-api/api-concepts/#watch-bookmarks
to get the latest resource version but the python client have it disabled for now https://github.com/kubernetes-client/python-base/pull/234/files .
That would have solved this for us

[kaxil]

@ephraimbuddy Can you remember why we closed this PR -- do we have an alternate solution?

[ephraimbuddy]

@ephraimbuddy Can you remember why we closed this PR -- do we have an alternate solution?

I have forgotten why I closed this but it looks related to the above allow_watch_bookmarks that I found.

[JeremieDoctrine]

@kaxil did you find an alternate solution. We are also facing stream disconnection.

[ecerulm]

I came across using allow_watch_bookmarks https://kubernetes.io/docs/reference/using-api/api-concepts/#watch-bookmarks
to get the latest resource version but the python client have it disabled for now https://github.com/kubernetes-client/python-base/pull/234/files .
That would have solved this for us

Even with allow_watch_bookmarks it's not guarantee that the kubernetes api would send those bookmark events at all as stated in the link

As a client, you can request BOOKMARK events by setting the allowWatchBookmarks=true query parameter to a watch request, *but you shouldn't assume bookmarks are returned at any specific interval, nor can clients assume that the API server will send any BOOKMARK event even when requested.

So I think the solution is more in the way of #23521 where the last known resource_version is reset to 0 in the event of any exception.