core: kraken: stream container logs

core/services/kraken/kraken.py

@@ -280,11 +280,16 @@ async def list_containers(self) -> List[DockerContainer]:
         containers: List[DockerContainer] = await self.client.containers.list(filter='{"status": ["running"]}')  # type: ignore
         return containers
 
-    async def load_logs(self, container_name: str) -> List[str]:
-        containers = await self.client.containers.list(filters={"name": {container_name: True}})  # type: ignore
+    async def stream_logs(self, container_name: str, timeout : int = 30) -> AsyncGenerator[str, None]:
+        containers = await self.client.containers.list(filters={"name": {container_name: True}}) # type: ignore
         if not containers:
             raise RuntimeError(f"Container not found: {container_name}")
-        return cast(List[str], await containers[0].log(stdout=True, stderr=True))
+
+        start_time = asyncio.get_event_loop().time()
+        async for log_line in containers[0].log(stdout=True, stderr=True, follow=True, stream=True):
+            if asyncio.get_event_loop().time() - start_time > timeout:
+                break
+            yield log_line
 
     # pylint: disable=too-many-locals
     async def load_stats(self) -> Dict[str, Any]:

core/services/kraken/main.py

@@ -2,13 +2,13 @@
 import argparse
 import asyncio
 import logging
-from typing import Any, List
+from typing import Any, Iterable
 
 from commonwealth.utils.apis import GenericErrorHandlingRoute
 from commonwealth.utils.general import limit_ram_usage
 from commonwealth.utils.logs import InterceptHandler, init_logger
 from fastapi import FastAPI, HTTPException, status
-from fastapi.responses import HTMLResponse, StreamingResponse
+from fastapi.responses import HTMLResponse, PlainTextResponse, StreamingResponse
 from fastapi_versioning import VersionedFastAPI, version
 from loguru import logger
 from pydantic import BaseModel
@@ -129,10 +129,10 @@ async def list_containers() -> Any:
     ]
 
 
-@app.get("/log", status_code=status.HTTP_200_OK)
+@app.get("/log", status_code=status.HTTP_200_OK, response_class=PlainTextResponse)
 @version(1, 0)
-async def log_containers(container_name: str) -> List[str]:
-    return await kraken.load_logs(container_name)
+async def log_containers(container_name: str) -> Iterable[bytes]:
+    return StreamingResponse(kraken.stream_logs(container_name)) # type: ignore
 
 
 @app.get("/stats", status_code=status.HTTP_200_OK)