✨ Add long running rpc interface to dynamic-sidecar (#8255)

Co-authored-by: Andrei Neagu <[email protected]>

Author: GitHK

.github/workflows/ci-testing-deploy.yml

@@ -1057,7 +1057,7 @@ jobs:
   unit-test-dynamic-sidecar:
     needs: changes
     if: ${{ needs.changes.outputs.dynamic-sidecar == 'true' || github.event_name == 'push' || github.event.inputs.force_all_builds == 'true' }}
-    timeout-minutes: 18 # if this timeout gets too small, then split the tests
+    timeout-minutes: 19 # if this timeout gets too small, then split the tests
     name: "[unit] dynamic-sidecar"
     runs-on: ${{ matrix.os }}
     strategy:

packages/models-library/src/models_library/api_schemas_directorv2/dynamic_services.py

@@ -5,6 +5,7 @@
 
 from ..resource_tracker import HardwareInfo, PricingInfo
 from ..services import ServicePortKey
+from ..services_creation import CreateServiceMetricsAdditionalParams
 from ..services_resources import ServiceResourcesDict, ServiceResourcesDictHelpers
 from ..wallets import WalletInfo
 from .dynamic_services_service import RunningDynamicServiceDetails, ServiceDetails
@@ -104,3 +105,11 @@ class GetProjectInactivityResponse(BaseModel):
     is_inactive: bool
 
     model_config = ConfigDict(json_schema_extra={"example": {"is_inactive": "false"}})
+
+
+class ContainersComposeSpec(BaseModel):
+    docker_compose_yaml: str
+
+
+class ContainersCreate(BaseModel):
+    metrics_params: CreateServiceMetricsAdditionalParams

packages/service-library/src/servicelib/fastapi/long_running_tasks/_client.py

@@ -88,7 +88,7 @@ def retry_on_http_errors(
     assert asyncio.iscoroutinefunction(request_func)
 
     @functools.wraps(request_func)
-    async def request_wrapper(zelf: "Client", *args, **kwargs) -> Any:
+    async def request_wrapper(zelf: "HttpClient", *args, **kwargs) -> Any:
         async for attempt in AsyncRetrying(
             stop=stop_after_attempt(max_attempt_number=3),
             wait=wait_exponential(min=1),
@@ -106,7 +106,7 @@ async def request_wrapper(zelf: "Client", *args, **kwargs) -> Any:
     return request_wrapper
 
 
-class Client:
+class HttpClient:
     """
     This is a client that aims to simplify the requests to get the
     status, result and/or cancel of a long running task.

packages/service-library/src/servicelib/fastapi/long_running_tasks/_context_manager.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import warnings
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from typing import Any, Final
@@ -15,7 +16,7 @@
     TaskId,
     TaskStatus,
 )
-from ._client import Client
+from ._client import HttpClient
 
 _logger = logging.getLogger(__name__)
 
@@ -69,7 +70,7 @@ async def update(
 
 @asynccontextmanager
 async def periodic_task_result(
-    client: Client,
+    client: HttpClient,
     task_id: TaskId,
     *,
     task_timeout: PositiveFloat,
@@ -95,6 +96,13 @@ async def periodic_task_result(
     raises: `asyncio.TimeoutError` NOTE: the remote task will also be removed
     """
 
+    warnings.warn(
+        "This context manager is deprecated and will be removed in future releases. "
+        "Please use the `servicelib.long_running_tasks.lrt_api` instead.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
+
     progress_manager = _ProgressManager(progress_callback)
 
     async def _status_update() -> TaskStatus:

packages/service-library/src/servicelib/fastapi/long_running_tasks/client.py

@@ -2,11 +2,11 @@
 Provides a convenient way to return the result given a TaskId.
 """
 
-from ._client import Client, setup
+from ._client import HttpClient, setup
 from ._context_manager import periodic_task_result  # attach to the same object!
 
 __all__: tuple[str, ...] = (
-    "Client",
+    "HttpClient",
     "periodic_task_result",
     "setup",
 )

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

@@ -0,0 +1,28 @@
+import logging
+
+from models_library.api_schemas_directorv2.dynamic_services import ContainersComposeSpec
+from models_library.projects_nodes_io import NodeID
+from models_library.rabbitmq_basic_types import RPCMethodName
+from pydantic import TypeAdapter
+
+from ....logging_utils import log_decorator
+from ... import RabbitMQRPCClient
+from ._utils import get_rpc_namespace
+
+_logger = logging.getLogger(__name__)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def create_compose_spec(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    containers_compose_spec: ContainersComposeSpec,
+) -> None:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("create_compose_spec"),
+        containers_compose_spec=containers_compose_spec,
+    )
+    assert result is None  # nosec

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

@@ -0,0 +1,172 @@
+import logging
+
+from models_library.api_schemas_directorv2.dynamic_services import ContainersCreate
+from models_library.projects_nodes_io import NodeID
+from models_library.rabbitmq_basic_types import RPCMethodName
+from pydantic import TypeAdapter
+from servicelib.long_running_tasks.models import LRTNamespace, TaskId
+
+from ....logging_utils import log_decorator
+from ... import RabbitMQRPCClient
+from ._utils import get_rpc_namespace
+
+_logger = logging.getLogger(__name__)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def pull_user_services_images(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("pull_user_services_images"),
+        lrt_namespace=lrt_namespace,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def create_user_services(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+    containers_create: ContainersCreate,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("create_user_services"),
+        lrt_namespace=lrt_namespace,
+        containers_create=containers_create,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def remove_user_services(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("remove_user_services"),
+        lrt_namespace=lrt_namespace,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def restore_user_services_state_paths(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("restore_user_services_state_paths"),
+        lrt_namespace=lrt_namespace,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def save_user_services_state_paths(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("save_user_services_state_paths"),
+        lrt_namespace=lrt_namespace,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def pull_user_services_input_ports(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+    port_keys: list[str] | None,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("pull_user_services_input_ports"),
+        lrt_namespace=lrt_namespace,
+        port_keys=port_keys,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def pull_user_services_output_ports(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+    port_keys: list[str] | None,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("pull_user_services_output_ports"),
+        lrt_namespace=lrt_namespace,
+        port_keys=port_keys,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def push_user_services_output_ports(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("push_user_services_output_ports"),
+        lrt_namespace=lrt_namespace,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def restart_user_services(
+    rabbitmq_rpc_client: RabbitMQRPCClient,
+    *,
+    node_id: NodeID,
+    lrt_namespace: LRTNamespace,
+) -> TaskId:
+    rpc_namespace = get_rpc_namespace(node_id)
+    result = await rabbitmq_rpc_client.request(
+        rpc_namespace,
+        TypeAdapter(RPCMethodName).validate_python("restart_user_services"),
+        lrt_namespace=lrt_namespace,
+    )
+    assert isinstance(result, TaskId)  # nosec
+    return result

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

@@ -13,7 +13,7 @@
 
 
 @log_decorator(_logger, level=logging.DEBUG)
-async def save_volume_state(
+async def update_volume_status(
     rabbitmq_rpc_client: RabbitMQRPCClient,
     *,
     node_id: NodeID,
@@ -23,7 +23,7 @@ async def save_volume_state(
     rpc_namespace = get_rpc_namespace(node_id)
     result = await rabbitmq_rpc_client.request(
         rpc_namespace,
-        TypeAdapter(RPCMethodName).validate_python("save_volume_state"),
+        TypeAdapter(RPCMethodName).validate_python("update_volume_status"),
         status=status,
         category=category,
     )