occasional KeyError: 'resourceVersion' during a long running list+watch

[asteven]

I have long running processes using the typical ListWatch pattern.

Occasionally I get this Error:

            |   File "lightkube/core/generic_client.py", line 425, in watch
            |     yield wd.process_one_line(line)
            |           ^^^^^^^^^^^^^^^^^^^^^^^^^
            |   File "lightkube/core/generic_client.py", line 86, in process_one_line
            |     self._version = obj["metadata"]["resourceVersion"]
            |                     ~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^
            | KeyError: 'resourceVersion'
            +------------------------------------

Line numbers are slightly off as I added some debug information before.

The problem is that the API server responds to this request:

BasicRequest(method='GET',
             url='api/v1/namespaces/default/pods',
             response_type=<class 'lightkube.resources.core_v1.Pod'>,
             params={'resourceVersion': '464635228', 'watch': 'true'},
             data=None,
             headers=None)

With this response line:

{'object': {'apiVersion': 'v1',
            'code': 410,
            'kind': 'Status',
            'message': 'too old resource version: 464635228 (464639593)',
            'metadata': {},
            'reason': 'Expired',
            'status': 'Failure'},
 'type': 'ERROR'}

which results in this obj dict:

{'apiVersion': 'v1',
 'code': 410,
 'kind': 'Status',
 'message': 'too old resource version: 464635228 (464639593)',
 'metadata': {},
 'reason': 'Expired',
 'status': 'Failure'}

which does not have the 'resourceVersion' key.

I guess lightkube should handle the Status object here explicitly and raise an ApiError instead of letting the KeyError through?
Other ideas or requirements on how to address this?
Are you willing to accept a patch to fix that?

[gtsystem]

Yes, it would be great if you can create a PR. The watch API, supports the on_error parameter. For the cases OnErrorAction.RAISE and OnErrorAction.STOP are requested, the expectation is clear. I'm not so sure what would be the right thing to do in case of OnErrorAction.RETRY. In theory we should attempt a retry, maybe resetting the resourceVersion to None, but I'm not sure if this could be surprising as it will yield changes already handled in the past.. I guess we could document this behavior? Is there a better solution?

[asteven]

Ah, ok. I was not aware of the on_error parameter.
Given that the caller can influence what happens when a error happens I'm not sure what to do/patch/change here.
For properly handling retry, that sounds tricky. Would probably need some exponential back-off to not end up with a loop.

For my specific case I think I'd already be happy if I could stick to the default of on_error=on_error_raise.
But I have no clue what kind of error could be raised, would basically have to catch Exception (which you currently also do) which is not ideal.

Would it make sense to only catch expected Exceptions (instead of Exception) and re-raise them as ApiError instead?
Then the user can catch ApiError and do whatever he wants - and there will not be any unknown side effects.
Personally I think I'd then drop the on_error parameter all together as the user has a well-known Exception subclass he can catch and handle according to his needs.

[gtsystem]

The on_error parameter is a custom function that allows the user to decide how to respond to an exception. In theory the user could decide to ignore retry for some errors and raise in other cases. The exception is captured and passed to the function which decide what to do and, in case of a retry, how long to wait (backoff strategy can be implemented in that function as well).
If the user don't want to use this feature, he can just keep using the default value which is just raising the exception.

I'll try to explain better what I had in mind:

1. process_one_line should be changed to detect the error and raise an ApiError. This change is necessary because we want to capture this specific case.
2. If the on_error functions returns OnErrorAction.RAISE, we just reraise the ApiError. My understanding is that this is what you are looking for and no other changes are needed
3. If the on_error functions returns OnErrorAction.STOP, we stop silently
4. If the on_error functions returns OnErrorAction.RETRY, we should retry, but we already know that the request will fail forever starting from the same resourceVersion. So I'm suggesting to verify this specific condition and reset the resourceVersion before retry

    def watch(self, br: BasicRequest, on_error: OnErrorHandler = on_error_raise):
        wd = WatchDriver(br, self._client.build_request, self._lazy)
        err_count = 0
        while True:
            req = wd.get_request(timeout=self._watch_timeout)
            resp = self.send(req, stream=True)
            try:
                resp.raise_for_status()
                err_count = 0
                for line in resp.iter_lines():
                    yield wd.process_one_line(line)
            except Exception as e:
                err_count += 1
                handle_error = on_error(e, err_count)
                if handle_error.action is OnErrorAction.RAISE:
                    raise
                if handle_error.action is OnErrorAction.STOP:
                    break
                if isinstance(e, ApiError) and e.status.code == 410:  # Detect unrecoverable error
                    wd.resetResourceVersion()  # made up function to unset the resourceVersion.
                if handle_error.sleep > 0:
                    time.sleep(handle_error.sleep)
                continue

Last thing to do, would be to document the behavior, so users understand that forcing a retry in case of the error above will reset the resourceVersion.

The alternative for the case 4. would be to ignore the user request to retry and raise the exception, but this looks to me more counter intuitive.

[gtsystem]

If you prefer, you could just implement the point 1 above, and I can take care of the rest of the logic at a later stage.

[asteven]

