feat: Set the AMQP connection name to service and hostname

app/datasources/queue/queue_provider.py

@@ -1,5 +1,6 @@
 # SPDX-License-Identifier: FSL-1.1-MIT
 import logging
+import socket
 from asyncio import AbstractEventLoop
 from collections.abc import Callable
 from typing import Any
@@ -23,6 +24,10 @@
 
 logger = logging.getLogger(__name__)
 
+# Name shown by RabbitMQ in the management UI and logs. The hostname is the pod name
+# in Kubernetes, so every replica can be told apart.
+AMQP_CONNECTION_NAME = f"safe-decoder-service@{socket.gethostname()}"
+
 
 class QueueProvider:
     _connection: AbstractRobustConnection | None
@@ -47,7 +52,9 @@ async def _connect(self, loop: AbstractEventLoop) -> None:
         try:
             logger.info("Connecting to RabbitMQ")
             self._connection = await aio_pika.connect_robust(
-                url=settings.RABBITMQ_AMQP_URL, loop=loop
+                url=settings.RABBITMQ_AMQP_URL,
+                loop=loop,
+                client_properties={"connection_name": AMQP_CONNECTION_NAME},
             )
             logger.info("Connected to RabbitMQ")
         except aio_pika.exceptions.AMQPConnectionError as e:

app/tests/datasources/queue/test_queue_provider.py

@@ -1,4 +1,6 @@
+# SPDX-License-Identifier: FSL-1.1-MIT
 import asyncio
+import socket
 import unittest
 from unittest.mock import patch
 
@@ -7,7 +9,7 @@
 
 from app.config import settings
 from app.datasources.queue.exceptions import QueueProviderUnableToConnectException
-from app.datasources.queue.queue_provider import QueueProvider
+from app.datasources.queue.queue_provider import AMQP_CONNECTION_NAME, QueueProvider
 
 
 class TestQueueProviderIntegration(unittest.IsolatedAsyncioTestCase):
@@ -22,6 +24,19 @@ async def test_connect_success(self):
         await self.provider.disconnect()
         self.assertFalse(self.provider.is_connected())
 
+    async def test_connect_connection_name(self):
+        self.assertEqual(
+            AMQP_CONNECTION_NAME, f"safe-decoder-service@{socket.gethostname()}"
+        )
+        await self.provider.connect(self.loop)
+        assert isinstance(self.provider._connection, aio_pika.RobustConnection)
+        # Kept on the connection, so reconnections send the same name
+        self.assertEqual(
+            self.provider._connection.kwargs["client_properties"],
+            {"connection_name": AMQP_CONNECTION_NAME},
+        )
+        await self.provider.disconnect()
+
     async def test_connect_failure(self):
         provider = QueueProvider()