♻️ volumes are removed via agent (⚠️ devops)

docs/coding-conventions.md

@@ -72,6 +72,13 @@ In short we use the following naming convention ( roughly  [PEP8](https://peps.p
 We should avoid merging PRs with ``TODO:`` and ``FIXME:`` into master. One of our bots detects those and flag them as code-smells. If we still want to keep this idea/fix noted in the code, those can be rewritten as ``NOTE:`` and should be extended with a link to a github issue with more details. For a context, see [discussion here](https://github.com/ITISFoundation/osparc-simcore/pull/3380#discussion_r979893502).
 
 
+## Retries
+
+[Tenacity](https://github.com/jd/tenacity) wherever a retry is required. While most retries are straight forward consider [the following article](https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/) regarding retries services and how to avoid overwhelming them.
+
+When retrying an API call (or some sort of request) to an external system, consider that that system can have trouble replying.
+It is most effective to create a retry using `wait_random_exponential` from tenacity which implements what the article above describes.
+
 ### CC2: No commented code
 
 Avoid commented code, but if you *really* want to keep it then add an explanatory `NOTE:`

packages/models-library/src/models_library/sidecar_volumes.py

@@ -47,6 +47,10 @@ class VolumeStatus(StrAutoEnum):
 
 class VolumeState(BaseModel):
     status: VolumeStatus
+    volume_names: list[str] = Field(
+        ...,
+        description="agent uses the volume's name to search it's status",
+    )
     last_changed: datetime = Field(default_factory=lambda: arrow.utcnow().datetime)
 
     def __eq__(self, other: object) -> bool:

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

@@ -5,10 +5,16 @@
 
 import aio_pika
 import aiormq
+from aio_pika.exceptions import AMQPConnectionError, ChannelInvalidStateError
 from aio_pika.patterns import RPC
-from pydantic import PositiveInt
+from pydantic import PositiveFloat
 from servicelib.logging_utils import log_catch, log_context
 from settings_library.rabbit import RabbitSettings
+from tenacity._asyncio import AsyncRetrying
+from tenacity.before_sleep import before_sleep_log
+from tenacity.retry import retry_if_exception_type
+from tenacity.stop import stop_after_delay
+from tenacity.wait import wait_random_exponential
 
 from .rabbitmq_errors import RemoteMethodNotRegisteredError, RPCNotInitializedError
 from .rabbitmq_utils import (
@@ -122,8 +128,7 @@ async def rpc_initialize(self) -> None:
             },
         )
         self._rpc_channel = await self._rpc_connection.channel()
-
-        self._rpc = RPC(self._rpc_channel)
+        self._rpc = RPC(self._rpc_channel, host_exceptions=True)
         await self._rpc.initialize()
 
     async def close(self) -> None:
@@ -314,13 +319,19 @@ async def rpc_request(
         namespace: RPCNamespace,
         method_name: RPCMethodName,
         *,
-        timeout_s: PositiveInt | None = 5,
+        timeout_s_method: PositiveFloat,
+        timeout_s_connection_error: PositiveFloat,
         **kwargs: dict[str, Any],
     ) -> Any:
         """
         Call a remote registered `handler` by providing it's `namespace`, `method_name`
         and `kwargs` containing the key value arguments expected by the remote `handler`.
 
+        param: `timeout_s_method` amount of seconds to wait for a reply from the remote handler
+            invoked via `method_name`
+        param: `timeout_s_connection_error` amount of seconds to wait for rabbit to
+            be available again in case there was a connection error
+
         :raises asyncio.TimeoutError: when message expired
         :raises CancelledError: when called :func:`RPC.cancel`
         :raises RuntimeError: internal error
@@ -335,13 +346,23 @@ async def rpc_request(
             namespace, method_name
         )
         try:
-            queue_expiration_timeout = timeout_s
-            awaitable = self._rpc.call(
-                namespaced_method_name,
-                expiration=queue_expiration_timeout,
-                kwargs=kwargs,
-            )
-            return await asyncio.wait_for(awaitable, timeout=timeout_s)
+            async for attempt in AsyncRetrying(
+                wait=wait_random_exponential(max=5),
+                stop=stop_after_delay(timeout_s_connection_error),
+                retry=retry_if_exception_type(
+                    (AMQPConnectionError, ChannelInvalidStateError)
+                ),
+                before_sleep=before_sleep_log(_logger, logging.WARNING),
+                reraise=True,
+            ):
+                with attempt:
+                    queue_expiration_timeout = timeout_s_method
+                    awaitable = self._rpc.call(
+                        namespaced_method_name,
+                        expiration=queue_expiration_timeout,
+                        kwargs=kwargs,
+                    )
+                    return await asyncio.wait_for(awaitable, timeout=timeout_s_method)
         except aio_pika.MessageProcessError as e:
             if e.args[0] == "Message has been returned":
                 raise RemoteMethodNotRegisteredError(
@@ -366,8 +387,14 @@ async def rpc_register_handler(
         if self._rpc is None:
             raise RPCNotInitializedError()
 
+        namespaced_method_name = RPCNamespacedMethodName.from_namespace_and_method(
+            namespace, method_name
+        )
+        _logger.info(
+            "RPC registered handler '%s' to queue '%s'", handler, namespaced_method_name
+        )
         await self._rpc.register(
-            RPCNamespacedMethodName.from_namespace_and_method(namespace, method_name),
+            namespaced_method_name,
             handler,
             auto_delete=True,
         )

packages/service-library/src/servicelib/serial_executor.py

@@ -0,0 +1,173 @@
+import asyncio
+import functools
+import logging
+from abc import abstractmethod
+from asyncio import Future, Queue, Task, create_task, wait_for
+from dataclasses import dataclass
+from typing import Any, Final, TypeAlias
+
+from .background_task import cancel_task
+from .logging_utils import log_context
+
+_logger = logging.getLogger(__name__)
+
+_CANCEL_TASK_TIMEOUT_S: Final[float] = 5
+
+ContextKey: TypeAlias = str
+
+
+@dataclass
+class _Request:
+    context_key: ContextKey
+    future: Future
+    args: tuple[Any]
+    kwargs: dict[str, Any]
+
+
+class BaseSerialExecutor:
+    """
+    If a bunch of wait_for_result operations are launched together,
+    they are tried to be executed in parallel except the ones that have the same context_key.
+
+    """
+
+    def __init__(self, polling_interval: float = 0.1) -> None:
+        self.polling_interval: float = polling_interval
+
+        self._requests_queue: Queue[_Request | None] = Queue()
+        self._request_ingestion_task: Task | None = None
+
+        # TODO maybe a better name here?
+        self._context_queue: Queue[int | None] = Queue()
+        self._context_task: Task | None = None
+
+        self._requests_to_start: dict[ContextKey, list[_Request]] = {}
+        self._running_requests: dict[ContextKey, Task] = {}
+
+    async def _handle_payload(self, request: _Request) -> None:
+        try:
+            request.future.set_result(await self.run(*request.args, **request.kwargs))
+        except Exception as e:  # pylint: disable=broad-exception-caught
+            request.future.set_exception(e)
+
+    async def _request_processor_worker(self) -> None:
+        with log_context(
+            _logger, logging.DEBUG, f"request processor for {self.__class__}"
+        ):
+            while True:
+                message: int | None = await self._context_queue.get()
+                if message is None:
+                    break
+
+                # NOTE: this entry logs are supposed to stop after all
+                # tasks for the context are processed
+                _logger.debug("Received request to start a task")
+
+                # find the next context_key that can be started
+                found_context_key: ContextKey | None = None
+                for context_key in self._requests_to_start:
+                    if context_key not in self._running_requests:
+                        found_context_key = context_key
+
+                # if we expect any other jobs to be started do this
+                if found_context_key is None and len(self._requests_to_start) != 0:
+                    # waiting a bit to give time for the current task to finish
+                    # before creating a new one
+                    await asyncio.sleep(self.polling_interval)
+                    await self._context_queue.put(1)  # trigger
+                    continue
+
+                if found_context_key is None:
+                    _logger.debug("Done processing enqueued requests")
+                    continue
+
+                # there are requests which can be picked up and started
+
+                requests: list[_Request] = self._requests_to_start[found_context_key]
+                request = requests.pop()
+
+                self._running_requests[request.context_key] = create_task(
+                    self._handle_payload(request)
+                )
+                self._running_requests[request.context_key].add_done_callback(
+                    functools.partial(
+                        lambda s, _: self._running_requests.pop(s, None),
+                        request.context_key,
+                    )
+                )
+
+    async def _request_ingestion_worker(self) -> None:
+        with log_context(
+            _logger, logging.DEBUG, f"request ingestion for {self.__class__}"
+        ):
+            while True:
+                request: _Request | None = await self._requests_queue.get()
+                if request is None:
+                    break
+
+                if request.context_key not in self._requests_to_start:
+                    self._requests_to_start[request.context_key] = []
+                self._requests_to_start[request.context_key].append(request)
+
+                await self._context_queue.put(1)  # trigger
+
+    async def start(self):
+        self._request_ingestion_task = create_task(self._request_ingestion_worker())
+        self._context_task = create_task(self._request_processor_worker())
+
+    async def stop(self):
+        if self._request_ingestion_task:
+            await self._requests_queue.put(None)
+            await cancel_task(
+                self._request_ingestion_task, timeout=_CANCEL_TASK_TIMEOUT_S
+            )
+        if self._context_task:
+            await self._context_queue.put(None)
+            await cancel_task(self._context_task, timeout=_CANCEL_TASK_TIMEOUT_S)
+
+        # cancel all existing tasks
+        for task in tuple(self._running_requests.values()):
+            await cancel_task(task, timeout=_CANCEL_TASK_TIMEOUT_S)
+
+    async def wait_for_result(
+        self, *args: Any, context_key: ContextKey, timeout: float, **kwargs: Any
+    ) -> Any:
+        """
+        Starts task and executes the code defined by `run`, waits for the task to
+        finish and returns its result.
+        All calls in parallel with the same `context_key` will get executed
+        sequentially. It guarantees only one task with the same `context_key`
+        is active at the same time.
+
+        If `run` raises an error that same error is raised in the context where this
+        method is called.
+
+        params:
+            context_key: calls sharing the same `context_key` will be ran in sequence,
+                all others will be ran in parallel
+            timeout: float seconds before giving up on waiting gor the result;
+                needs to take into consideration the fact that other tasks might be already
+                started for the same `context_key`, before those are finished this request
+                will not execute
+
+        raises:
+            asyncio.TimeoutError: if no result is provided after timeout seconds have massed
+        """
+        future = Future()
+        request = _Request(
+            context_key=context_key, future=future, args=args, kwargs=kwargs
+        )
+        await self._requests_queue.put(request)
+
+        # NOTE: this raises TimeoutError which has to be handled by tha caller
+        await wait_for(future, timeout=timeout)
+
+        # NOTE: will raise an exception if the future has raised na exception
+        # This also has to be handled by the caller
+        result = future.result()
+
+        return result
+
+    @abstractmethod
+    async def run(self, *args: Any, **kwargs: Any) -> Any:
+        """code to be executed for each request"""

packages/service-library/src/servicelib/sidecar_volumes.py

@@ -0,0 +1,75 @@
+import os
+from pathlib import Path
+from typing import Final
+from uuid import UUID
+
+from attr import dataclass
+from pydantic import PositiveInt
+
+from .docker_constants import PREFIX_DYNAMIC_SIDECAR_VOLUMES
+
+_UUID_LEN: Final[PositiveInt] = 36
+_UNDER_SCORE_LEN: Final[PositiveInt] = 1
+REGULAR_SOURCE_PORTION_LEN: Final[PositiveInt] = (
+    len(PREFIX_DYNAMIC_SIDECAR_VOLUMES) + 2 * _UUID_LEN + 3 * _UNDER_SCORE_LEN
+)
+
+STORE_FILE_NAME: Final[str] = "data.json"
+
+
+@dataclass
+class VolumeInfo:
+    node_uuid: UUID
+    run_id: UUID
+    possible_volume_name: str
+
+
+class VolumeUtils:
+    _MAX_VOLUME_NAME_LEN: Final[int] = 255
+
+    @classmethod
+    def get_name(cls, path: Path) -> str:
+        return f"{path}".replace(os.sep, "_")
+
+    @classmethod
+    def get_source(cls, path: Path, node_uuid: UUID, run_id: UUID) -> str:
+        """Returns a valid and unique volume name that is composed out of identifiers, namely
+            - relative target path
+            - node_uuid
+            - run_id
+
+        Guarantees that the volume name is unique between runs while also
+        taking into consideration the limit for the volume name's length
+        (255 characters).
+
+        SEE examples in `tests/unit/test_modules_dynamic_sidecar_volumes_resolver.py`
+        """
+        # NOTE: issues can occur when the paths of the mounted outputs, inputs
+        # and state folders are very long and share the same subdirectory path.
+        # Reversing volume name to prevent these issues from happening.
+        reversed_volume_name = cls.get_name(path)[::-1]
+        unique_name = f"{PREFIX_DYNAMIC_SIDECAR_VOLUMES}_{run_id}_{node_uuid}_{reversed_volume_name}"
+        return unique_name[: cls._MAX_VOLUME_NAME_LEN]
+
+    @classmethod
+    def get_volume_info(cls, source: str) -> VolumeInfo:
+        print(f"{source=}")
+        if len(source) <= REGULAR_SOURCE_PORTION_LEN:
+            raise ValueError(
+                f"source '{source}' must be at least {REGULAR_SOURCE_PORTION_LEN} characters"
+            )
+
+        # example: dyv_5813058f-8ec4-4aa9-bae1-f46c01040481_e3d42f3f-e1ad-418b-90e1-c44a95e97b91
+        without_volume_name = source[: REGULAR_SOURCE_PORTION_LEN - 1]
+
+        # example:_erots_pmet_
+        possible_reverted_volume_name = source[REGULAR_SOURCE_PORTION_LEN:]
+
+        _, run_id_str, node_uuid_str = without_volume_name.split("_")
+        possible_volume_name = possible_reverted_volume_name[::-1]
+
+        return VolumeInfo(
+            node_uuid=UUID(node_uuid_str),
+            run_id=UUID(run_id_str),
+            possible_volume_name=possible_volume_name,
+        )