Now I understand. Thanks for the details.

So we basically have 2 things here:

1. the KeyError that is currently thrown is a bug, that should not happen
2. we could add logic to deal with some special cases

I can create a patch for the first case.
Would something like this work for you?

diff --git a/lightkube/core/generic_client.py b/lightkube/core/generic_client.py
index bc1cdfb..dc09ba8 100644
--- a/lightkube/core/generic_client.py
+++ b/lightkube/core/generic_client.py
@@ -62,6 +62,15 @@ class BasicRequest:
     headers: Dict[str, str] = None


+# TODO: maybe different/better base exception?
+class ApiStatusError(Exception):
+    def __init__(self, message, code, reason, status):
+        self.message = message
+        self.code = code
+        self.reason = reason
+        self.status = status
+
+
 class WatchDriver:
     def __init__(self, br: BasicRequest, build_request, lazy):
         self._version = br.params.get("resourceVersion")
@@ -80,6 +89,14 @@ class WatchDriver:
         line = json.loads(line)
         tp = line["type"]
         obj = line["object"]
+        if tp == 'ERROR':
+            # TODO: are there other cases here?
+            raise ApiStatusError(
+                obj['message'],
+                obj['code'],
+                obj['reason'],
+                obj['status'],
+            )
         self._version = obj["metadata"]["resourceVersion"]
         return tp, self._convert(obj, lazy=self._lazy)

@@ -354,11 +371,11 @@ class GenericSyncClient(GenericClient):
                 err_count = 0
                 for line in resp.iter_lines():
                     yield wd.process_one_line(line)
-            except Exception as e:
+            except ApiStatusError as e:
                 err_count += 1
                 handle_error = on_error(e, err_count)
                 if handle_error.action is OnErrorAction.RAISE:
-                    raise
+                    raise ApiError(e.message, request=req, response=resp) from e
                 if handle_error.action is OnErrorAction.STOP:
                     break
                 if handle_error.sleep > 0:
@@ -412,11 +429,11 @@ class GenericAsyncClient(GenericClient):
                 err_count = 0
                 async for line in resp.aiter_lines():
                     yield wd.process_one_line(line)
-            except Exception as e:
+            except ApiStatusError as e:
                 err_count += 1
                 handle_error = on_error(e, err_count)
                 if handle_error.action is OnErrorAction.RAISE:
-                    raise
+                    raise ApiError(e.message, request=req, response=resp) from e
                 if handle_error.action is OnErrorAction.STOP:
                     break
                 if handle_error.sleep > 0:

Note that I went from catching Exception to just ApiStatusError - are there any other well known exceptions that can happen here? Or just catch both?

[asteven]

Re the 2. point "raise or ignore or retry plus add logic to deal with some special cases":

Going from your example in the docs
https://lightkube.readthedocs.io/en/latest/list-watch/

The whole on_error handling machinery seems much more complex then just catching an Exception and
dealing with it.

e.g.

seen_pods = {}
try:
    async for pod in (podlist := client.list(Pod)):
        seen_pods[pod.metadata.name] = pod
    async for event, pod in client.watch(Pod, resource_version=podlist.resourceVersion):
        match event:
            case "ADDED" | "MODIFIED":
                seen_pods[pod.metadata.name] = pod
            case "DELETED":
                del seen_pods[pod.metadata.name]
except ApiError as e:
    # decide what to do myself
    ...

vs

def handle_error(...):
    # do something with error

seen_pods = {}
async for pod in (podlist := client.list(Pod)):
    seen_pods[pod.metadata.name] = pod
async for event, pod in client.watch(Pod, resource_version=podlist.resourceVersion, on_error=handle_error):
    match event:
        case "ADDED" | "MODIFIED":
            seen_pods[pod.metadata.name] = pod
        case "DELETED":
            del seen_pods[pod.metadata.name]

IMHO the first example is easier to understand and deal with for a user, and needs zero documentation (other then knowing that watch can raise ApiError.

Personally I'd keep the existing code for back compat, but promote and document the first way of dealing with errors.
Off course maybe there's other use cases that I'm not aware of, so things might be more complicated then it seems.

[gtsystem]

Note that I went from catching Exception to just ApiStatusError - are there any other well known exceptions that can happen here? Or just catch both?

Yes, there could be some other errors, like network interruptions, server side connection closing, decoding issues.. So I would prefer to keep a wider exception (maybe httpx HTTPError?) there and handle the ApiStatusError to ApiError conversion manually.

except HTTPError as e:
  if isinstance(e, ApiStatusError):
    e = ApiError(e.message, request=req, response=resp)
  ...
  if handle_error.action is OnErrorAction.RAISE:
    raise e

IMHO the first example is easier to understand and deal with for a user, and needs zero documentation (other then knowing that watch can raise ApiError.

The first example is missing a while loop (which is needed to implement retry) and so contains two extra levels of nesting + error recovering code making the core business logic less readable (IMHO). But as I said, one pattern doesn't preclude the other, and I don't really want to convince you :)

[asteven]

If you prefer, you could just implement the point 1 above, and I can take care of the rest of the logic at a later stage.

Done #109.