rerouted restart user services

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

@@ -93,3 +93,19 @@ async def stop_dynamic_service(
         timeout_s=timeout_s,
     )
     assert result is None  # nosec
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def restart_user_services(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    timeout_s: NonNegativeInt,
+) -> None:
+    result = await rabbitmq_rpc_client.request(
+        DYNAMIC_SCHEDULER_RPC_NAMESPACE,
+        _RPC_METHOD_NAME_ADAPTER.validate_python("restart_user_services"),
+        node_id=node_id,
+        timeout_s=timeout_s,
+    )
+    assert result is None  # nosec

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

@@ -56,3 +56,8 @@ async def stop_dynamic_service(
     return await scheduler_interface.stop_dynamic_service(
         app, dynamic_service_stop=dynamic_service_stop
     )
+
+
+@router.expose()
+async def restart_user_services(app: FastAPI, *, node_id: NodeID) -> None:
+    await scheduler_interface.restart_user_services(app, node_id=node_id)

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

@@ -108,6 +108,9 @@ async def list_tracked_dynamic_services(
         )
         return TypeAdapter(list[DynamicServiceGet]).validate_python(response.json())
 
+    async def restart_user_services(self, *, node_id: NodeID) -> None:
+        await self.thin_client.post_restart(node_id=node_id)
+
 
 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

@@ -124,3 +124,8 @@ async def get_dynamic_services(
             "/dynamic_services",
             params=as_dict_exclude_unset(user_id=user_id, project_id=project_id),
         )
+
+    @retry_on_errors()
+    @expect_status(status.HTTP_204_NO_CONTENT)
+    async def post_restart(self, *, node_id: NodeID) -> Response:
+        return await self.client.post(f"/dynamic_services/{node_id}:restart")

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

@@ -73,3 +73,12 @@ async def stop_dynamic_service(
     )
 
     await set_request_as_stopped(app, dynamic_service_stop)
+
+
+async def restart_user_services(app: FastAPI, *, node_id: NodeID) -> None:
+    settings: ApplicationSettings = app.state.settings
+    if settings.DYNAMIC_SCHEDULER_USE_INTERNAL_SCHEDULER:
+        raise NotImplementedError
+
+    director_v2_client = DirectorV2Client.get_from_app_state(app)
+    await director_v2_client.restart_user_services(node_id=node_id)

services/dynamic-scheduler/tests/unit/api_rpc/test_api_rpc__services.py

@@ -490,3 +490,25 @@ async def test_stop_dynamic_service_serializes_generic_errors(
             ),
             timeout_s=5,
         )
+
+
+@pytest.fixture
+def mock_director_v2_restart_user_services(node_id: NodeID) -> Iterator[None]:
+    with respx.mock(
+        base_url="http://director-v2:8000/v2",
+        assert_all_called=False,
+        assert_all_mocked=True,  # IMPORTANT: KEEP always True!
+    ) as mock:
+        mock.post(f"/dynamic_services/{node_id}:restart").respond(
+            status.HTTP_204_NO_CONTENT
+        )
+
+        yield None
+
+
+async def test_restart_user_services(
+    mock_director_v2_restart_user_services: None,
+    rpc_client: RabbitMQRPCClient,
+    node_id: NodeID,
+):
+    await services.restart_user_services(rpc_client, node_id=node_id, timeout_s=5)

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

@@ -65,24 +65,6 @@ async def request_retrieve_dyn_service(
         )
 
 
