core:services:kraken: Fix 5 min timeout

* Make sure kraken does not keeps reattaching to stream logs due to
  internal aiohttp 5 minutes timeout

Author: joaomariolago

core/services/kraken/harbor/container.py

@@ -113,7 +113,7 @@ async def get_running_container_by_name(cls, container_name: str) -> ContainerMo
 
     @classmethod
     async def get_container_log_by_name(cls, container_name: str) -> AsyncGenerator[str, None]:
-        async with DockerCtx() as client:
+        async with DockerCtx(timeout=0) as client:
             try:
                 container = await cls.get_raw_container_by_name(client, container_name)
             except ContainerNotFound as error:

core/services/kraken/harbor/contexts.py

@@ -1,5 +1,6 @@
-from typing import Any
+from typing import Any, Optional
 
+import aiohttp
 from aiodocker import Docker
 
 
@@ -8,8 +9,20 @@ class DockerCtx:
     Context manager for Docker clients.
     """
 
-    def __init__(self) -> None:
-        self._client: Docker = Docker()
+    def __init__(self, timeout: Optional[int] = None) -> None:
+        if timeout is None:
+            self._client: Docker = Docker()
+        else:
+            # aiodocker will not create a session if is different from None
+            self._client: Docker = Docker(session=True)  # type: ignore
+            # We insert a new session with desired timeout
+            self._client.session = self._client.session = aiohttp.ClientSession(
+                connector=self._client.connector,
+                timeout=aiohttp.ClientTimeout(
+                    total=None if timeout == 0 else timeout,
+                    sock_read=None if timeout == 0 else timeout,
+                ),
+            )
 
     async def __aenter__(self) -> Docker:
         return self._client

core/services/kraken/kraken.py

@@ -165,6 +165,12 @@ async def kill_dangling_containers(self) -> None:
                 except Exception as e:
                     logger.warning(f"Dangling container {container_name} could not be removed: {e}")
 
+    async def check_extension_logs_task(self) -> None:
+        try:
+            await self.extension_log_publisher.sync_with_running_extensions()
+        except Exception as error:
+            logger.debug(f"Failed to sync extension log streams: {error}")
+
     async def start_starter_task(self) -> None:
         while self.is_running:
             await self.init_dead_extensions()
@@ -177,17 +183,10 @@ async def start_cleaner_task(self) -> None:
             await self.kill_invalid_extensions()
             await self.kill_dangling_containers()
             await self.cleanup_temporary_extensions()
+            await self.check_extension_logs_task()
 
             await asyncio.sleep(60)
 
-    async def start_extension_logs_task(self) -> None:
-        while self.is_running:
-            try:
-                await self.extension_log_publisher.sync_with_running_extensions()
-            except Exception as error:
-                logger.debug(f"Failed to sync extension log streams: {error}")
-            await asyncio.sleep(2)
-
     async def stop(self) -> None:
         self.is_running = False
         await self.extension_log_publisher.shutdown()

core/services/kraken/main.py

@@ -38,7 +38,6 @@ async def main() -> None:
     # Launch background tasks
     asyncio.create_task(kraken.start_cleaner_task())
     asyncio.create_task(kraken.start_starter_task())
-    asyncio.create_task(kraken.start_extension_logs_task())
     asyncio.create_task(jobs.start())
 
     await server.serve()