rerouted retrieve via dynamic-scheduler

Author: Andrei Neagu

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

@@ -1,7 +1,10 @@
 import logging
 from typing import Final
 
-from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceGet
+from models_library.api_schemas_directorv2.dynamic_services import (
+    DynamicServiceGet,
+    RetrieveDataOutEnveloped,
+)
 from models_library.api_schemas_dynamic_scheduler import DYNAMIC_SCHEDULER_RPC_NAMESPACE
 from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
     DynamicServiceStart,
@@ -10,6 +13,7 @@
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
 from models_library.projects_nodes_io import NodeID
 from models_library.rabbitmq_basic_types import RPCMethodName
+from models_library.services_types import ServicePortKey
 from pydantic import NonNegativeInt, TypeAdapter
 from servicelib.logging_utils import log_decorator
 from servicelib.rabbitmq import RabbitMQRPCClient
@@ -73,3 +77,22 @@ async def stop_dynamic_service(
         timeout_s=timeout_s,
     )
     assert result is None  # nosec
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def retrieve_data_on_ports(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    port_keys: list[ServicePortKey],
+    timeout_s: NonNegativeInt,
+) -> RetrieveDataOutEnveloped:
+    result = await rabbitmq_rpc_client.request(
+        DYNAMIC_SCHEDULER_RPC_NAMESPACE,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("retrieve_data_on_ports"),
+        node_id=node_id,
+        port_keys=port_keys,
+        timeout_s=timeout_s,
+    )
+    assert isinstance(result, RetrieveDataOutEnveloped)  # nosec
+    return result

services/docker-compose.yml

@@ -568,6 +568,7 @@ services:
       DYNAMIC_SCHEDULER_PROFILING: ${DYNAMIC_SCHEDULER_PROFILING}
       DYNAMIC_SCHEDULER_TRACING: ${DYNAMIC_SCHEDULER_TRACING}
       DYNAMIC_SCHEDULER_UI_STORAGE_SECRET: ${DYNAMIC_SCHEDULER_UI_STORAGE_SECRET}
+      DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT: ${DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT}
       TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT: ${TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT}
       TRACING_OPENTELEMETRY_COLLECTOR_PORT: ${TRACING_OPENTELEMETRY_COLLECTOR_PORT}
   static-webserver:

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/api/rpc/_services.py

@@ -1,11 +1,15 @@
 from fastapi import FastAPI
-from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceGet
+from models_library.api_schemas_directorv2.dynamic_services import (
+    DynamicServiceGet,
+    RetrieveDataOutEnveloped,
+)
 from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
     DynamicServiceStart,
     DynamicServiceStop,
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
 from models_library.projects_nodes_io import NodeID
