moved service_extras to catalog service

Author: Andrei Neagu

services/catalog/src/simcore_service_catalog/api/rest/_services_extras.py

@@ -16,5 +16,7 @@ async def get_service_extras(
     service_version: ServiceVersion,
     director_client: Annotated[DirectorApi, Depends(get_director_api)],
 ) -> dict[str, Any]:
-    labels = await director_client.get_service_extras(service_key, service_version)
-    return Envelope[dict[str, Any]](data=labels).model_dump(mode="json")
+    service_extras = await director_client.get_service_extras(
+        service_key, service_version
+    )
+    return Envelope[dict[str, Any]](data=service_extras).model_dump(mode="json")

services/catalog/src/simcore_service_catalog/core/application.py

@@ -50,9 +50,7 @@ def create_app(settings: ApplicationSettings | None = None) -> FastAPI:
         setup_tracing(app, settings.CATALOG_TRACING, APP_NAME)
 
     # STARTUP-EVENT
-    app.add_event_handler(
-        "startup", create_on_startup(app, tracing_settings=settings.CATALOG_TRACING)
-    )
+    app.add_event_handler("startup", create_on_startup(app))
 
     # PLUGIN SETUP
     setup_function_services(app)

services/catalog/src/simcore_service_catalog/core/events.py

@@ -5,7 +5,6 @@
 from fastapi import FastAPI
 from servicelib.fastapi.db_asyncpg_engine import close_db_connection, connect_to_db
 from servicelib.logging_utils import log_context
-from settings_library.tracing import TracingSettings
 
 from .._meta import APP_FINISHED_BANNER_MSG, APP_STARTED_BANNER_MSG
 from ..db.events import setup_default_product
@@ -27,9 +26,7 @@ def _flush_finished_banner() -> None:
     print(APP_FINISHED_BANNER_MSG, flush=True)  # noqa: T201
 
 
-def create_on_startup(
-    app: FastAPI, tracing_settings: TracingSettings | None
-) -> EventCallable:
+def create_on_startup(app: FastAPI) -> EventCallable:
     async def _() -> None:
         _flush_started_banner()
 
@@ -40,7 +37,7 @@ async def _() -> None:
 
         if app.state.settings.CATALOG_DIRECTOR:
             # setup connection to director
-            await setup_director(app, tracing_settings=tracing_settings)
+            await setup_director(app)
 
             # FIXME: check director service is in place and ready. Hand-shake??
             # SEE https://github.com/ITISFoundation/osparc-simcore/issues/1728

services/catalog/src/simcore_service_catalog/core/settings.py

@@ -7,7 +7,14 @@
 )
 from models_library.basic_types import LogLevel
 from models_library.services_resources import ResourcesDict, ResourceValue
-from pydantic import AliasChoices, ByteSize, Field, PositiveInt, TypeAdapter
+from pydantic import (
+    AliasChoices,
+    ByteSize,
+    Field,
+    NonNegativeInt,
+    PositiveInt,
+    TypeAdapter,
+)
 from servicelib.logging_utils_filtering import LoggerName, MessageSubstring
 from settings_library.application import BaseApplicationSettings
 from settings_library.base import BaseCustomSettings
@@ -103,3 +110,6 @@ class ApplicationSettings(BaseApplicationSettings, MixinLoggingSettings):
         json_schema_extra={"auto_default_from_env": True},
         description="settings for opentelemetry tracing",
     )
+
+    DIRECTOR_DEFAULT_MAX_MEMORY: NonNegativeInt = Field(default=0)
+    DIRECTOR_DEFAULT_MAX_NANO_CPUS: NonNegativeInt = Field(default=0)

services/catalog/src/simcore_service_catalog/services/access_rights.py

@@ -6,8 +6,7 @@
 import operator
 from collections.abc import Callable
 from datetime import UTC, datetime
-from typing import Any, cast
-from urllib.parse import quote_plus
+from typing import cast
 
 import arrow
 from fastapi import FastAPI
@@ -36,12 +35,8 @@ async def _is_old_service(app: FastAPI, service: ServiceMetaDataPublished) -> bo
     # NOTE: https://github.com/ITISFoundation/osparc-simcore/pull/6003#discussion_r1658200909
     # get service build date
     client = get_director_api(app)
-    data = cast(
-        dict[str, Any],
-        await client.get(
-            f"/service_extras/{quote_plus(service.key)}/{service.version}"
-        ),
-    )
+
+    data = await client.get_service_extras(service.key, service.version)
     if not data or "build_date" not in data:
         return True
     service_build_data = arrow.get(data["build_date"]).datetime