-@log_decorator(logger=_log)
-async def restart_dynamic_service(app: web.Application, node_uuid: str) -> None:
-    """Restarts the user service(s) started by the the node_uuid's sidecar
-
-    NOTE: this operation will NOT restart
-    sidecar services (``dy-sidecar`` or ``dy-proxy`` services),
-    but ONLY user services (the ones defined by the compose spec).
-    """
-    settings: DirectorV2Settings = get_plugin_settings(app)
-    await request_director_v2(
-        app,
-        "POST",
-        url=settings.base_url / f"dynamic_services/{node_uuid}:restart",
-        expected_status=web.HTTPOk,
-        timeout=settings.DIRECTOR_V2_RESTART_DYNAMIC_SERVICE_TIMEOUT,
-    )
-
-
 @log_decorator(logger=_log)
 async def update_dynamic_service_networks_in_project(
     app: web.Application, project_id: ProjectID

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

@@ -19,7 +19,6 @@
 from ._core_dynamic_services import (
     get_project_inactivity,
     request_retrieve_dyn_service,
-    restart_dynamic_service,
     retrieve,
     update_dynamic_service_networks_in_project,
 )
@@ -39,7 +38,6 @@
     "is_healthy",
     "is_pipeline_running",
     "request_retrieve_dyn_service",
-    "restart_dynamic_service",
     "retrieve",
     "set_project_run_policy",
     "stop_pipeline",

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

@@ -33,14 +33,6 @@ def base_url(self) -> URL:
     # - Mostly in floats (aiohttp.Client/) but sometimes in ints
     # - Typically in seconds but occasionally in ms
 
-    DIRECTOR_V2_RESTART_DYNAMIC_SERVICE_TIMEOUT: PositiveInt = Field(
-        1 * _MINUTE,
-        description="timeout of containers restart",
-        validation_alias=AliasChoices(
-            "DIRECTOR_V2_RESTART_DYNAMIC_SERVICE_TIMEOUT",
-        ),
-    )
-
     DIRECTOR_V2_STORAGE_SERVICE_UPLOAD_DOWNLOAD_TIMEOUT: PositiveInt = Field(
         _HOUR,
         description=(

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

@@ -148,3 +148,20 @@ async def stop_dynamic_services_in_project(
         ]
 
         await logged_gather(*services_to_stop)
+
+
+async def restart_user_services(app: web.Application, *, node_id: NodeID) -> None:
+    """Restarts the user service(s) started by the the node_uuid's sidecar
+
+    NOTE: this operation will NOT restart
+    sidecar services (``dy-sidecar`` or ``dy-proxy`` services),
+    but ONLY user services (the ones defined by the compose spec).
+    """
+    settings: DynamicSchedulerSettings = get_plugin_settings(app)
+    await services.restart_user_services(
+        get_rabbitmq_rpc_client(app),
+        node_id=node_id,
+        timeout_s=int(
+            settings.DYNAMIC_SCHEDULER_RESTART_DYNAMIC_SERVICE_TIMEOUT.total_seconds()
+        ),
+    )

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

@@ -26,6 +26,14 @@ class DynamicSchedulerSettings(BaseCustomSettings, MixinServiceSettings):
         ),
     )
 
+    DYNAMIC_SCHEDULER_RESTART_DYNAMIC_SERVICE_TIMEOUT: datetime.timedelta = Field(
+        datetime.timedelta(minutes=1),
+        description="timeout of containers restart",
+        validation_alias=AliasChoices(
+            "DIRECTOR_V2_RESTART_DYNAMIC_SERVICE_TIMEOUT",
+        ),
+    )
+
 
 def get_plugin_settings(app: web.Application) -> DynamicSchedulerSettings:
     settings = app[APP_SETTINGS_KEY].WEBSERVER_DYNAMIC_SCHEDULER

services/web/server/src/simcore_service_webserver/projects/_nodes_handlers.py

@@ -412,7 +412,9 @@ async def restart_node(request: web.Request) -> web.Response:
 
     path_params = parse_request_path_parameters_as(NodePathParams, request)
 
-    await director_v2_api.restart_dynamic_service(request.app, f"{path_params.node_id}")
+    await dynamic_scheduler_api.restart_user_services(
+        request.app, node_id=path_params.node_id
+    )
 
     return web.json_response(status=status.HTTP_204_NO_CONTENT)