Watch raising asyncio.CancelledError

[alchibis]

I have code that is asynchronously creating 5 different watchers for 5 different kubernetes resources (by for looping asynchronously over a set of kubernetes resources), and despite a high timeout that I set, the watcher code raises an asyncio.CancelledError before the timeout. The behavior of it is quite strange (I've determined it by via numerous print statements):

1. It will finish watching for four of the five watcher loops, but it will not exit the watcher code (it just stalls, even when I return)
2. For the fifth call it won't even begin iterating through the events, it just stalls
3. Then it raises a Cancelled Error and everything fails.

This is the watcher code I use:

dyn_client = await kubernetes_asyncio.dynamic.DynamicClient(
            kubernetes_asyncio.client.api_client.ApiClient(
                configuration=kubernetes_asyncio.config.load_incluster_config()
            )
        )
dynamic_resource_api = await dyn_client.resources.get(
            api_version=api_version, kind=kind
        )
watcher = watch.Watch()
        try:
            async for event in resource_api.watch(
                    namespace=namespace, watcher=watcher, label_selector=label_selector
            ):
                obj = event["object"]
                obj_name = obj.metadata.name
                event_type = event["type"]
                if event_type == "CREATE:
                    logger.info("Found event")
                    break
        except asyncio.CancelledError:
            print("Watch cancelled (possibly by shutdown or timeout)")
        finally:
            await watcher.close()

I've also tried numerous other small iterations of the above code, but the behavior is exactly the same across instances. This feels related to this issue that was raised, but they did not mention the CancelledError so I am creating a new one.

Some of the things I've tried:

1. Setting the timeout to None
2. Using the dynamic_client.watch() with the resource_api.get passed in
3. Various timeouts
4. Using watcher.stream directly
5. Using the asyncio.wait instead of asyncio.wait_for on the above code (which is wrapped in a bigger function)

The version of kubernetes_asyncio library I'm using is 32.0.0, which is the recommended version from that issue, so that did not seem to solve it. The syntax of my code is also a bit different from that issue... I was following the example that is defined in the watch function in the kubernetes_asyncio.dynamic.client file.

Please let me know if you need more details/if this issue has been resolved already. Thank you very much.

[tomplus]

Could you try to use the previous version v31.1.1? Does it solve the issue?

[tomplus]

Could you share a stack trace with the exception?

[alchibis]

Hello, apologies for the delayed response ! I tried 31.1.1 but that didn't fix it. Here is a stack trace:

Traceback (most recent call last):
  File "/lib/my_service/server/my_code.py", line 525, in my_func1
    await self.my_func2(
  File "/lib/my_service/server/my_code.py", line 841, in my_func2
    async for event in self.dynamic_client.watch(
  File "/usr/local/lib/python3.10/site-packages/kubernetes_asyncio/dynamic/client.py", line 221, in watch
    async for event in watcher.stream(
  File "/usr/local/lib/python3.10/site-packages/kubernetes_asyncio/watch/watch.py", line 135, in __anext__
    return await self.next()
  File "/usr/local/lib/python3.10/site-packages/kubernetes_asyncio/watch/watch.py", line 165, in next
    line = await self.resp.content.readline()
  File "/usr/local/lib/python3.10/site-packages/aiohttp/streams.py", line 352, in readline
    return await self.readuntil()
  File "/usr/local/lib/python3.10/site-packages/aiohttp/streams.py", line 386, in readuntil
    await self._wait("readuntil")
  File "/usr/local/lib/python3.10/site-packages/aiohttp/streams.py", line 347, in _wait
    await waiter
asyncio.exceptions.CancelledError