added rpc lrt interface to dynamic-sidecar

Author: Andrei Neagu

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/dynamic_sidecar/containers_long_running_tasks.py

@@ -14,7 +14,7 @@
 
 
 @log_decorator(_logger, level=logging.DEBUG)
-async def pull_user_services_docker_images(
+async def pull_user_services_docker_images_task(
     rabbitmq_rpc_client: RabbitMQRPCClient,
     *,
     node_id: NodeID,
@@ -23,7 +23,9 @@ async def pull_user_services_docker_images(
     rpc_namespace = get_rpc_namespace(node_id)
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
-        TypeAdapter(RPCMethodName).validate_python("pull_user_services_docker_images"),
+        TypeAdapter(RPCMethodName).validate_python(
+            "pull_user_services_docker_images_task"
+        ),
         lrt_namespace=lrt_namespace,
     )
     return TaskId(result)

services/director-v2/src/simcore_service_director_v2/cli/_close_and_save_service.py

@@ -15,7 +15,7 @@
 )
 from servicelib.fastapi.http_client_thin import UnexpectedStatusError
 from servicelib.fastapi.long_running_tasks.client import (
-    Client,
+    HttpClient,
     periodic_task_result,
     setup,
 )
@@ -47,7 +47,7 @@ async def _minimal_app() -> AsyncIterator[FastAPI]:
 
 
 async def _track_and_display(
-    client: Client,
+    client: HttpClient,
     task_id: TaskId,
     update_interval: PositiveFloat,
     task_timeout: PositiveFloat,
@@ -105,7 +105,7 @@ async def async_close_and_save_service(
                     f"{node_id}", is_disabled=True
                 )
 
-        client = Client(
+        client = HttpClient(
             app=app,
             async_client=thin_dv2_localhost_client.client,
             base_url=f"{TypeAdapter(AnyHttpUrl).validate_python(thin_dv2_localhost_client.BASE_ADDRESS)}",

services/director-v2/src/simcore_service_director_v2/modules/dynamic_sidecar/api_client/_public.py

@@ -22,7 +22,10 @@
     BaseHttpClientError,
     UnexpectedStatusError,
 )
-from servicelib.fastapi.long_running_tasks.client import Client, periodic_task_result
+from servicelib.fastapi.long_running_tasks.client import (
+    HttpClient,
+    periodic_task_result,
+)
 from servicelib.logging_utils import log_context, log_decorator
 from servicelib.long_running_tasks.models import (
     ProgressCallback,
@@ -288,8 +291,8 @@ async def submit_docker_compose_spec(
             dynamic_sidecar_endpoint, compose_spec=compose_spec
         )
 
-    def _get_client(self, dynamic_sidecar_endpoint: AnyHttpUrl) -> Client:
-        return Client(
+    def _get_client(self, dynamic_sidecar_endpoint: AnyHttpUrl) -> HttpClient:
+        return HttpClient(
             app=self._app,
             async_client=self._async_client,
             base_url=f"{dynamic_sidecar_endpoint}",

services/director-v2/tests/integration/02/test_dynamic_sidecar_nodeports_integration.py

@@ -50,7 +50,10 @@
 from pytest_simcore.helpers.host import get_localhost_ip
 from pytest_simcore.helpers.monkeypatch_envs import setenvs_from_dict
 from pytest_simcore.helpers.typing_env import EnvVarsDict
-from servicelib.fastapi.long_running_tasks.client import Client, periodic_task_result
+from servicelib.fastapi.long_running_tasks.client import (
+    HttpClient,
+    periodic_task_result,
+)
 from servicelib.long_running_tasks.models import (
     ProgressMessage,
     ProgressPercent,
@@ -863,7 +866,7 @@ async def _debug_progress_callback(
         logger.debug("%s: %.2f %s", task_id, percent, message)
 
     async with periodic_task_result(
-        Client(
+        HttpClient(
             app=initialized_app,
             async_client=director_v2_client,
             base_url=TypeAdapter(AnyHttpUrl).validate_python(

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/rpc/_containers.py

@@ -0,0 +1,17 @@
+from fastapi import FastAPI
+from models_library.api_schemas_directorv2.dynamic_services import ContainersComposeSpec
+from servicelib.rabbitmq import RPCRouter
+
+from ...core.validation import InvalidComposeSpecError
+from ...services import containers
+
+router = RPCRouter()
+
+
+@router.expose(reraise_if_error_type=(InvalidComposeSpecError,))
+async def store_compose_spec(
+    app: FastAPI,
+    *,
+    containers_compose_spec: ContainersComposeSpec,
+) -> None:
+    await containers.store_compose_spec(app, containers_compose_spec)

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/rpc/_containers_long_running_tasks.py

@@ -0,0 +1,101 @@
+from fastapi import FastAPI
+from models_library.api_schemas_directorv2.dynamic_services import ContainersCreate
+from servicelib.long_running_tasks.models import LRTNamespace, TaskId
+from servicelib.rabbitmq import RPCRouter
+
+from ...core.rabbitmq import get_rabbitmq_rpc_client
+from ...services import containers_long_running_tasks
+
+router = RPCRouter()
+
+
+@router.expose()
+async def pull_user_services_docker_images_task(
+    app: FastAPI, *, lrt_namespace: LRTNamespace
+) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.pull_user_services_docker_images(
+        rpc_client, lrt_namespace
+    )
+
+
+@router.expose()
+async def create_service_containers_task(
+    app: FastAPI, *, lrt_namespace: LRTNamespace, containers_create: ContainersCreate
+) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.create_service_containers_task(
+        rpc_client, lrt_namespace, containers_create
+    )
+
+
+@router.expose()
+async def runs_docker_compose_down_task(
+    app: FastAPI, *, lrt_namespace: LRTNamespace
+) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.runs_docker_compose_down_task(
+        rpc_client, lrt_namespace
+    )
+
+
+@router.expose()
+async def state_restore_task(app: FastAPI, *, lrt_namespace: LRTNamespace) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.state_restore_task(
+        rpc_client, lrt_namespace
+    )
+
+
+@router.expose()
+async def state_save_task(app: FastAPI, *, lrt_namespace: LRTNamespace) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.state_save_task(
+        rpc_client, lrt_namespace
+    )
+
+
+@router.expose()
+async def ports_inputs_pull_task(
+    app: FastAPI,
+    *,
+    lrt_namespace: LRTNamespace,
+    port_keys: list[str] | None,
+) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.ports_inputs_pull_task(
+        rpc_client, lrt_namespace, port_keys
+    )
+
+
+@router.expose()
+async def ports_outputs_pull_task(
+    app: FastAPI,
+    *,
+    lrt_namespace: LRTNamespace,
+    port_keys: list[str] | None,
+) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.ports_outputs_pull_task(
+        rpc_client, lrt_namespace, port_keys
+    )
+
+
+@router.expose()
+async def ports_outputs_push_task(
+    app: FastAPI, *, lrt_namespace: LRTNamespace
+) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.ports_outputs_push_task(
+        rpc_client, lrt_namespace
+    )
+
+
+@router.expose()
+async def containers_restart_task(
+    app: FastAPI, *, lrt_namespace: LRTNamespace
+) -> TaskId:
+    rpc_client = get_rabbitmq_rpc_client(app)
+    return await containers_long_running_tasks.containers_restart_task(
+        rpc_client, lrt_namespace
+    )

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/api/rpc/routes.py

@@ -4,9 +4,11 @@
 
 from ...core.rabbitmq import get_rabbitmq_rpc_server
 from ...core.settings import ApplicationSettings
-from . import _disk, _disk_usage, _volumes
+from . import _containers, _containers_long_running_tasks, _disk, _disk_usage, _volumes
 
 ROUTERS: list[RPCRouter] = [
+    _containers_long_running_tasks.router,
+    _containers.router,
     _disk_usage.router,
     _disk.router,
     _volumes.router,

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/core/rabbitmq.py

@@ -121,6 +121,11 @@ def get_rabbitmq_rpc_server(app: FastAPI) -> RabbitMQRPCClient:
     return cast(RabbitMQRPCClient, app.state.rabbitmq_rpc_server)
 
 
+def get_rabbitmq_rpc_client(app: FastAPI) -> RabbitMQRPCClient:
+    _raise_if_not_initialized(app, "rabbitmq_rpc_client")
+    return cast(RabbitMQRPCClient, app.state.rabbitmq_rpc_server)
+
+
 def setup_rabbitmq(app: FastAPI) -> None:
     async def on_startup() -> None:
         app_settings: ApplicationSettings = app.state.settings
@@ -134,7 +139,12 @@ async def on_startup() -> None:
             )
         with log_context(_logger, logging.INFO, msg="Create RabbitMQRPCClient"):
             app.state.rabbitmq_rpc_server = await RabbitMQRPCClient.create(
-                client_name=f"dynamic-sidecar_rpc_{app_settings.DY_SIDECAR_NODE_ID}",
+                client_name=f"dynamic-sidecar_rpc_server_{app_settings.DY_SIDECAR_NODE_ID}",
+                settings=settings,
+            )
+        with log_context(_logger, logging.INFO, msg="Create RabbitMQRPCClient"):
+            app.state.rabbitmq_rpc_client = await RabbitMQRPCClient.create(
+                client_name=f"dynamic-sidecar_rpc_client_{app_settings.DY_SIDECAR_NODE_ID}",
                 settings=settings,
             )
 
@@ -143,6 +153,8 @@ async def on_shutdown() -> None:
             await app.state.rabbitmq_client.close()
         if app.state.rabbitmq_rpc_server:
             await app.state.rabbitmq_rpc_server.close()
+        if app.state.rabbitmq_rpc_client:
+            await app.state.rabbitmq_rpc_client.close()
 
     app.add_event_handler("startup", on_startup)
     app.add_event_handler("shutdown", on_shutdown)

services/dynamic-sidecar/src/simcore_service_dynamic_sidecar/modules/long_running_tasks.py

@@ -651,51 +651,51 @@ async def on_startup() -> None:
 
         task_context.update(
             {
-                task_pull_user_servcices_docker_images: dict(
-                    shared_store=shared_store,
-                    app=app,
-                ),
-                task_create_service_containers: dict(
-                    app=app,
-                    settings=settings,
-                    shared_store=shared_store,
-                    application_health=application_health,
-                ),
-                task_runs_docker_compose_down: dict(
-                    app=app,
-                    settings=settings,
-                    shared_store=shared_store,
-                    mounted_volumes=mounted_volumes,
-                ),
-                task_restore_state: dict(
-                    app=app,
-                    settings=settings,
-                    mounted_volumes=mounted_volumes,
-                ),
-                task_save_state: dict(
-                    app=app,
-                    settings=settings,
-                    mounted_volumes=mounted_volumes,
-                ),
-                task_ports_inputs_pull: dict(
-                    app=app,
-                    settings=settings,
-                    mounted_volumes=mounted_volumes,
-                    inputs_state=inputs_state,
-                ),
-                task_ports_outputs_pull: dict(
-                    app=app,
-                    mounted_volumes=mounted_volumes,
-                ),
-                task_ports_outputs_push: dict(
-                    app=app,
-                    outputs_manager=outputs_manager,
-                ),
-                task_containers_restart: dict(
-                    app=app,
-                    settings=settings,
-                    shared_store=shared_store,
-                ),
+                task_pull_user_servcices_docker_images: {
+                    "shared_store": shared_store,
+                    "app": app,
+                },
+                task_create_service_containers: {
+                    "app": app,
+                    "settings": settings,
+                    "shared_store": shared_store,
+                    "application_health": application_health,
+                },
+                task_runs_docker_compose_down: {
+                    "app": app,
+                    "settings": settings,
+                    "shared_store": shared_store,
+                    "mounted_volumes": mounted_volumes,
+                },
+                task_restore_state: {
+                    "app": app,
+                    "settings": settings,
+                    "mounted_volumes": mounted_volumes,
+                },
+                task_save_state: {
+                    "app": app,
+                    "settings": settings,
+                    "mounted_volumes": mounted_volumes,
+                },
+                task_ports_inputs_pull: {
+                    "app": app,
+                    "settings": settings,
+                    "mounted_volumes": mounted_volumes,
+                    "inputs_state": inputs_state,
+                },
+                task_ports_outputs_pull: {
+                    "app": app,
+                    "mounted_volumes": mounted_volumes,
+                },
+                task_ports_outputs_push: {
+                    "app": app,
+                    "outputs_manager": outputs_manager,
+                },
+                task_containers_restart: {
+                    "app": app,
+                    "settings": settings,
+                    "shared_store": shared_store,
+                },
             }
         )