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

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

@@ -6,11 +6,17 @@
 from typing import Any, Awaitable, Callable, Final, Optional
 
 import aio_pika
-from aio_pika.exceptions import ChannelClosed
 from aio_pika.patterns import RPC
-from pydantic import PositiveInt
+from aiormq import AMQPConnectionError
+from packaging.version import Version
+from pydantic import PositiveFloat
 from servicelib.logging_utils import log_context
 from settings_library.rabbit import RabbitSettings
+from tenacity._asyncio import AsyncRetrying
+from tenacity.before import before_log
+from tenacity.retry import retry_if_exception
+from tenacity.stop import stop_after_delay
+from tenacity.wait import wait_random
 
 from .rabbitmq_errors import RemoteMethodNotRegisteredError, RPCNotInitializedError
 from .rabbitmq_utils import RPCMethodName, RPCNamespace, RPCNamespacedMethodName
@@ -30,18 +36,12 @@ def _connection_close_callback(sender: Any, exc: Optional[BaseException]) -> Non
             )
 
 
-def _channel_close_callback(sender: Any, exc: Optional[BaseException]) -> None:
+def _channel_close_callback(_: Any, exc: Optional[BaseException]) -> None:
     if exc:
         if isinstance(exc, asyncio.CancelledError):
             log.info("Rabbit channel was cancelled")
-        elif isinstance(exc, ChannelClosed):
-            log.info("%s", exc)
         else:
-            log.error(
-                "Rabbit channel closed with exception from %s:%s",
-                sender,
-                exc,
-            )
+            log.error("Rabbit channel closed with exception from %s", exc)
 
 
 async def _get_connection(
@@ -93,7 +93,13 @@ async def rpc_initialize(self) -> None:
         )
         self._rpc_channel = await self._rpc_connection.channel()
 
-        self._rpc = RPC(self._rpc_channel, host_exceptions=True)
+        if Version(aio_pika.__version__) >= Version("9.0.4"):
+            log.warning(
+                "When all libraries have `aio_pika>=9.0.4` use:"
+                "`self._rpc = RPC(self._rpc_channel, host_exceptions=True)`"
+                "on the line below"
+            )
+        self._rpc = RPC(self._rpc_channel)
         await self._rpc.initialize()
 
     async def close(self) -> None:
