Improvements to internal messaging debug logging

Author: adamcharnock

docs/explanation/internal-architecture.md

@@ -18,3 +18,4 @@ Lightbus' internal workings are composed of:
 ![Internal Architecture Diagram][diagram]
 
 [diagram]: ../static/images/internal-architecture.png
+

lightbus/client/docks/base.py

@@ -29,8 +29,12 @@ def __init__(
         self.api_registry = api_registry
         self.config = config
         self.error_queue = error_queue
-        self.producer = InternalProducer(queue=produce_to, error_queue=error_queue)
-        self.consumer = InternalConsumer(queue=consume_from, error_queue=error_queue)
+        self.producer = InternalProducer(
+            name=self.__class__.__name__, queue=produce_to, error_queue=error_queue
+        )
+        self.consumer = InternalConsumer(
+            name=self.__class__.__name__, queue=consume_from, error_queue=error_queue
+        )
 
         self.producer.start()
         self.consumer.start(self.handle)

lightbus/client/internal_messaging/consumer.py

@@ -17,10 +17,11 @@ class InternalConsumer:
     consumption by the transport
     """
 
-    def __init__(self, queue: InternalQueue, error_queue: ErrorQueueType):
+    def __init__(self, name: str, queue: InternalQueue, error_queue: ErrorQueueType):
         self._consumer_task: Optional[asyncio.Task] = None
         self._running_commands = set()
         self._ready = asyncio.Event()
+        self.name = name
         self.queue = queue
         self.error_queue = error_queue
 
@@ -30,7 +31,7 @@ def start(self, handler: Callable):
         Use `stop()` to shutdown the invoker.
         """
         logger.debug(
-            f"Starting consumer for handler {handler.__qualname__}(). This should report ready"
+            f"Starting {self.name} consumer for handler {handler.__qualname__}(). This should report ready"
             " shortly..."
         )
         self._consumer_task = asyncio.ensure_future(
@@ -58,7 +59,9 @@ async def close(self):
 
     async def _consumer_loop(self, queue, handler):
         """Continually fetch commands from the queue and handle them"""
-        logger.debug(f"Consumer loop is ready with handler {handler.__qualname__}()")
+        logger.debug(
+            f"Consumer loop for {self.name} is ready with handler {handler.__qualname__}()"
+        )
         self._ready.set()
 
         while True:
@@ -75,7 +78,7 @@ def handle_in_background(self, queue: InternalQueue, handler, command, on_done:
 
         This execution happens in the background.
         """
-        logger.debug(f"Handling command {command}")
+        logger.debug(f"Handling command {command} in consumer {self.name}")
 
         def when_task_finished(fut: asyncio.Future):
             self._running_commands.remove(fut)

lightbus/client/internal_messaging/producer.py

@@ -50,13 +50,14 @@ class InternalProducer:
     # How often should the queue sizes be monitored
     monitor_interval = 0.1
 
-    def __init__(self, queue: InternalQueue, error_queue: ErrorQueueType):
+    def __init__(self, name: str, queue: InternalQueue, error_queue: ErrorQueueType):
         """Initialise the invoker
 
         The callable specified by `on_exception` will be called with a single positional argument,
         which is the exception which occurred. This should take care of shutting down the invoker,
         as well as any other cleanup which needs to happen.
         """
+        self.name = name
         self._queue_monitor_task: Optional[asyncio.Task] = None
         self._monitor_ready = asyncio.Event()
         self.queue = queue
@@ -105,20 +106,18 @@ async def _queue_monitor(self):
 
                     logger.warning(
                         "Queue in %s has shrunk back down to %s commands.%s",
-                        self.__class__.__name__,
+                        self.name,
                         current_size,
                         everything_ok,
                     )
                 elif show_size_warning:
-                    logger.warning(
-                        "Queue in %s now has %s commands.", self.__class__.__name__, current_size
-                    )
+                    logger.warning("Queue in %s now has %s commands.", self.name, current_size)
 
             previous_size = current_size
             await asyncio.sleep(self.monitor_interval)
 
     def send(self, command) -> asyncio.Event:
-        logger.debug(f"Sending command {command}")
+        logger.debug(f"Sending command {command} in producer {self.name}")
         event = asyncio.Event()
         self.queue.put_nowait((command, event))
         return event

lightbus/client/subclients/base.py

@@ -26,8 +26,12 @@ def __init__(
         self.config = config
         self.schema = schema
         self.error_queue = error_queue
-        self.producer = InternalProducer(queue=produce_to, error_queue=error_queue)
-        self.consumer = InternalConsumer(queue=consume_from, error_queue=error_queue)
+        self.producer = InternalProducer(
+            name=self.__class__.__name__, queue=produce_to, error_queue=error_queue
+        )
+        self.consumer = InternalConsumer(
+            name=self.__class__.__name__, queue=consume_from, error_queue=error_queue
+        )
 
         self.producer.start()
         self.consumer.start(self.handle)

lightbus/transports/redis/rpc.py

@@ -27,14 +27,14 @@
 
 
 class RedisRpcTransport(RedisTransportMixin, RpcTransport):
-    """ Redis RPC transport providing at-most-once delivery
+    """Redis RPC transport providing at-most-once delivery
 
     This transport uses a redis list and a blocking pop operation
     to distribute an RPC call to a single RPC consumer.
 
     Each call also has a corresponding expiry key created. Once the
     key expires it should be assumed that the RPC call has timed
-    out and that therefore is should be discarded rather than
+    out and that therefore should be discarded rather than
     be processed.
     """
 
@@ -177,6 +177,7 @@ async def _consume_rpcs(self, apis: Sequence[Api]) -> Sequence[RpcMessage]:
                 # We need to manually close the connection here otherwise the aioredis
                 # pool will emit warnings saying that this connection still has pending
                 # commands (i.e. the above blocking pop)
+                logger.debug("Closing redis connection")
                 redis.close()
                 raise
 

lightbus/utilities/async_tools.py

@@ -181,5 +181,5 @@ async def call_on_schedule(callback, schedule: "Job", also_run_immediately: bool
 
 
 async def delayed_startup(coroutine: Coroutine, delay: float):
-    await asyncio.sleep(delay)
+    # await asyncio.sleep(delay)
     return await coroutine

tests/client/internal_messaging/conftest.py

@@ -12,7 +12,9 @@ async def consumer():
     def _on_exception(e):
         raise e
 
-    consumer = InternalConsumer(queue=InternalQueue(), error_queue=InternalQueue())
+    consumer = InternalConsumer(
+        name="TestConsumer", queue=InternalQueue(), error_queue=InternalQueue()
+    )
     yield consumer
     await consumer.close()
 
@@ -22,7 +24,9 @@ async def producer():
     def _on_exception(e):
         raise e
 
-    producer = InternalProducer(queue=InternalQueue(), error_queue=InternalQueue())
+    producer = InternalProducer(
+        name="TestProducer", queue=InternalQueue(), error_queue=InternalQueue()
+    )
     yield producer
     await producer.close()
 

tests/client/internal_messaging/test_producer.py

@@ -41,15 +41,15 @@ async def test_queue_monitor(producer: InternalProducer, caplog: LogCaptureFixtu
 
     # Now we have logging
     assert len(caplog.records) == 1
-    assert caplog.records[0].getMessage() == "Queue in InternalProducer now has 3 commands."
+    assert caplog.records[0].getMessage() == "Queue in TestProducer now has 3 commands."
     caplog.clear()  # Clear the log messages
 
     # Let's check we get another messages when the queue gets bigger again
     producer.queue.put_nowait(None)
     await asyncio.sleep(0.05)
 
     assert len(caplog.records) == 1
-    assert caplog.records[0].getMessage() == "Queue in InternalProducer now has 4 commands."
+    assert caplog.records[0].getMessage() == "Queue in TestProducer now has 4 commands."
     caplog.clear()  # Clear the log messages
 
     # Now check we get logging when the queue shrinks, but is still above the warning level
@@ -58,7 +58,7 @@ async def test_queue_monitor(producer: InternalProducer, caplog: LogCaptureFixtu
 
     assert len(caplog.records) == 1
     assert caplog.records[0].getMessage() == (
-        "Queue in InternalProducer has shrunk back down to 3 commands."
+        "Queue in TestProducer has shrunk back down to 3 commands."
     )
     caplog.clear()  # Clear the log messages
 
@@ -68,7 +68,7 @@ async def test_queue_monitor(producer: InternalProducer, caplog: LogCaptureFixtu
 
     assert len(caplog.records) == 1
     assert caplog.records[0].getMessage() == (
-        "Queue in InternalProducer has shrunk back down to 2 commands. "
+        "Queue in TestProducer has shrunk back down to 2 commands. "
         "Queue is now at an OK size again."
     )
     caplog.clear()  # Clear the log messages