kraken: add support for zenoh queryables

Author: nicoschmdt

core/services/kraken/api/app.py

@@ -1,6 +1,9 @@
 from os import path
 
+from contextlib import asynccontextmanager
+from typing import AsyncGenerator
 from commonwealth.utils.apis import GenericErrorHandlingRoute
+from commonwealth.utils.zenoh_helper import ZenohRouter, zenoh_session
 from fastapi import FastAPI
 from fastapi.responses import RedirectResponse
 from fastapi.staticfiles import StaticFiles
@@ -14,8 +17,19 @@
     index_router_v2,
     jobs_router_v2,
     manifest_router_v2,
+    zenoh_container_router,
+    zenoh_extension_router,
+    zenoh_jobs_router,
+    zenoh_manifest_router,
 )
 
+
+@asynccontextmanager
+async def lifespan(fastapi_app: FastAPI) -> AsyncGenerator[None, None]:  # pylint: disable=unused-argument
+    yield
+    zenoh_session.close()
+
+
 application = FastAPI(
     title="Kraken API",
     description="Kraken is the BlueOS service responsible for installing and managing extensions.",
@@ -33,7 +47,16 @@
 application.include_router(jobs_router_v2)
 application.include_router(manifest_router_v2)
 
-application = VersionedFastAPI(application, prefix_format="/v{major}.{minor}", enable_latest=True)
+application = VersionedFastAPI(application, prefix_format="/v{major}.{minor}", enable_latest=True, lifespan=lifespan)
+
+# Zenoh
+zenoh_router = ZenohRouter("kraken")
+zenoh_router.include_router(zenoh_container_router)
+zenoh_router.include_router(zenoh_extension_router)
+zenoh_router.include_router(zenoh_jobs_router)
+zenoh_router.include_router(zenoh_manifest_router)
+
+zenoh_router.declare()
 
 
 @application.get("/", status_code=200)

core/services/kraken/api/v2/routers/__init__.py

@@ -1,8 +1,18 @@
 # pylint: disable=W0406
-from .container import container_router_v2
-from .extension import extension_router_v2
+from .container import container_router_v2, zenoh_container_router
+from .extension import extension_router_v2, zenoh_extension_router
 from .index import index_router_v2
-from .jobs import jobs_router_v2
-from .manifest import manifest_router_v2
+from .jobs import jobs_router_v2, zenoh_jobs_router
+from .manifest import manifest_router_v2, zenoh_manifest_router
 
-__all__ = ["container_router_v2", "extension_router_v2", "index_router_v2", "jobs_router_v2", "manifest_router_v2"]
+__all__ = [
+    "container_router_v2",
+    "extension_router_v2",
+    "index_router_v2",
+    "jobs_router_v2",
+    "manifest_router_v2",
+    "zenoh_container_router",
+    "zenoh_extension_router",
+    "zenoh_jobs_router",
+    "zenoh_manifest_router",
+]

core/services/kraken/api/v2/routers/container.py

@@ -2,6 +2,7 @@
 from typing import Any, Callable, Optional, Tuple
 
 from commonwealth.utils.streaming import streamer, timeout_streamer
+from commonwealth.utils.zenoh_helper import ZenohRouter, apply_route_decorator
 from fastapi import APIRouter, HTTPException, status
 from fastapi.responses import StreamingResponse
 from fastapi_versioning import versioned_api_route
@@ -10,12 +11,15 @@
 from harbor.exceptions import ContainerNotFound
 from harbor.models import ContainerModel, ContainerUsageModel
 
-container_router_v2 = APIRouter(
+original_container_router_v2 = APIRouter(
     prefix="/container",
     tags=["container_v2"],
     route_class=versioned_api_route(2, 0),
     responses={status.HTTP_404_NOT_FOUND: {"description": "Not found"}},
 )
+container_router_v2 = apply_route_decorator(original_container_router_v2)
+
+zenoh_container_router = ZenohRouter("container")
 
 
 def container_to_http_exception(endpoint: Callable[..., Any]) -> Callable[..., Any]:
@@ -31,6 +35,7 @@ async def wrapper(*args: Tuple[Any], **kwargs: dict[str, Any]) -> Any:
     return wrapper
 
 
+@zenoh_container_router.queryable()
 @container_router_v2.get("/", status_code=status.HTTP_200_OK)
 @container_to_http_exception
 async def list_container() -> list[ContainerModel]:
@@ -40,6 +45,7 @@ async def list_container() -> list[ContainerModel]:
     return await ContainerManager.get_running_containers()
 
 
+@zenoh_container_router.queryable()
 @container_router_v2.get("/{container_name}/details", status_code=status.HTTP_200_OK)
 @container_to_http_exception
 async def fetch_container(container_name: str) -> ContainerModel:
@@ -49,6 +55,7 @@ async def fetch_container(container_name: str) -> ContainerModel:
     return await ContainerManager.get_running_container_by_name(container_name)
 
 
+@zenoh_container_router.queryable()
 @container_router_v2.get("/{container_name}/log", status_code=status.HTTP_200_OK)
 @container_to_http_exception
 async def fetch_log_by_container_name(container_name: str, timeout: Optional[int] = None) -> StreamingResponse:
@@ -64,6 +71,7 @@ async def fetch_log_by_container_name(container_name: str, timeout: Optional[int
     return StreamingResponse(streamer(stream, heartbeats=0.1), media_type="text/plain")
 
 
+@zenoh_container_router.queryable()
 @container_router_v2.get("/stats", status_code=status.HTTP_200_OK)
 @container_to_http_exception
 async def list_stats() -> dict[str, ContainerUsageModel]:
@@ -73,6 +81,7 @@ async def list_stats() -> dict[str, ContainerUsageModel]:
     return await ContainerManager.get_containers_stats()
 
 
+@zenoh_container_router.queryable()
 @container_router_v2.get("/{container_name}/stats", status_code=status.HTTP_200_OK)
 @container_to_http_exception
 async def fetch_stats_by_container_name(container_name: str) -> ContainerUsageModel:

core/services/kraken/api/v2/routers/extension.py

@@ -3,6 +3,7 @@
 from typing import Any, Callable, List, Tuple, cast
 
 from commonwealth.utils.streaming import streamer
+from commonwealth.utils.zenoh_helper import ZenohRouter, apply_route_decorator
 from fastapi import APIRouter, HTTPException, status
 from fastapi.responses import Response, StreamingResponse
 from fastapi_versioning import versioned_api_route
@@ -15,12 +16,15 @@
 from extension.extension import Extension
 from extension.models import ExtensionSource
 
-extension_router_v2 = APIRouter(
+original_extension_router_v2 = APIRouter(
     prefix="/extension",
     tags=["extension_v2"],
     route_class=versioned_api_route(2, 0),
     responses={status.HTTP_404_NOT_FOUND: {"description": "Not found"}},
 )
+extension_router_v2 = apply_route_decorator(original_extension_router_v2)
+
+zenoh_extension_router = ZenohRouter("extension")
 
 
 def extension_to_http_exception(endpoint: Callable[..., Any]) -> Callable[..., Any]:
@@ -42,6 +46,7 @@ async def wrapper(*args: Tuple[Any], **kwargs: dict[str, Any]) -> Any:
     return wrapper
 
 
+@zenoh_extension_router.queryable()
 @extension_router_v2.get("/", status_code=status.HTTP_200_OK)
 @extension_to_http_exception
 async def fetch() -> list[ExtensionSource]:
@@ -52,6 +57,7 @@ async def fetch() -> list[ExtensionSource]:
     return [ext.source for ext in extensions]
 
 
+@zenoh_extension_router.queryable()
 @extension_router_v2.get("/{identifier}/details", status_code=status.HTTP_200_OK)
 @extension_to_http_exception
 async def fetch_by_identifier(identifier: str) -> list[ExtensionSource]:
@@ -62,6 +68,7 @@ async def fetch_by_identifier(identifier: str) -> list[ExtensionSource]:
     return [ext.source for ext in extensions]
 
 
+@zenoh_extension_router.queryable()
 @extension_router_v2.get("/{identifier}/{tag}/details", status_code=status.HTTP_200_OK)
 @extension_to_http_exception
 async def fetch_by_identifier_and_tag(identifier: str, tag: str) -> ExtensionSource:

core/services/kraken/api/v2/routers/jobs.py

@@ -4,17 +4,21 @@
 
 from fastapi import APIRouter, Body, HTTPException, status
 from fastapi_versioning import versioned_api_route
+from commonwealth.utils.zenoh_helper import ZenohRouter, apply_route_decorator
 
 from jobs import JobsManager
 from jobs.exceptions import JobNotFound
 from jobs.models import Job, JobMethod
 
-jobs_router_v2 = APIRouter(
+original_jobs_router_v2 = APIRouter(
     prefix="/jobs",
     tags=["jobs_v2"],
     route_class=versioned_api_route(2, 0),
     responses={status.HTTP_404_NOT_FOUND: {"description": "Not found"}},
 )
+jobs_router_v2 = apply_route_decorator(original_jobs_router_v2)
+
+zenoh_jobs_router = ZenohRouter("jobs")
 
 
 def jobs_to_http_exception(endpoint: Callable[..., Any]) -> Callable[..., Any]:
@@ -40,12 +44,14 @@ async def create(
     return job
 
 
+@zenoh_jobs_router.queryable()
 @jobs_router_v2.get("/", status_code=status.HTTP_200_OK)
 @jobs_to_http_exception
 async def fetch() -> List[Job]:
     return JobsManager.get()
 
 
+@zenoh_jobs_router.queryable()
 @jobs_router_v2.get("/{identifier}", status_code=status.HTTP_200_OK)
 @jobs_to_http_exception
 async def fetch_by_identifier(identifier: str) -> Job:

core/services/kraken/api/v2/routers/manifest.py

@@ -3,6 +3,7 @@
 
 from fastapi import APIRouter, HTTPException, status
 from fastapi_versioning import versioned_api_route
+from commonwealth.utils.zenoh_helper import ZenohRouter, apply_route_decorator
 
 from manifest import ManifestManager
 from manifest.exceptions import (
@@ -19,15 +20,17 @@
     UpdateManifestSource,
 )
 
-manifest_router_v2 = APIRouter(
+original_manifest_router_v2 = APIRouter(
     prefix="/manifest",
     tags=["manifest_v2"],
     route_class=versioned_api_route(2, 0),
     responses={status.HTTP_404_NOT_FOUND: {"description": "Not found"}},
 )
-
+manifest_router_v2 = apply_route_decorator(original_manifest_router_v2)
 manifest_manager = ManifestManager.instance()
 
+zenoh_manifest_router = ZenohRouter("manifest")
+
 
 def manifest_to_http_exception(endpoint: Callable[..., Any]) -> Callable[..., Any]:
     @wraps(endpoint)
@@ -46,6 +49,7 @@ async def wrapper(*args: Tuple[Any], **kwargs: dict[str, Any]) -> Any:
     return wrapper
 
 
+@zenoh_manifest_router.queryable()
 @manifest_router_v2.get("/", status_code=status.HTTP_200_OK)
 @manifest_to_http_exception
 async def fetch(data: bool = True, enabled: bool = False) -> list[Manifest]:
@@ -56,6 +60,7 @@ async def fetch(data: bool = True, enabled: bool = False) -> list[Manifest]:
     return await manifest_manager.fetch(data, enabled)
 
 
+@zenoh_manifest_router.queryable()
 @manifest_router_v2.get("/{identifier}/details", status_code=status.HTTP_200_OK)
 @manifest_to_http_exception
 async def fetch_by_identifier(identifier: str, data: bool = True) -> Manifest:
@@ -65,6 +70,7 @@ async def fetch_by_identifier(identifier: str, data: bool = True) -> Manifest:
     return await manifest_manager.fetch_by_identifier(identifier, data)
 
 
+@zenoh_manifest_router.queryable()
 @manifest_router_v2.get("/consolidated", status_code=status.HTTP_200_OK)
 @manifest_to_http_exception
 async def fetch_consolidated() -> list[RepositoryEntry]:
@@ -75,6 +81,7 @@ async def fetch_consolidated() -> list[RepositoryEntry]:
     return await manifest_manager.fetch_consolidated()
 
 
+@zenoh_manifest_router.queryable()
 @manifest_router_v2.get("/tags/{manifest_identifier}/{extension_identifier}/", status_code=status.HTTP_200_OK)
 @manifest_to_http_exception
 async def fetch_ext_tags_from_manifest(
@@ -88,6 +95,7 @@ async def fetch_ext_tags_from_manifest(
     return [str(version) for version in versions]
 
 
+@zenoh_manifest_router.queryable()
 @manifest_router_v2.get("/tags/{extension_identifier}", status_code=status.HTTP_200_OK)
 @manifest_to_http_exception
 async def fetch_ext_tags_from_consolidated(extension_identifier: str, stable: bool = False) -> list[str]: