Unable to stop kopf - Streaming tasks are not stopped: finishing normally

[sujeetkp]

Long story short

I am unable to stop kopf and it keeps showing the below errors and eventually times out.
And once I restart and delete the CRs, kopf never removes the finalizers and leaves orphan resources.

[2025-04-09 00:27:53,086] kopf._cogs.clients.w [DEBUG   ] Stopping the watch-stream for pods.v1 cluster-wide.
[2025-04-09 00:27:53,087] kopf._core.reactor.r [DEBUG   ] Resource observer is cancelled.
[2025-04-09 00:27:53,111] kopf._core.engines.p [DEBUG   ] Keep-alive in 'default' cluster-wide: ok.
[2025-04-09 00:27:55,089] kopf._core.reactor.q [WARNING ] Unprocessed streams left for [(pods.v1, '6e302b10-19a8-413a-916f-30f5646af6bd')].
^C[2025-04-09 00:28:03,091] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:28:13,101] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:28:23,103] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:28:33,113] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:28:43,123] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:28:53,133] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:29:03,143] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:29:13,153] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:29:23,167] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:29:28,015] urllib3.connectionpo [WARNING ] Retrying (Retry(total=1, connect=None, read=None, redirect=None, status=None)) after connection broken by 'ConnectTimeoutError(<urllib3.connection.HTTPSConnection object at 0x7fffe5d6e960>, 'Connection to xxxxxxxxxxxxx.eks.amazonaws.com timed out. (connect timeout=None)')': /apis/hf.sym.example.com/v1alpha1/namespaces/sujeetkp-sym/reqrem/req-rem-b0sr8y8sxtvt
[2025-04-09 00:29:33,177] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:29:43,187] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:29:53,197] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:30:03,207] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:30:13,217] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:30:23,227] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:30:33,237] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}
[2025-04-09 00:30:43,248] kopf._core.reactor.o [DEBUG   ] Streaming tasks are not stopped: finishing normally; tasks left: {<Task cancelling name='watcher for pods.v1@None' coro=<guard() running at /var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py:96> wait_for=<Future pending cb=[shield.<locals>._outer_done_callback() at /nix/store/zv1kaq7f1q20x62kbjv6pfjygw5jmwl6-python3-3.12.7/lib/python3.12/asyncio/tasks.py:922, Task.task_wakeup()]>>}

Kopf version

1.37.5

Kubernetes version

1.31

Python version

3.12.7

Code

Logs

Additional information

No response

[nolar]

Can you please hint what is the original stopping reason/trigger (in order to reproduce it)?

[sujeetkp]

@nolar - I am testing the code and just wanted to kill (Ctrl + c) the current run and restart a new one to reflect the new code change.

[sujeetkp]

Also, I am getting the below credential error as well and after which again "Streaming tasks are not stopped" error pops up.
I am using "kubernetes" python client for login, which takes creds from my local kube config file.

@kopf.on.login()
def custom_login_fn(**kwargs: Dict[Any, Any]) -> kopf.ConnectionInfo:
    """Custom login function for kopf."""
    return kopf.login_with_service_account(**kwargs) or kopf.login_via_client(**kwargs)

File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/structs/credentials.py", line 214, in select
    raise LoginError("Ran out of valid credentials. Consider installing "
kopf._cogs.structs.credentials.LoginError: Ran out of valid credentials. Consider installing an API client library or adding a login handler. See more: https://kopf.readthedocs.io/en/stable/authentication/
[2025-04-09 04:32:34,344] kopf._core.reactor.o [ERROR   ] Watcher for req-mac.v1alpha1.hf.sym.example.com@none has failed: Ran out of valid credentials. Consider installing an API client library or adding a login handler. See more: https://kopf.readthedocs.io/en/stable/authentication/
Traceback (most recent call last):
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/aiokits/aiotasks.py", line 96, in guard
    await coro
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_core/reactor/queueing.py", line 175, in watcher
    async for raw_event in stream:
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/clients/watching.py", line 86, in infinite_watch
    async for raw_event in stream:
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/clients/watching.py", line 201, in continuous_watch
    async for raw_input in stream:
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/clients/watching.py", line 266, in watch_objs
    async for raw_input in api.stream(
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/clients/api.py", line 201, in stream
    response = await request(
               ^^^^^^^^^^^^^^
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/clients/auth.py", line 48, in wrapper
    async for key, info, context in vault.extended(APIContext, 'contexts'):
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/structs/credentials.py", line 158, in extended
    async for key, item in self._items():
  File "/var/shared/sujeetkp/workspace/repos/hg.hf/hg/hf/hg.hf/src/oss/venv/lib/python3.12/site-packages/kopf/_cogs/structs/credentials.py", line 195, in _items
    yielded_key, yielded_item = self.select()
                                ^^^^^^^^^^^^^

[sujeetkp]

@nolar - Did you get a chance to look at it ?
I am getting this error frequently.