@@ -190,13 +196,19 @@ async def rpc_request(
         namespace: RPCNamespace,
         method_name: RPCMethodName,
         *,
-        timeout_s: Optional[PositiveInt] = 5,
+        timeout_s: PositiveFloat = 5,
+        connection_error_timeout_s: PositiveFloat = 60,
         **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` amount of seconds to wait for a reply once the message
+            was accepted by the remove replier
+        param: `connection_error_timeout_s` 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
@@ -211,13 +223,21 @@ 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(2),
+                stop=stop_after_delay(connection_error_timeout_s),
+                retry=retry_if_exception(AMQPConnectionError),
+                before=before_log(log, logging.DEBUG),
+                reraise=True,
+            ):
+                with attempt:
+                    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)
         except aio_pika.MessageProcessError as e:
             if e.args[0] == "Message has been returned":
                 raise RemoteMethodNotRegisteredError(
@@ -239,8 +259,14 @@ async def rpc_register_handler(
         if self._rpc is None:
             raise RPCNotInitializedError()
 
+        namespaced_method_name = RPCNamespacedMethodName.from_namespace_and_method(
+            namespace, method_name
+        )
+        log.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/rabbitmq_errors.py

@@ -20,3 +20,13 @@ class RemoteMethodNotRegisteredError(BaseRPCError):
         "Could not find a remote method named: '{method_name}'. "
         "Message from remote server was returned: {incoming_message}. "
     )
+
+
+class GatheredRuntimeErrors(BaseRPCError):
+    """
+    Has to be defined in both `requester` and `replier`
+    source codes to work as expected.
+    """
+
+    code = f"{_ERROR_PREFIX}.executing_raised_errors"
+    msg_template = "The following errors occurred: {errors}"

packages/service-library/tests/test_rabbitmq_rpc.py

@@ -2,9 +2,11 @@
 # pylint:disable=unused-argument
 
 import asyncio
-from typing import Any, Awaitable, Final
+from typing import Any, Awaitable, Callable, Final
 
 import pytest
+from docker.client import DockerClient
+from docker.models.containers import Container
 from pydantic import NonNegativeInt, ValidationError
 from pytest import LogCaptureFixture
 from servicelib.rabbitmq import RabbitMQClient
@@ -21,6 +23,8 @@
 
 MULTIPLE_REQUESTS_COUNT: Final[NonNegativeInt] = 100
 
+# FIXTURES
+
 
 @pytest.fixture
 def namespace() -> RPCNamespace:
@@ -43,6 +47,32 @@ async def rabbit_replier(rabbit_service: RabbitSettings) -> RabbitMQClient:
     await client.close()
 
 
+@pytest.fixture
+def restart_rabbit(
+    docker_stack: dict,
+    testing_environ_vars: dict,
+    docker_client: DockerClient,
+) -> Callable:
+    prefix = testing_environ_vars["SWARM_STACK_NAME"]
+    service_name = f"{prefix}_rabbit"
+    assert service_name in docker_stack["services"]
+
+    async def _reboot() -> None:
+        containers = docker_client.containers.list(
+            filters={"label": f"com.docker.swarm.service.name={service_name}"}
+        )
+        assert len(containers) == 1
+        container: Container = containers[0]
+        # killing the container will cause the service to be unavailable
+        #  and swarm to restart it. Exactly what we are trying to test
+        container.kill()
+
+    return _reboot
+
+
+# UTILS
+
+
 async def add_me(*, x: Any, y: Any) -> Any:
     result = x + y
     # NOTE: types are not enforced
@@ -65,6 +95,9 @@ def __add__(self, other: "CustomClass") -> "CustomClass":
         return CustomClass(x=self.x + other.x, y=self.y + other.y)
 
 
+# TESTS
+
+
 @pytest.mark.parametrize(
     "x,y,expected_result,expected_type",
     [
@@ -369,3 +402,25 @@ async def _a_handler() -> None:
         assert "ensure this value has at most 255 characters" in f"{exec_info.value}"
     else:
         await rabbit_replier.rpc_register_handler("a", handler_name, _a_handler)
+
+
+async def test_rabbit_unavailable_just_before_request(
+    rabbit_requester: RabbitMQClient,
+    rabbit_replier: RabbitMQClient,
+    namespace: RPCNamespace,
+    restart_rabbit: Callable,
+):
+    times_called = 0
+
+    async def _func() -> None:
+        nonlocal times_called
+        times_called += 1
+
+    await rabbit_replier.rpc_register_handler(namespace, _func.__name__, _func)
+
+    await restart_rabbit()
+
+    # this function will be retried because rabbitmq is restarting
+    await rabbit_requester.rpc_request(namespace, _func.__name__)
+
+    assert times_called == 1

services/agent/requirements/_base.in

@@ -10,6 +10,7 @@
 --requirement ../../../packages/settings-library/requirements/_base.in
 --requirement ../../../packages/service-library/requirements/_base.in
 
+aio-pika
 aiodocker
 fastapi
 packaging

services/agent/requirements/_base.txt

@@ -1,11 +1,13 @@
 #
-# This file is autogenerated by pip-compile with python 3.9
-# To update, run:
+# This file is autogenerated by pip-compile with Python 3.9
+# by the following command:
 #
 #    pip-compile --output-file=requirements/_base.txt --resolver=backtracking --strip-extras requirements/_base.in
 #
 aio-pika==8.2.5
-    # via -r requirements/../../../packages/service-library/requirements/_base.in
+    # via
+    #   -r requirements/../../../packages/service-library/requirements/_base.in
+    #   -r requirements/_base.in
 aiodebug==2.3.0
     # via -r requirements/../../../packages/service-library/requirements/_base.in
 aiodocker==0.21.0

services/agent/requirements/_test.in

@@ -21,3 +21,4 @@ pytest-asyncio
 pytest-cov
 pytest-mock
 pytest-runner
+python-dotenv

services/agent/requirements/_test.txt

@@ -207,6 +207,8 @@ python-dateutil==2.8.2
     #   botocore
     #   faker
     #   moto
+python-dotenv==1.0.0
+    # via -r requirements/_test.in
 python-jose==3.3.0
     # via moto
 pyyaml==5.4.1

services/agent/src/simcore_service_agent/core/application.py

@@ -15,7 +15,7 @@
     SUMMARY,
     VERSION,
 )
-from ..modules import task_monitor
+from ..modules import low_priority_managers, rabbitmq, task_monitor
 from ._routes import router
 from .settings import ApplicationSettings
 
@@ -50,8 +50,10 @@ def create_app() -> FastAPI:
     # ROUTERS
     app.include_router(router)
 
-    # EVENTS
+    # SUBMODULES
+    low_priority_managers.setup(app)
     task_monitor.setup(app)
+    rabbitmq.setup(app)
 
     async def _on_startup() -> None:
         print(APP_STARTED_BANNER_MSG, flush=True)

services/agent/src/simcore_service_agent/core/errors.py

@@ -0,0 +1,10 @@
+from pydantic.errors import PydanticErrorMixin
+
+
+class AgentRuntimeError(PydanticErrorMixin, RuntimeError):
+    msg_template: str = "Autoscaling unexpected error"
+
+
+class ConfigurationError(AgentRuntimeError):
+    code: str = "agent.application_configuration"
+    msg_template: str = "Application misconfiguration: {msg}"

services/agent/src/simcore_service_agent/core/settings.py

@@ -4,6 +4,7 @@
 from pydantic import Field, NonNegativeInt, validator
 from settings_library.base import BaseCustomSettings
 from settings_library.r_clone import S3Provider
+from settings_library.rabbit import RabbitSettings
 from settings_library.utils_logging import MixinLoggingSettings
 
 _MINUTE: Final[NonNegativeInt] = 60
@@ -39,6 +40,9 @@ class ApplicationSettings(BaseCustomSettings, MixinLoggingSettings):
         60 * _MINUTE, description="interval at which to repeat volumes cleanup"
     )
 
+    AGENT_DOCKER_NODE_ID: str = Field(..., description="used by the rabbitmq module")
+    AGENT_RABBITMQ: Optional[RabbitSettings] = Field(auto_default_from_env=True)
+
     @validator("AGENT_VOLUMES_CLEANUP_S3_ENDPOINT", pre=True)
     @classmethod
     def ensure_scheme(cls, v: str, values) -> str: