Stream Pod logs with a generator (#182)

docs/examples/index.md

@@ -5,6 +5,7 @@ This section contains code examples to demonstrate what you can do with `kr8s`.
 ```{toctree}
 creating_resources
 listing_resources
+inspecting_resources
 modifying_resources
 labelling_operator
 ```

docs/examples/inspecting_resources.md

@@ -0,0 +1,58 @@
+# Inspecting resources
+
+## Reading Pod logs
+
+Print out the logs from a Pod.
+
+`````{tab-set}
+
+````{tab-item} Sync
+```python
+from kr8s.objects import Pod
+
+pod = Pod.get("my-pod", namespace="ns")
+for line in pod.logs():
+    print(line)
+```
+````
+
+````{tab-item} Async
+```python
+from kr8s.asyncio.objects import Pod
+
+pod = await Pod.get("my-pod", namespace="ns")
+async for line in pod.logs():
+    print(line)
+```
+````
+
+`````
+
+
+## Follow Pod logs until a timeout
+
+Print out all the logs from a Pod and keep following until a timeout or the Pod is deleted.
+
+`````{tab-set}
+
+````{tab-item} Sync
+```python
+from kr8s.objects import Pod
+
+pod = Pod.get("my-pod", namespace="ns")
+for line in pod.logs(follow=True, timeout=3600):
+    print(line)
+```
+````
+
+````{tab-item} Async
+```python
+from kr8s.asyncio.objects import Pod
+
+pod = await Pod.get("my-pod", namespace="ns")
+async for line in pod.logs(follow=True, timeout=3600):
+    print(line)
+```
+````
+
+`````

docs/object.md

@@ -66,7 +66,7 @@ Some objects also have additional methods that are unique to them.
 
 ```python
 # Get Pod logs
-logs = pod.logs()
+logs = [line for line in pod.logs()]
 
 # Check if Pod containers are ready
 pod.ready()

kr8s/_objects.py

@@ -8,7 +8,7 @@
 import pathlib
 import re
 import time
-from typing import Any, Dict, List, Optional, Type, Union
+from typing import Any, AsyncGenerator, Dict, List, Optional, Type, Union
 
 import anyio
 import httpx
@@ -657,8 +657,48 @@ async def logs(
         timestamps=False,
         tail_lines=None,
         limit_bytes=None,
-    ) -> str:
+        follow=False,
+        timeout=3600,
+    ) -> AsyncGenerator[str, None, None]:
+        """Streams logs from a Pod.
+
+        Args:
+            container: The container to get logs from. Defaults to the first container in the Pod.
+            pretty: If True, return pretty logs. Defaults to False.
+            previous: If True, return previous terminated container logs. Defaults to False.
+            since_seconds: If set, return logs since this many seconds ago.
+            since_time: If set, return logs since this time.
+            timestamps: If True, prepend each log line with a timestamp. Defaults to False.
+            tail_lines: If set, return this many lines from the end of the logs.
+            limit_bytes: If set, return this many bytes from the end of the logs.
+            follow: If True, follow the logs until the timeout is reached. Defaults to False.
+            timeout: If following timeout after this many seconds. Set to None to disable timeout.
+
+        Returns:
+            An async generator yielding log lines.
+
+
+        Example:
+            >>> from kr8s.objects import Pod
+            >>> pod = Pod.get("my-pod")
+            >>> for line in pod.logs():
+            ...     print(line)
+
+            We can also follow logs as they are generated, the generator will yield a new log line as
+            it is generated by the Pod. This blocks indefinitely so we can set a timeout to break
+            after some period of time, the default is ``3600`` (1hr) but can be set to ``None`` to
+            disable the timeout.
+
+            >>> from kr8s.objects import Pod
+            >>> pod = Pod.get("my-pod", namespace="ns")
+            >>> for line in pod.logs(follow=True, timeout=60):
+            ...     # Will continue streaming logs until 60 seconds or the Pod is terminated
+            ...     print(line)
+
+        """
         params = {}
+        if follow:
+            params["follow"] = "true"
         if container is not None:
             params["container"] = container
         if pretty is not None:
@@ -676,14 +716,18 @@ async def logs(
         if limit_bytes is not None:
             params["limitBytes"] = int(limit_bytes)
 
-        async with self.api.call_api(
-            "GET",
-            version=self.version,
-            url=f"{self.endpoint}/{self.name}/log",
-            namespace=self.namespace,
-            params=params,
-        ) as resp:
-            return resp.text
+        with contextlib.suppress(httpx.ReadTimeout):
+            async with self.api.call_api(
+                "GET",
+                version=self.version,
+                url=f"{self.endpoint}/{self.name}/log",
+                namespace=self.namespace,
+                params=params,
+                stream=True,
+                timeout=timeout,
+            ) as resp:
+                async for line in resp.aiter_lines():
+                    yield line
 
     def portforward(self, remote_port: int, local_port: int = None) -> int:
         """Port forward a pod.

kr8s/tests/test_objects.py

@@ -439,7 +439,7 @@ async def test_pod_logs(example_pod_spec):
     await pod.create()
     while not await pod.ready():
         await asyncio.sleep(0.1)
-    log = await pod.logs(container="pause")
+    log = "\n".join([line async for line in pod.logs(container="pause")])
     assert isinstance(log, str)
     await pod.delete()