+from models_library.services_types import ServicePortKey
 from servicelib.rabbitmq import RPCRouter
 from servicelib.rabbitmq.rpc_interfaces.dynamic_scheduler.errors import (
     ServiceWaitingForManualInterventionError,
@@ -45,3 +49,12 @@ async def stop_dynamic_service(
     return await scheduler_interface.stop_dynamic_service(
         app, dynamic_service_stop=dynamic_service_stop
     )
+
+
+@router.expose()
+async def retrieve_data_on_ports(
+    app: FastAPI, *, node_id: NodeID, port_keys: list[ServicePortKey]
+) -> RetrieveDataOutEnveloped:
+    return await scheduler_interface.retrieve_data_on_ports(
+        app, node_id=node_id, port_keys=port_keys
+    )

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

@@ -62,12 +62,26 @@ class _BaseApplicationSettings(BaseApplicationSettings, MixinLoggingSettings):
 
     DYNAMIC_SCHEDULER_STOP_SERVICE_TIMEOUT: datetime.timedelta = Field(
         default=datetime.timedelta(minutes=60),
+        validation_alias=AliasChoices(
+            "DYNAMIC_SIDECAR_API_SAVE_RESTORE_STATE_TIMEOUT",
+            "DYNAMIC_SCHEDULER_STOP_SERVICE_TIMEOUT",
+        ),
         description=(
             "Time to wait before timing out when stopping a dynamic service. "
             "Since services require data to be stopped, this operation is timed out after 1 hour"
         ),
     )
 
+    DYNAMIC_SCHEDULER_SERVICE_UPLOAD_DOWNLOAD_TIMEOUT: datetime.timedelta = Field(
+        default=datetime.timedelta(minutes=60),
+        description=(
+            "When dynamic services upload and download data from storage, "
+            "sometimes very big payloads are involved. In order to handle "
+            "such payloads it is required to have long timeouts which "
+            "allow the service to finish the operation."
+        ),
+    )
+
     DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER: bool = Field(
         default=False,
         description=(

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/director_v2/_public_client.py

@@ -2,12 +2,16 @@
 from typing import Any
 
 from fastapi import FastAPI, status
-from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceGet
+from models_library.api_schemas_directorv2.dynamic_services import (
+    DynamicServiceGet,
+    RetrieveDataOutEnveloped,
+)
 from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
     DynamicServiceStart,
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
 from models_library.projects_nodes_io import NodeID
+from models_library.services_types import ServicePortKey
 from pydantic import TypeAdapter
 from servicelib.fastapi.app_state import SingletonInAppStateMixin
 from servicelib.fastapi.http_client import AttachLifespanMixin, HasClientSetupInterface
@@ -73,7 +77,7 @@ async def stop_dynamic_service(
         node_id: NodeID,
         simcore_user_agent: str,
         save_state: bool,
-        timeout: datetime.timedelta
+        timeout: datetime.timedelta  # noqa: ASYNC109
     ) -> None:
         try:
             await self.thin_client.delete_dynamic_service(
@@ -98,6 +102,19 @@ async def stop_dynamic_service(
 
             raise
 
+    async def retrieve_data_on_ports(
+        self,
+        *,
+        node_id: NodeID,
+        port_keys: list[ServicePortKey],
+        timeout: datetime.timedelta  # noqa: ASYNC109
+    ) -> RetrieveDataOutEnveloped:
+        response = await self.thin_client.dynamic_service_retrieve(
+            node_id=node_id, port_keys=port_keys, timeout=timeout
+        )
+        dict_response: dict[str, Any] = response.json()
+        return TypeAdapter(RetrieveDataOutEnveloped).validate_python(dict_response)
+
 
 def setup_director_v2(app: FastAPI) -> None:
     public_client = DirectorV2Client(app)

services/dynamic-scheduler/src/simcore_service_dynamic_scheduler/services/director_v2/_thin_client.py

@@ -9,6 +9,7 @@
 )
 from models_library.projects_nodes_io import NodeID
 from models_library.services_resources import ServiceResourcesDictHelpers
+from models_library.services_types import ServicePortKey
 from servicelib.common_headers import (
     X_DYNAMIC_SIDECAR_REQUEST_DNS,
     X_DYNAMIC_SIDECAR_REQUEST_SCHEME,
@@ -88,7 +89,7 @@ async def delete_dynamic_service(
         node_id: NodeID,
         simcore_user_agent: str,
         save_state: bool,
-        timeout: datetime.timedelta,
+        timeout: datetime.timedelta,  # noqa: ASYNC109
     ) -> Response:
         @retry_on_errors(total_retry_timeout_overwrite=timeout.total_seconds())
         @expect_status(status.HTTP_204_NO_CONTENT)
@@ -108,3 +109,17 @@ async def _(
             )
 
         return await _(self)
+
+    async def dynamic_service_retrieve(
+        self,
+        *,
+        node_id: NodeID,
+        port_keys: list[ServicePortKey],
+        timeout: datetime.timedelta,  # noqa: ASYNC109
+    ) -> Response:
+        post_data = {"port_keys": port_keys}
+        return await self.client.post(
+            f"/dynamic_services/{node_id}:retrieve",
+            content=json_dumps(post_data),
+            timeout=timeout.total_seconds(),
+        )

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

@@ -1,11 +1,15 @@
 from fastapi import FastAPI
-from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceGet
+from models_library.api_schemas_directorv2.dynamic_services import (
+    DynamicServiceGet,
+    RetrieveDataOutEnveloped,
+)
 from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
     DynamicServiceStart,
     DynamicServiceStop,
 )
 from models_library.api_schemas_webserver.projects_nodes import NodeGet, NodeGetIdle
 from models_library.projects_nodes_io import NodeID
+from models_library.services_types import ServicePortKey
 
 from ..core.settings import ApplicationSettings
 from .director_v2 import DirectorV2Client
@@ -58,3 +62,18 @@ async def stop_dynamic_service(
     )
 
     await set_request_as_stopped(app, dynamic_service_stop)
+
+
+async def retrieve_data_on_ports(
+    app: FastAPI, *, node_id: NodeID, port_keys: list[ServicePortKey]
+) -> RetrieveDataOutEnveloped:
+    settings: ApplicationSettings = app.state.settings
+    if settings.DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER:
+        raise NotImplementedError
+
+    director_v2_client = DirectorV2Client.get_from_app_state(app)
+    return await director_v2_client.retrieve_data_on_ports(
+        node_id=node_id,
+        port_keys=port_keys,
+        timeout=settings.DYNAMIC_SCHEDULER_SERVICE_UPLOAD_DOWNLOAD_TIMEOUT,
+    )

services/web/server/src/simcore_service_webserver/director_v2/_core_dynamic_services.py

@@ -9,14 +9,12 @@
 from aiohttp import web
 from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceGet
 from models_library.projects import ProjectID
-from models_library.services import ServicePortKey
 from pydantic import BaseModel, NonNegativeInt, TypeAdapter
 from pydantic.types import PositiveInt
 from servicelib.logging_utils import log_decorator
 from yarl import URL
 
 from ._core_base import DataType, request_director_v2
-from .exceptions import DirectorServiceError
 from .settings import DirectorV2Settings, get_plugin_settings
 
 _log = logging.getLogger(__name__)
@@ -52,51 +50,6 @@ async def list_dynamic_services(
     return TypeAdapter(list[DynamicServiceGet]).validate_python(services)
 
 
-# NOTE: ANE https://github.com/ITISFoundation/osparc-simcore/issues/3191
-@log_decorator(logger=_log)
-async def retrieve(
-    app: web.Application, service_uuid: str, port_keys: list[ServicePortKey]
-) -> DataType:
-    """Pulls data from connections to the dynamic service inputs"""
-    settings: DirectorV2Settings = get_plugin_settings(app)
-    result = await request_director_v2(
-        app,
-        "POST",
-        url=settings.base_url / f"dynamic_services/{service_uuid}:retrieve",
-        data={"port_keys": port_keys},
-        timeout=settings.get_service_retrieve_timeout(),
-    )
-    assert isinstance(result, dict)  # nosec
-    return result
-
-
-# NOTE: ANE https://github.com/ITISFoundation/osparc-simcore/issues/3191
-# notice that this function is identical to retrieve except that it does NOT raises
-@log_decorator(logger=_log)
-async def request_retrieve_dyn_service(
-    app: web.Application, service_uuid: str, port_keys: list[str]
-) -> None:
-    settings: DirectorV2Settings = get_plugin_settings(app)
-    body = {"port_keys": port_keys}
-
-    try:
-        await request_director_v2(
-            app,
-            "POST",
-            url=settings.base_url / f"dynamic_services/{service_uuid}:retrieve",
-            data=body,
-            timeout=settings.get_service_retrieve_timeout(),
-        )
-    except DirectorServiceError as exc:
-        _log.warning(
-            "Unable to call :retrieve endpoint on service %s, keys: [%s]: error: [%s:%s]",
-            service_uuid,
-            port_keys,
-            exc.status,
-            exc.reason,
-        )
-
-
 @log_decorator(logger=_log)
 async def restart_dynamic_service(app: web.Application, node_uuid: str) -> None:
     """User restart the dynamic dynamic service started in the node_uuid

services/web/server/src/simcore_service_webserver/director_v2/api.py

@@ -19,9 +19,7 @@
 from ._core_dynamic_services import (
     get_project_inactivity,
     list_dynamic_services,
-    request_retrieve_dyn_service,
     restart_dynamic_service,
-    retrieve,
     update_dynamic_service_networks_in_project,
 )
 from ._core_utils import is_healthy
@@ -40,9 +38,7 @@
     "is_healthy",
     "is_pipeline_running",
     "list_dynamic_services",
-    "request_retrieve_dyn_service",
     "restart_dynamic_service",
-    "retrieve",
     "set_project_run_policy",
     "stop_pipeline",
     "update_dynamic_service_networks_in_project",

services/web/server/src/simcore_service_webserver/dynamic_scheduler/api.py

@@ -3,7 +3,10 @@
 from functools import partial
 
 from aiohttp import web
-from models_library.api_schemas_directorv2.dynamic_services import DynamicServiceGet
+from models_library.api_schemas_directorv2.dynamic_services import (
+    DynamicServiceGet,
+    RetrieveDataOutEnveloped,
+)
 from models_library.api_schemas_dynamic_scheduler.dynamic_services import (
     DynamicServiceStart,
     DynamicServiceStop,
@@ -18,6 +21,7 @@
 from models_library.projects import ProjectID
 from models_library.projects_nodes_io import NodeID
 from models_library.rabbitmq_messages import ProgressRabbitMessageProject, ProgressType
+from models_library.services import ServicePortKey
 from pydantic.types import PositiveInt
 from servicelib.progress_bar import ProgressBarData
 from servicelib.rabbitmq import RabbitMQClient, RPCServerError
@@ -133,3 +137,18 @@ async def stop_dynamic_services_in_project(
         ]
 
         await logged_gather(*services_to_stop)
+
+
+# NOTE: ANE https://github.com/ITISFoundation/osparc-simcore/issues/3191
+async def retrieve(
+    app: web.Application, node_id: NodeID, port_keys: list[ServicePortKey]
+) -> RetrieveDataOutEnveloped:
+    settings: DynamicSchedulerSettings = get_plugin_settings(app)
+    return await services.retrieve_data_on_ports(
+        get_rabbitmq_rpc_client(app),
+        node_id=node_id,
+        port_keys=port_keys,
+        timeout_s=int(
+            settings.DYNAMIC_SCHEDULER_SERVICE_UPLOAD_DOWNLOAD_TIMEOUT.total_seconds()
+        ),
+    )