kraken: get historical extension logs through zenoh queryable

Author: nicoschmdt

core/services/kraken/api/app.py

@@ -49,6 +49,7 @@ async def lifespan(fastapi_app: FastAPI) -> AsyncGenerator[None, None]:  # pylin
 zenoh_router = ZenohRouter(SERVICE_NAME)
 zenoh_router.add_routes_to_zenoh(application)
 
+
 application = VersionedFastAPI(application, prefix_format="/v{major}.{minor}", enable_latest=True, lifespan=lifespan)
 
 

core/services/kraken/harbor/container.py

@@ -123,6 +123,16 @@ async def get_container_log_by_name(cls, container_name: str) -> AsyncGenerator[
                 yield log_line
             logger.info(f"Finished streaming logs for {container_name}")
 
+    @classmethod
+    async def get_container_historical_logs(cls, container_name: str) -> List[str]:
+        async with DockerCtx() as client:
+            try:
+                container = await cls.get_raw_container_by_name(client, container_name)
+            except ContainerNotFound as error:
+                raise StackedHTTPException(status_code=status.HTTP_404_NOT_FOUND, error=error) from error
+
+            return await container.log(stdout=True, stderr=True, follow=False, stream=False)  # type: ignore
+
     @classmethod
     async def get_containers_stats(cls) -> Dict[str, ContainerUsageModel]:
         async with DockerCtx() as client:

core/services/kraken/main.py

@@ -15,6 +15,9 @@
 from api import application
 from jobs import JobsManager
 from kraken import Kraken
+from zenoh_handlers import (  # noqa: F401 # pylint: disable=unused-import
+    app as zenoh_app,
+)
 
 kraken = Kraken()
 jobs = JobsManager()

core/services/kraken/zenoh_handlers/__init__.py

@@ -0,0 +1,3 @@
+from . import app
+
+__all__ = ["app"]

core/services/kraken/zenoh_handlers/app.py

@@ -0,0 +1,10 @@
+from commonwealth.utils.zenoh_helper import ZenohRouter, ZenohSession
+from config import SERVICE_NAME
+from zenoh_handlers.extension_handler import ExtensionHandlers
+
+session = ZenohSession(SERVICE_NAME)
+router = ZenohRouter(SERVICE_NAME)
+
+# Extension
+extension_handlers = ExtensionHandlers(router)
+extension_handlers.register_queryables()

core/services/kraken/zenoh_handlers/extension_handler.py

@@ -0,0 +1,54 @@
+from typing import Any, List, cast
+
+from commonwealth.utils.zenoh_helper import ZenohRouter
+from extension.extension import Extension
+from extension_logs import ExtensionLogPublisher
+from harbor import ContainerManager
+from loguru import logger
+from settings import ExtensionSettings
+
+
+class ExtensionHandlers:
+    def __init__(self, router: ZenohRouter) -> None:
+        self.router = router
+
+    async def logs_request_handler(self, extension_name: str) -> dict[str, Any]:
+        if not extension_name:
+            return {"error": "extension_name parameter is required"}
+
+        try:
+            extensions = cast(List[ExtensionSettings], Extension._fetch_settings())
+            extension = next((ext for ext in extensions if extension_name in (ext.identifier, ext.name)), None)
+
+            if not extension:
+                return {"error": f"Extension {extension_name} not found"}
+
+            if not extension.enabled:
+                return {"error": f"Extension {extension_name} is not enabled"}
+
+            topic = ExtensionLogPublisher._topic_for(extension)
+
+            container_name = extension.container_name()
+
+            raw_logs = await ContainerManager.get_container_historical_logs(container_name)
+            formatted_messages = []
+            for raw_line in raw_logs:
+                level, _ = ExtensionLogPublisher._extract_level(raw_line)
+                formatted_messages.append(
+                    {
+                        "level": level,
+                        "message": raw_line,
+                    }
+                )
+            return {
+                "status": "success",
+                "messages": formatted_messages,
+                "total_lines": len(formatted_messages),
+                "topic": topic,
+            }
+        except Exception as e:
+            logger.exception(f"Error handling logs request for {extension_name}")
+            return {"error": str(e), "error_type": type(e).__name__}
+
+    def register_queryables(self) -> None:
+        self.router.add_queryable("extension/logs/request", self.logs_request_handler)