Merge branch 'master' into 1973-add-celery-worker-to-api-server

Author: bisgaard-itis

.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/long_running_tasks/_serialization.py

@@ -78,4 +78,6 @@ def loads(obj_str: str) -> Any:
             msg = f"Could not reconstruct object from data: {data}"
             raise ValueError(msg) from e
 
+    if isinstance(data, Exception):
+        raise data
     return data

packages/service-library/src/servicelib/long_running_tasks/errors.py

@@ -34,6 +34,14 @@ class TaskExceptionError(BaseLongRunningError):
     )
 
 
+class TaskRaisedUnserializableError(BaseLongRunningError):
+    msg_template: str = (
+        "Task {task_id} raised an exception that could not be serialized.\n"
+        "Original exception: '{original_exception_str}'\n"
+        "As a consequence, the following error was raised: '{exception}'"
+    )
+
+
 class TaskClientTimeoutError(BaseLongRunningError):
     msg_template: str = (
         "Timed out after {timeout} seconds while awaiting '{task_id}' to complete"

packages/service-library/src/servicelib/long_running_tasks/task.py

@@ -32,6 +32,7 @@
     TaskNotCompletedError,
     TaskNotFoundError,
     TaskNotRegisteredError,
+    TaskRaisedUnserializableError,
 )
 from .models import (
     LRTNamespace,
@@ -353,7 +354,30 @@ async def _tasks_monitor(self) -> None:
                                 },
                             ),
                         )
-                    result_field = ResultField(str_error=dumps(e))
+                    try:
+                        result_field = ResultField(str_error=dumps(e))
+                    except (
+                        Exception  # pylint:disable=broad-except
+                    ) as serialization_error:
+                        _logger.exception(
+                            **create_troubleshootting_log_kwargs(
+                                (
+                                    f"Execution of {task_id=} finished with an error "
+                                    f"which could not be serialized"
+                                ),
+                                error=serialization_error,
+                                tip="Check the error above for more details",
+                            ),
+                        )
+                        result_field = ResultField(
+                            str_error=dumps(
+                                TaskRaisedUnserializableError(
+                                    task_id=task_id,
+                                    exception=serialization_error,
+                                    original_exception_str=f"{e}",
+                                )
+                            )
+                        )
 
                 # update and store in Redis
                 updates = {"is_done": is_done, "result_field": task_data.result_field}

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