ensure instrumentation messages are non exclusive queues (ITISFoundation#3685)

Author: sanderegg

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

@@ -94,8 +94,15 @@ async def ping(self) -> bool:
             return connection.connected.is_set()
 
     async def subscribe(
-        self, exchange_name: str, message_handler: MessageHandler
+        self,
+        exchange_name: str,
+        message_handler: MessageHandler,
+        exclusive_queue: bool = True,
     ) -> None:
+        """subscribe to exchange_name calling message_handler for every incoming message
+        - exclusive_queue: True means that every instance of this application will receive the incoming messages
+        - exclusive_queue: False means that only one instance of this application will reveice the incoming message
+        """
         assert self._channel_pool  # nosec
         async with self._channel_pool.acquire() as channel:
             channel: aio_pika.RobustChannel
@@ -110,11 +117,15 @@ async def subscribe(
             # consumer/publisher must set the same configuration for same queue
             # exclusive means that the queue is only available for THIS very client
             # and will be deleted when the client disconnects
-            queue = await channel.declare_queue(
+            queue_parameters = dict(
                 durable=True,
-                exclusive=True,
+                exclusive=exclusive_queue,
                 arguments={"x-message-ttl": _RABBIT_QUEUE_MESSAGE_DEFAULT_TTL_S},
             )
+            if not exclusive_queue:
+                # NOTE: setting a name will ensure multiple instance will take their data here
+                queue_parameters |= {"name": exchange_name}
+            queue = await channel.declare_queue(**queue_parameters)
             await queue.bind(exchange)
 
             async def _on_message(

services/web/server/src/simcore_service_webserver/computation_subscribe.py

@@ -119,22 +119,22 @@ async def events_message_parser(app: web.Application, data: bytes) -> bool:
     (
         LoggerRabbitMessage.get_channel_name(),
         log_message_parser,
-        {"no_ack": True},
+        {},
     ),
     (
         ProgressRabbitMessage.get_channel_name(),
         progress_message_parser,
-        {"no_ack": True},
+        {},
     ),
     (
         InstrumentationRabbitMessage.get_channel_name(),
         instrumentation_message_parser,
-        {"no_ack": False},
+        dict(exclusive_queue=False),
     ),
     (
         EventRabbitMessage.get_channel_name(),
         events_message_parser,
-        {"no_ack": False},
+        {},
     ),
 )
 
@@ -151,9 +151,9 @@ async def setup_rabbitmq_consumer(app: web.Application) -> AsyncIterator[None]:
     ):
         rabbit_client = RabbitMQClient("webserver", settings)
 
-        for exchange_name, parser_fct, _exchange_kwargs in EXCHANGE_TO_PARSER_CONFIG:
+        for exchange_name, parser_fct, queue_kwargs in EXCHANGE_TO_PARSER_CONFIG:
             await rabbit_client.subscribe(
-                exchange_name, functools.partial(parser_fct, app)
+                exchange_name, functools.partial(parser_fct, app), **queue_kwargs
             )
 
     yield