services/catalog/src/simcore_service_catalog/services/director.py

@@ -1,10 +1,12 @@
 import asyncio
 import functools
+import json
 import logging
 import urllib.parse
 from collections.abc import Awaitable, Callable
 from contextlib import suppress
-from typing import Any
+from pprint import pformat
+from typing import Any, Final
 
 import httpx
 from common_library.json_serialization import json_dumps
@@ -13,13 +15,13 @@
 from models_library.services_types import ServiceKey, ServiceVersion
 from servicelib.fastapi.tracing import setup_httpx_client_tracing
 from servicelib.logging_utils import log_context
-from settings_library.tracing import TracingSettings
 from starlette import status
 from tenacity.asyncio import AsyncRetrying
 from tenacity.before_sleep import before_sleep_log
 from tenacity.stop import stop_after_delay
 from tenacity.wait import wait_random
 
+from ..core.settings import ApplicationSettings
 from ..exceptions.errors import DirectorUnresponsiveError
 
 _logger = logging.getLogger(__name__)
@@ -98,6 +100,114 @@ async def request_wrapper(
     return request_wrapper
 
 
+_SERVICE_RUNTIME_SETTINGS: Final[str] = "simcore.service.settings"
+_ORG_LABELS_TO_SCHEMA_LABELS: Final[dict[str, str]] = {
+    "org.label-schema.build-date": "build_date",
+    "org.label-schema.vcs-ref": "vcs_ref",
+    "org.label-schema.vcs-url": "vcs_url",
+}
+
+_CONTAINER_SPEC_ENTRY_NAME = "ContainerSpec".lower()
+_RESOURCES_ENTRY_NAME = "Resources".lower()
+
+
+def _validate_kind(entry_to_validate: dict[str, Any], kind_name: str):
+    for element in (
+        entry_to_validate.get("value", {})
+        .get("Reservations", {})
+        .get("GenericResources", [])
+    ):
+        if element.get("DiscreteResourceSpec", {}).get("Kind") == kind_name:
+            return True
+    return False
+
+
+async def _get_service_extras(
+    director_client: "DirectorApi", image_key: str, image_tag: str
+) -> dict[str, Any]:
+    # check physical node requirements
+    # all nodes require "CPU"
+    result: dict[str, Any] = {
+        "node_requirements": {
+            "CPU": director_client.default_max_nano_cpus / 1.0e09,
+            "RAM": director_client.default_max_memory,
+        }
+    }
+
+    labels = await director_client.get_service_labels(image_key, image_tag)
+    _logger.debug("Compiling service extras from labels %s", pformat(labels))
+
+    if _SERVICE_RUNTIME_SETTINGS in labels:
+        service_settings: list[dict[str, Any]] = json.loads(
+            labels[_SERVICE_RUNTIME_SETTINGS]
+        )
+        for entry in service_settings:
+            entry_name = entry.get("name", "").lower()
+            entry_value = entry.get("value")
+            invalid_with_msg = None
+
+            if entry_name == _RESOURCES_ENTRY_NAME:
+                if entry_value and isinstance(entry_value, dict):
+                    res_limit = entry_value.get("Limits", {})
+                    res_reservation = entry_value.get("Reservations", {})
+                    # CPU
+                    result["node_requirements"]["CPU"] = (
+                        float(res_limit.get("NanoCPUs", 0))
+                        or float(res_reservation.get("NanoCPUs", 0))
+                        or director_client.default_max_nano_cpus
+                    ) / 1.0e09
+                    # RAM
+                    result["node_requirements"]["RAM"] = (
+                        res_limit.get("MemoryBytes", 0)
+                        or res_reservation.get("MemoryBytes", 0)
+                        or director_client.default_max_memory
+                    )
+                else:
+                    invalid_with_msg = f"invalid type for resource [{entry_value}]"
+
+                # discrete resources (custom made ones) ---
+                # check if the service requires GPU support
+                if not invalid_with_msg and _validate_kind(entry, "VRAM"):
+
+                    result["node_requirements"]["GPU"] = 1
+                if not invalid_with_msg and _validate_kind(entry, "MPI"):
+                    result["node_requirements"]["MPI"] = 1
+
+            elif entry_name == _CONTAINER_SPEC_ENTRY_NAME:
+                # NOTE: some minor validation
+                # expects {'name': 'ContainerSpec', 'type': 'ContainerSpec', 'value': {'Command': [...]}}
+                if (
+                    entry_value
+                    and isinstance(entry_value, dict)
+                    and "Command" in entry_value
+                ):
+                    result["container_spec"] = entry_value
+                else:
+                    invalid_with_msg = f"invalid container_spec [{entry_value}]"
+
+            if invalid_with_msg:
+                _logger.warning(
+                    "%s entry [%s] encoded in settings labels of service image %s:%s",
+                    invalid_with_msg,
+                    entry,
+                    image_key,
+                    image_tag,
+                )
+
+    # get org labels
+    result.update(
+        {
+            sl: labels[dl]
+            for dl, sl in _ORG_LABELS_TO_SCHEMA_LABELS.items()
+            if dl in labels
+        }
+    )
+
+    _logger.debug("Following service extras were compiled: %s", pformat(result))
+
+    return result
+
+
 class DirectorApi:
     """
     - wrapper around thin-client to simplify director's API
@@ -108,16 +218,22 @@ class DirectorApi:
     SEE services/catalog/src/simcore_service_catalog/api/dependencies/director.py
     """
 
-    def __init__(
-        self, base_url: str, app: FastAPI, tracing_settings: TracingSettings | None
-    ):
+    def __init__(self, base_url: str, app: FastAPI):
+        settings: ApplicationSettings = app.state.settings
+
+        assert settings.CATALOG_CLIENT_REQUEST  # nosec
         self.client = httpx.AsyncClient(
             base_url=base_url,
-            timeout=app.state.settings.CATALOG_CLIENT_REQUEST.HTTP_CLIENT_REQUEST_TOTAL_TIMEOUT,
+            timeout=settings.CATALOG_CLIENT_REQUEST.HTTP_CLIENT_REQUEST_TOTAL_TIMEOUT,
         )
-        if tracing_settings:
+        if settings.CATALOG_TRACING:
             setup_httpx_client_tracing(self.client)
-        self.vtag = app.state.settings.CATALOG_DIRECTOR.DIRECTOR_VTAG
+
+        assert settings.CATALOG_DIRECTOR  # nosec
+        self.vtag = settings.CATALOG_DIRECTOR.DIRECTOR_VTAG
+
+        self.default_max_memory = settings.DIRECTOR_DEFAULT_MAX_MEMORY
+        self.default_max_nano_cpus = settings.DIRECTOR_DEFAULT_MAX_NANO_CPUS
 
     async def close(self):
         await self.client.aclose()
@@ -172,32 +288,18 @@ async def get_service_extras(
         service_key: ServiceKey,
         service_version: ServiceVersion,
     ) -> dict[str, Any]:
-        response = await self.get(
-            f"/service_extras/{urllib.parse.quote_plus(service_key)}/{service_version}"
-        )
-        assert isinstance(response, dict)  # nosec
-        return response
+        return await _get_service_extras(self, service_key, service_version)
 
 
-async def setup_director(
-    app: FastAPI, tracing_settings: TracingSettings | None
-) -> None:
+async def setup_director(app: FastAPI) -> None:
     if settings := app.state.settings.CATALOG_DIRECTOR:
         with log_context(
             _logger, logging.DEBUG, "Setup director at %s", f"{settings.base_url=}"
         ):
             async for attempt in AsyncRetrying(**_director_startup_retry_policy):
-                client = DirectorApi(
-                    base_url=settings.base_url,
-                    app=app,
-                    tracing_settings=tracing_settings,
-                )
+                client = DirectorApi(base_url=settings.base_url, app=app)
                 with attempt:
-                    client = DirectorApi(
-                        base_url=settings.base_url,
-                        app=app,
-                        tracing_settings=tracing_settings,
-                    )
+                    client = DirectorApi(base_url=settings.base_url, app=app)
                     if not await client.is_responsive():
                         with suppress(Exception):
                             await client.close()

services/director/src/simcore_service_director/api/rest/_service_extras.py

@@ -6,7 +6,7 @@
     http_exception_as_json_response,
 )
 
-from . import _health, _running_interactive_services, _service_extras, _services
+from . import _health, _running_interactive_services, _services
 
 _V0_VTAG: Final[str] = "v0"
 
@@ -22,7 +22,6 @@ def setup_api_routes(app: FastAPI):
     # include the rest under /vX
     api_router = APIRouter(prefix=f"/{_V0_VTAG}")
     api_router.include_router(_services.router, tags=["services"])
-    api_router.include_router(_service_extras.router, tags=["services"])
     api_router.include_router(_running_interactive_services.router, tags=["services"])
     app.include_router(api_router)