dynamic-scheduler now uses RPC to reach the catalog

packages/pytest-simcore/src/pytest_simcore/helpers/catalog_rpc.py

@@ -24,6 +24,7 @@
     PageLimitInt,
     PageRpc,
 )
+from models_library.service_settings_labels import SimcoreServiceLabels
 from models_library.services_enums import ServiceType
 from models_library.services_history import ServiceRelease
 from models_library.services_regex import (
@@ -192,6 +193,22 @@ async def get_service_ports(
             ServicePortGet.model_json_schema()["examples"],
         )
 
+    @validate_call(config={"arbitrary_types_allowed": True})
+    async def get_service_labels(
+        self,
+        rpc_client: RabbitMQRPCClient | MockType,
+        *,
+        service_key: ServiceKey,
+        service_version: ServiceVersion,
+    ) -> SimcoreServiceLabels:
+        assert rpc_client
+        assert service_key
+        assert service_version
+
+        return TypeAdapter(SimcoreServiceLabels).validate_python(
+            SimcoreServiceLabels.model_json_schema()["examples"][1],
+        )
+
     @validate_call(config={"arbitrary_types_allowed": True})
     async def list_all_services_summaries_paginated(
         self,
@@ -256,6 +273,7 @@ class ZeroListingCatalogRpcSideEffects:
     async def list_services_paginated(self, *args, **kwargs): ...
     async def get_service(self, *args, **kwargs): ...
     async def update_service(self, *args, **kwargs): ...
+    async def get_service_labels(self, *args, **kwargs): ...
     async def get_service_ports(self, *args, **kwargs): ...
     async def list_my_service_history_latest_first(self, *args, **kwargs):  # noqa: ARG002
         return PageRpc[ServiceRelease].create(

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/catalog/services.py

@@ -33,6 +33,7 @@
     PageLimitInt,
     PageRpc,
 )
+from models_library.service_settings_labels import SimcoreServiceLabels
 from models_library.services_types import ServiceKey, ServiceVersion
 from models_library.users import UserID
 from pydantic import TypeAdapter, validate_call
@@ -255,6 +256,28 @@ async def get_service_ports(
     return cast(list[ServicePortGet], result)
 
 
+@validate_call(config={"arbitrary_types_allowed": True})
+@log_decorator(_logger, level=logging.DEBUG)
+async def get_service_labels(
+    rpc_client: RabbitMQRPCClient,
+    *,
+    service_key: ServiceKey,
+    service_version: ServiceVersion,
+) -> SimcoreServiceLabels:
+    """Gets the docker image labels of a specific service version
+
+    Raises:
+        ValidationError: on invalid arguments
+    """
+    result = await rpc_client.request(
+        CATALOG_RPC_NAMESPACE,
+        TypeAdapter(RPCMethodName).validate_python("get_service_labels"),
+        service_key=service_key,
+        service_version=service_version,
+    )
+    return TypeAdapter(SimcoreServiceLabels).validate_python(result)
+
+
 @validate_call(config={"arbitrary_types_allowed": True})
 async def list_all_services_summaries_paginated(  # pylint: disable=too-many-arguments
     rpc_client: RabbitMQRPCClient,

services/catalog/src/simcore_service_catalog/api/rpc/_services.py

@@ -16,6 +16,7 @@
 from models_library.products import ProductName
 from models_library.rest_pagination import PageOffsetInt
 from models_library.rpc_pagination import DEFAULT_NUMBER_OF_ITEMS_PER_PAGE, PageLimitInt
+from models_library.service_settings_labels import SimcoreServiceLabels
 from models_library.services_types import ServiceKey, ServiceVersion
 from models_library.users import UserID
 from pydantic import TypeAdapter, ValidationError, validate_call
@@ -324,6 +325,22 @@ async def get_service_ports(
     ]
 
 
+@router.expose(reraise_if_error_type=(ValidationError,))
+@validate_call(config={"arbitrary_types_allowed": True})
+async def get_service_labels(
+    app: FastAPI,
+    *,
+    service_key: ServiceKey,
+    service_version: ServiceVersion,
+) -> SimcoreServiceLabels:
+    """Get the docker image labels of a specific service version"""
+    return await catalog_services.get_catalog_service_labels(
+        get_director_client(app),
+        service_key=service_key,
+        service_version=service_version,
+    )
+
+
 @router.expose(reraise_if_error_type=(CatalogForbiddenRpcError, ValidationError))
 @_profile_rpc_call
 @validate_call(config={"arbitrary_types_allowed": True})

services/catalog/src/simcore_service_catalog/service/catalog_services.py

@@ -18,12 +18,13 @@
 from models_library.groups import GroupID
 from models_library.products import ProductName
 from models_library.rest_pagination import PageLimitInt, PageOffsetInt, PageTotalCount
+from models_library.service_settings_labels import SimcoreServiceLabels
 from models_library.services_access import ServiceGroupAccessRightsV2
 from models_library.services_history import Compatibility, ServiceRelease
 from models_library.services_metadata_published import ServiceMetaDataPublished
 from models_library.services_types import ServiceKey, ServiceVersion
 from models_library.users import UserID
-from pydantic import HttpUrl
+from pydantic import HttpUrl, TypeAdapter
 from servicelib.rabbitmq.rpc_interfaces.catalog.errors import (
     CatalogForbiddenRpcError,
     CatalogInconsistentRpcError,
@@ -818,3 +819,10 @@ async def get_catalog_service_extras(
     director_api: DirectorClient, service_key: ServiceKey, service_version: VersionStr
 ) -> ServiceExtras:
     return await director_api.get_service_extras(service_key=service_key, service_version=service_version)
+
+
+async def get_catalog_service_labels(
+    director_api: DirectorClient, service_key: ServiceKey, service_version: ServiceVersion
+) -> SimcoreServiceLabels:
+    labels = await director_api.get_service_labels(service_key=service_key, service_version=service_version)
+    return TypeAdapter(SimcoreServiceLabels).validate_python(labels)

services/catalog/tests/unit/with_dbs/test_api_rpc.py

@@ -20,6 +20,7 @@
     DEFAULT_NUMBER_OF_ITEMS_PER_PAGE,
     MAXIMUM_NUMBER_OF_ITEMS_PER_PAGE,
 )
+from models_library.service_settings_labels import SimcoreServiceLabels
 from models_library.services_enums import ServiceType
 from models_library.services_history import ServiceRelease
 from models_library.services_types import ServiceKey, ServiceVersion
@@ -737,6 +738,27 @@ async def test_rpc_get_service_ports_successful_retrieval(
     assert len(ports) == len(expected_inputs) + len(expected_outputs)
 
 
+async def test_rpc_get_service_labels(
+    background_sync_task_mocked: None,
+    mocked_director_rest_api: MockRouter,
+    rpc_client: RabbitMQRPCClient,
+    app: FastAPI,
+    expected_director_rest_api_list_services: list[dict[str, Any]],
+):
+    assert app
+
+    expected_service = expected_director_rest_api_list_services[0]
+
+    labels = await catalog_rpc.get_service_labels(
+        rpc_client,
+        service_key=expected_service["key"],
+        service_version=expected_service["version"],
+    )
+
+    assert isinstance(labels, SimcoreServiceLabels)
+    assert labels.needs_dynamic_sidecar is False
+
+
 async def test_rpc_get_service_ports_not_found(
     background_sync_task_mocked: None,
     mocked_director_rest_api: MockRouter,

services/docker-compose.yml

@@ -888,8 +888,6 @@ services:
         - *redis_settings
         - *tracing_open_telemetry_environments
 
-      CATALOG_HOST: ${CATALOG_HOST}
-      CATALOG_PORT: ${CATALOG_PORT}
       DIRECTOR_V2_HOST: ${DIRECTOR_V2_HOST}
       DIRECTOR_V2_PORT: ${DIRECTOR_V2_PORT}
 

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/core/application.py

@@ -28,7 +28,6 @@
 from ..api.frontend import configure_frontend
 from ..api.rest.routes import configure_rest_api
 from ..api.rpc.routes import configure_rpc_api
-from ..services.catalog import configure_catalog
 from ..services.deferred_manager import configure_deferred_manager
 from ..services.director_v0 import configure_director_v0
 from ..services.director_v2 import configure_director_v2
@@ -60,7 +59,6 @@ def _configure_plugins(
     configure_fire_and_forget(app_lifespan)
     configure_director_v2(app_lifespan)
     configure_director_v0(app_lifespan)
-    configure_catalog(app_lifespan)
     configure_rabbitmq_client(app_lifespan, settings=settings.DYNAMIC_SCHEDULER_RABBITMQ)
     configure_rpc_api(app_lifespan)
     configure_redis_clients(app_lifespan, settings=settings.DYNAMIC_SCHEDULER_REDIS)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/core/settings.py

@@ -8,7 +8,6 @@
 from servicelib.logging_utils import LogLevelInt
 from settings_library.application import BaseApplicationSettings
 from settings_library.basic_types import LogLevel, VersionTag
-from settings_library.catalog import CatalogSettings
 from settings_library.director_v0 import DirectorV0Settings
 from settings_library.director_v2 import DirectorV2Settings
 from settings_library.docker_api_proxy import DockerApiProxysettings
@@ -159,11 +158,6 @@ class ApplicationSettings(_BaseApplicationSettings):
         description="settings for director-v2 service",
     )
 
-    DYNAMIC_SCHEDULER_CATALOG_SETTINGS: CatalogSettings = Field(
-        json_schema_extra={"auto_default_from_env": True},
-        description="settings for catalog service",
-    )
-
     DYNAMIC_SCHEDULER_PROMETHEUS_INSTRUMENTATION_ENABLED: bool = True
 
     DYNAMIC_SCHEDULER_PROFILING: bool = False

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/common_interface.py

@@ -17,12 +17,13 @@
 from models_library.services_types import ServicePortKey
 from models_library.users import UserID
 from pydantic import NonNegativeInt
+from servicelib.rabbitmq.rpc_interfaces.catalog import services as catalog_rpc
 from servicelib.utils import fire_and_forget_task
 
 from ..core.settings import ApplicationSettings
-from .catalog._public_client import CatalogPublicClient
 from .director_v2 import DirectorV2Client
 from .fire_and_forget import FireAndForgetCollection
+from .rabbitmq import get_rabbitmq_rpc_client
 from .service_tracker import (
     get_tracked_service,
     set_request_as_running,
@@ -80,9 +81,10 @@ async def stop_dynamic_service(app: FastAPI, *, dynamic_service_stop: DynamicSer
     tracked_service = await get_tracked_service(app, dynamic_service_stop.node_id)
 
     if tracked_service and tracked_service.dynamic_service_start:
-        service_labels = await CatalogPublicClient.get_from_app_state(app).get_docker_image_labels(
-            tracked_service.dynamic_service_start.key,
-            tracked_service.dynamic_service_start.version,
+        service_labels = await catalog_rpc.get_service_labels(
+            get_rabbitmq_rpc_client(app),
+            service_key=tracked_service.dynamic_service_start.key,
+            service_version=tracked_service.dynamic_service_start.version,
         )
         if not service_labels.needs_dynamic_sidecar:
             # LEGACY services