merge branch 'pr-osparc-long-running-refactor-4' of github.com:GitHK/osparc-simcore-forked into pr-osparc-long-running-refactor-4

Author: Andrei Neagu

.env-devel

@@ -395,6 +395,7 @@ WEBSERVER_PROJECTS={}
 WEBSERVER_PROMETHEUS_API_VERSION=v1
 WEBSERVER_PROMETHEUS_URL=http://prometheus:9090
 WEBSERVER_PUBLICATIONS=1
+WEBSERVER_REALTIME_COLLABORATION='{"RTC_MAX_NUMBER_OF_USERS":3}'
 WEBSERVER_SCICRUNCH={}
 WEBSERVER_SESSION_SECRET_KEY='REPLACE_ME_with_result__Fernet_generate_key='
 WEBSERVER_SOCKETIO=1

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

@@ -13,7 +13,7 @@
 
 from ..basic_types import IDStr
 from ..projects import ProjectID
-from ..projects_nodes_io import NodeID
+from ..projects_nodes_io import NodeID, SimcoreS3FileID
 from ..projects_pipeline import ComputationTask
 from ..users import UserID
 from ..wallets import WalletInfo
@@ -126,6 +126,30 @@ class TaskLogFileGet(BaseModel):
     ] = None
 
 
+class TaskLogFileIdGet(BaseModel):
+    task_id: NodeID
+    file_id: SimcoreS3FileID | None
+
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {
+                    "task_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+                    "file_id": "1c46752c-b096-11ea-a3c4-02420a00392e/3fa85f64-5717-4562-b3fc-2c963f66afa6/logs/task_logs.txt",
+                },
+                {
+                    "task_id": "6ba7b810-9dad-11d1-80b4-00c04fd430c8",
+                    "file_id": "1c46752c-b096-11ea-a3c4-02420a00392e/6ba7b810-9dad-11d1-80b4-00c04fd430c8/logs/debug.log",
+                },
+                {
+                    "task_id": "6ba7b811-9dad-11d1-80b4-00c04fd430c8",
+                    "file_id": None,
+                },
+            ]
+        }
+    )
+
+
 class TasksSelection(BaseModel):
     nodes_ids: list[NodeID]
 

packages/models-library/src/models_library/api_schemas_rpc_async_jobs/async_jobs.py

@@ -30,6 +30,17 @@ class AsyncJobResult(BaseModel):
 
 
 class AsyncJobGet(BaseModel):
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {
+                    "job_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6",
+                    "job_name": "export_data_task",
+                }
+            ]
+        }
+    )
+
     job_id: AsyncJobId
     job_name: AsyncJobName
 
@@ -42,6 +53,18 @@ class AsyncJobAbort(BaseModel):
 class AsyncJobFilter(AsyncJobFilterBase):
     """Data for controlling access to an async job"""
 
