Repository: tomplus/kubernetes_asyncio (language: python)

Issue:
Infinite watch stream loop
Hi! The below snippet works just fine on version 30.1.1 when I invoke on a pod in "Succeeded" state:

```python
import asyncio

from kubernetes_asyncio import client, watch, config
from kubernetes_asyncio.client.api_client import ApiClient

if __name__ == "__main__":
    async def main():
        pod = "..."
        namespace = "..."
        await config.load_kube_config()

        async with ApiClient() as api:
            watcher = watch.Watch()
            async for log in watcher.stream(client.CoreV1Api(api).read_namespaced_pod_log, pod, namespace):
                print(log, end="")

    asyncio.run(main())
```
the loop ends and execution stops after the logs finish.

However same snippet keeps looping forever on version 30.3.0. Once the logs finish, it starts over and repeats.


Discussion / hints:
Thanks for reporting this issue and sorry for the inconvenience.

I try to fix it but it looks like the Watch() tries to stream "events" forever (fixed in 30.3.0) and I'm not sure if it's possible to differentiate timeout and end of logs. I'll dig into it...

Is there any reason why you use Watch() in this scenario? You can try https://github.com/tomplus/kubernetes_asyncio/blob/master/examples/tail.py#L41 this approach, should work in you case.

Fix patch:
diff --git a/kubernetes_asyncio/watch/watch.py b/kubernetes_asyncio/watch/watch.py
index 493c700c..ab09f2e3 100644
--- a/kubernetes_asyncio/watch/watch.py
+++ b/kubernetes_asyncio/watch/watch.py
@@ -174,16 +174,20 @@ async def next(self):
 
             line = line.decode('utf8')
 
+            # Special case for faster log streaming
+            if self.return_type == 'str':
+                if line == '':
+                    # end of log
+                    raise StopAsyncIteration
+                return line
+
             # Stop the iterator if K8s sends an empty response. This happens when
             # eg the supplied timeout has expired.
             if line == '':
                 if watch_forever:
                     self._reconnect()
                     continue
-
-            # Special case for faster log streaming
-            if self.return_type == 'str':
-                return line
+                raise StopAsyncIteration
 
             # retry 410 error only once
             try:


Test patch:
diff --git a/kubernetes_asyncio/watch/watch_test.py b/kubernetes_asyncio/watch/watch_test.py
index 28dedbc0..8b513793 100644
--- a/kubernetes_asyncio/watch/watch_test.py
+++ b/kubernetes_asyncio/watch/watch_test.py
@@ -74,7 +74,7 @@ async def test_watch_for_follow(self):
         fake_resp = AsyncMock()
         fake_resp.content.readline = AsyncMock()
         fake_resp.release = Mock()
-        side_effects = ['log_line_1', 'log_line_2']
+        side_effects = ['log_line_1', 'log_line_2', '']
         side_effects = [_.encode('utf8') for _ in side_effects]
         side_effects.extend([AssertionError('Should not have been called')])
         fake_resp.content.readline.side_effect = side_effects
@@ -84,16 +84,12 @@ async def test_watch_for_follow(self):
         fake_api.read_namespaced_pod_log.__doc__ = ':param follow:\n:type follow: bool\n:rtype: str'
 
         watch = kubernetes_asyncio.watch.Watch()
-        count = 1
+        logs = []
         async with watch:
             async for e in watch.stream(fake_api.read_namespaced_pod_log):
-                self.assertEqual("log_line_1", e)
-                # Stop the watch. This must not return the next event which would
-                # be an AssertionError exception.
-                count += 1
-                if count == len(side_effects) - 1:
-                    watch.stop()
+                logs.append(e)
 
+        self.assertListEqual(logs, ['log_line_1', 'log_line_2'])
         fake_api.read_namespaced_pod_log.assert_called_once_with(
             _preload_content=False, follow=True)
         fake_resp.release.assert_called_once_with()