+    model_config = ConfigDict(
+        json_schema_extra={
+            "examples": [
+                {
+                    "product_name": "osparc",
+                    "user_id": 123,
+                    "client_name": "web_client",
+                }
+            ]
+        },
+    )
+
     product_name: ProductName
     user_id: UserID
     client_name: Annotated[

packages/postgres-database/docker/Dockerfile

@@ -55,11 +55,5 @@ COPY entrypoint.bash /home/entrypoint.bash
 
 RUN chmod +x /home/entrypoint.bash
 
-ENV POSTGRES_USER=scu \
-  POSTGRES_PASSWORD=adminadmin \
-  POSTGRES_HOST=postgres \
-  POSTGRES_PORT=5432 \
-  POSTGRES_DB=simcoredb
-
 ENTRYPOINT [ "/bin/bash", "/home/entrypoint.bash" ]
 CMD [ "sc-pg", "upgrade" ]

packages/pytest-simcore/src/pytest_simcore/helpers/director_v2_rpc_server.py

@@ -0,0 +1,30 @@
+# pylint: disable=no-self-use
+# pylint: disable=not-context-manager
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+
+from models_library.api_schemas_directorv2.computations import TaskLogFileIdGet
+from models_library.projects import ProjectID
+from pydantic import TypeAdapter, validate_call
+from pytest_mock import MockType
+from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
+
+
+class DirectorV2SideEffects:
+    # pylint: disable=no-self-use
+    @validate_call(config={"arbitrary_types_allowed": True})
+    async def get_computation_task_log_file_ids(
+        self,
+        rpc_client: RabbitMQRPCClient | MockType,
+        *,
+        project_id: ProjectID,
+    ) -> list[TaskLogFileIdGet]:
+        assert rpc_client
+        assert project_id
+
+        return TypeAdapter(list[TaskLogFileIdGet]).validate_python(
+            TaskLogFileIdGet.model_json_schema()["examples"],
+        )

packages/pytest-simcore/src/pytest_simcore/helpers/storage_rpc_server.py

@@ -0,0 +1,48 @@
+# pylint: disable=no-self-use
+# pylint: disable=not-context-manager
+# pylint: disable=protected-access
+# pylint: disable=redefined-outer-name
+# pylint: disable=unused-argument
+# pylint: disable=unused-variable
+
+
+from typing import Literal
+
+from models_library.api_schemas_rpc_async_jobs.async_jobs import (
+    AsyncJobFilter,
+    AsyncJobGet,
+)
+from models_library.api_schemas_webserver.storage import PathToExport
+from models_library.products import ProductName
+from models_library.users import UserID
+from pydantic import TypeAdapter, validate_call
+from pytest_mock import MockType
+from servicelib.rabbitmq._client_rpc import RabbitMQRPCClient
+
+
+class StorageSideEffects:
+    # pylint: disable=no-self-use
+    @validate_call(config={"arbitrary_types_allowed": True})
+    async def start_export_data(
+        self,
+        rabbitmq_rpc_client: RabbitMQRPCClient | MockType,
+        *,
+        user_id: UserID,
+        product_name: ProductName,
+        paths_to_export: list[PathToExport],
+        export_as: Literal["path", "download_link"],
+    ) -> tuple[AsyncJobGet, AsyncJobFilter]:
+        assert rabbitmq_rpc_client
+        assert user_id
+        assert product_name
+        assert paths_to_export
+        assert export_as
+
+        async_job_get = TypeAdapter(AsyncJobGet).validate_python(
+            AsyncJobGet.model_json_schema()["examples"][0],
+        )
+        async_job_filter = TypeAdapter(AsyncJobFilter).validate_python(
+            AsyncJobFilter.model_json_schema()["examples"][0],
+        )
+
+        return async_job_get, async_job_filter

packages/service-library/src/servicelib/deferred_tasks/_deferred_manager.py

@@ -8,7 +8,13 @@
 
 import arrow
 from faststream.exceptions import NackMessage, RejectMessage
-from faststream.rabbit import ExchangeType, RabbitBroker, RabbitExchange, RabbitRouter
+from faststream.rabbit import (
+    ExchangeType,
+    RabbitBroker,
+    RabbitExchange,
+    RabbitQueue,
+    RabbitRouter,
+)
 from pydantic import NonNegativeInt
 from servicelib.logging_utils import log_catch, log_context
 from servicelib.redis import RedisClientSDK
@@ -149,10 +155,14 @@ def __init__(
         self._global_resources_prefix = f"{calling_module_name}"
 
         self.common_exchange = RabbitExchange(
-            f"{self._global_resources_prefix}_common", type=ExchangeType.DIRECT
+            f"{self._global_resources_prefix}_common",
+            durable=True,
+            type=ExchangeType.DIRECT,
         )
         self.cancellation_exchange = RabbitExchange(
-            f"{self._global_resources_prefix}_cancellation", type=ExchangeType.FANOUT
+            f"{self._global_resources_prefix}_cancellation",
+            durable=True,
+            type=ExchangeType.FANOUT,
         )
 
     def patch_based_deferred_handlers(self) -> None:
@@ -243,8 +253,10 @@ def un_patch_base_deferred_handlers(cls) -> None:
                         subclass.is_present.original_is_present  # type: ignore
                     )
 
-    def _get_global_queue_name(self, queue_name: _FastStreamRabbitQueue) -> str:
-        return f"{self._global_resources_prefix}_{queue_name}"
+    def _get_global_queue(self, queue_name: _FastStreamRabbitQueue) -> RabbitQueue:
+        return RabbitQueue(
+            f"{self._global_resources_prefix}_{queue_name}", durable=True
+        )
 
     def __get_subclass(
         self, class_unique_reference: ClassUniqueReference
@@ -259,7 +271,7 @@ async def __publish_to_queue(
     ) -> None:
         await self.broker.publish(
             task_uid,
-            queue=self._get_global_queue_name(queue),
+            queue=self._get_global_queue(queue),
             exchange=(
                 self.cancellation_exchange
                 if queue == _FastStreamRabbitQueue.MANUALLY_CANCELLED
@@ -569,47 +581,43 @@ def _register_subscribers(self) -> None:
         # pylint:disable=unexpected-keyword-arg
         # pylint:disable=no-value-for-parameter
         self._fs_handle_scheduled = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.SCHEDULED),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.SCHEDULED),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_scheduled)
 
         self._fs_handle_submit_task = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.SUBMIT_TASK),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.SUBMIT_TASK),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_submit_task)
 
         self._fs_handle_worker = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.WORKER),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.WORKER),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_worker)
 
         self._fs_handle_error_result = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.ERROR_RESULT),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.ERROR_RESULT),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_error_result)
 
         self._fs_handle_finished_with_error = self.router.subscriber(
-            queue=self._get_global_queue_name(
-                _FastStreamRabbitQueue.FINISHED_WITH_ERROR
-            ),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.FINISHED_WITH_ERROR),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_finished_with_error)
 
         self._fs_handle_deferred_result = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.DEFERRED_RESULT),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.DEFERRED_RESULT),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_deferred_result)
 
         self._fs_handle_manually_cancelled = self.router.subscriber(
-            queue=self._get_global_queue_name(
-                _FastStreamRabbitQueue.MANUALLY_CANCELLED
-            ),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.MANUALLY_CANCELLED),
             exchange=self.cancellation_exchange,
             retry=True,
         )(self._fs_handle_manually_cancelled)

packages/service-library/src/servicelib/rabbitmq/_client_rpc.py

@@ -49,7 +49,7 @@ async def _rpc_initialize(self) -> None:
         self._channel = await self._connection.channel()
 
         self._rpc = aio_pika.patterns.RPC(self._channel)
-        await self._rpc.initialize()
+        await self._rpc.initialize(durable=True, auto_delete=True)
 
     async def close(self) -> None:
         with log_context(
@@ -134,6 +134,7 @@ async def register_handler(
             RPCNamespacedMethodName.from_namespace_and_method(namespace, method_name),
             handler,
             auto_delete=True,
+            durable=True,
         )
 
     async def register_router(

packages/service-library/src/servicelib/rabbitmq/_utils.py

@@ -85,6 +85,19 @@ async def declare_queue(
         # NOTE: setting a name will ensure multiple instance will take their data here
         queue_parameters |= {"name": queue_name}
 
+    # avoids deprecated `transient_nonexcl_queues` warning in RabbitMQ
+    if (
+        queue_parameters.get("durable", False) is False
+        and queue_parameters.get("exclusive", False) is False
+    ):
+        msg = (
+            "Queue must be `durable` or `exclusive`, but not both. "
+            "This is to avoid the `transient_nonexcl_queues` warning. "
+            "NOTE: if both `durable` and `exclusive` are missing they are considered False. "
+            f"{queue_parameters=}"
+        )
+        raise ValueError(msg)
+
     # NOTE: if below line raises something similar to ``ChannelPreconditionFailed: PRECONDITION_FAILED``
     # most likely someone changed the signature of the queues (parameters etc...)
     # Safest way to deal with it:

packages/service-library/src/servicelib/rabbitmq/rpc_interfaces/director_v2/computations_tasks.py

@@ -0,0 +1,41 @@
+# pylint: disable=too-many-arguments
+import logging
+from typing import Final
+
+from models_library.api_schemas_directorv2 import (
+    DIRECTOR_V2_RPC_NAMESPACE,
+)
+from models_library.api_schemas_directorv2.computations import TaskLogFileIdGet
+from models_library.projects import ProjectID
+from models_library.rabbitmq_basic_types import RPCMethodName
+from pydantic import TypeAdapter
+
+from ....logging_utils import log_decorator
+from ... import RabbitMQRPCClient
+
+_logger = logging.getLogger(__name__)
+
+
+_RPC_METHOD_NAME_ADAPTER: TypeAdapter[RPCMethodName] = TypeAdapter(RPCMethodName)
+
+_GET_COMPUTATION_TASK_LOG_FILE_IDS: Final[RPCMethodName] = (
+    _RPC_METHOD_NAME_ADAPTER.validate_python("get_computation_task_log_file_ids")
+)
+
+
+@log_decorator(_logger, level=logging.DEBUG)
+async def get_computation_task_log_file_ids(
+    rabbitmq_rpc_client: RabbitMQRPCClient, *, project_id: ProjectID
+) -> list[TaskLogFileIdGet]:
+    """
+    Raises:
+        ComputationalTaskMissingError
+    """
+    result = await rabbitmq_rpc_client.request(
+        DIRECTOR_V2_RPC_NAMESPACE,
+        _GET_COMPUTATION_TASK_LOG_FILE_IDS,
+        project_id=project_id,
+    )
+    assert isinstance(result, list)  # nosec
+    assert all(isinstance(item, TaskLogFileIdGet) for item in result)  # nosec
+    return result