Merge pull request #783 from Bogdanp/fix/rabbitmq-pika-logging

Fix/rabbitmq pika logging

Author: synweap15

dramatiq/brokers/rabbitmq.py

@@ -208,16 +208,29 @@ def channel(self):
             except Exception:
                 self.logger.exception("Encountered exception while closing Channel.")
 
+    def _ignore_pika_logs(self) -> None:
+        """Ensures that pika logs are filtered.
+
+        The main thread may keep connections open for a long time
+        w/o publishing heartbeats, which means that they'll end up
+        being closed by the time the broker is closed.  When that
+        happens, pika logs a bunch of scary stuff so we want to
+        filter that out.
+        """
+
+        logging_filter = _IgnoreScaryLogs()
+        ignored_loggers = ["pika.adapters.base_connection", "pika.adapters.blocking_connection"]
+
+        # Make sure the filter is added only once.
+        for logger_name in ignored_loggers:
+            ignored_logger = logging.getLogger(logger_name)
+            if not any(isinstance(f, _IgnoreScaryLogs) for f in ignored_logger.filters):
+                ignored_logger.addFilter(logging_filter)
+
     def close(self) -> None:
         """Close all open RabbitMQ connections."""
-        # The main thread may keep connections open for a long time
-        # w/o publishing heartbeats, which means that they'll end up
-        # being closed by the time the broker is closed.  When that
-        # happens, pika logs a bunch of scary stuff so we want to
-        # filter that out.
-        logging_filter = _IgnoreScaryLogs()
-        logging.getLogger("pika.adapters.base_connection").addFilter(logging_filter)
-        logging.getLogger("pika.adapters.blocking_connection").addFilter(logging_filter)
+
+        self._ignore_pika_logs()
 
         self.logger.debug("Closing channels and connections...")
         for channel_or_conn in chain(self.channels, self.connections):

tests/test_rabbitmq.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import os
 import time
 from threading import Event
@@ -400,6 +401,40 @@ def test_ignore_scary_logs_filter_ignores_logs():
     assert log_filter.filter(record)
 
 
+def test_rabbitmq_close_only_registers_ignore_filter_once():
+    # Given a RabbitmqBroker
+    broker = RabbitmqBroker()
+
+    base_logger = logging.getLogger("pika.adapters.base_connection")
+    blocking_logger = logging.getLogger("pika.adapters.blocking_connection")
+
+    # And snapshots of the current filters
+    original_base_filters = list(base_logger.filters)
+    original_blocking_filters = list(blocking_logger.filters)
+
+    try:
+        # When I close the broker twice
+        broker.close()
+        broker.close()
+
+        base_filters = [f for f in base_logger.filters if isinstance(f, _IgnoreScaryLogs)]
+        blocking_filters = [f for f in blocking_logger.filters if isinstance(f, _IgnoreScaryLogs)]
+
+        # Then only one ignore filter is registered per logger
+        assert len(base_filters) == 1
+        assert len(blocking_filters) == 1
+    finally:
+        # And the filters are removed after the test run
+        # so they don't affect the global state.
+        for log_filter in list(base_logger.filters):
+            if log_filter not in original_base_filters:
+                base_logger.removeFilter(log_filter)
+
+        for log_filter in list(blocking_logger.filters):
+            if log_filter not in original_blocking_filters:
+                blocking_logger.removeFilter(log_filter)
+
+
 def test_rabbitmq_broker_can_join_with_timeout(rabbitmq_broker, rabbitmq_worker):
     # Given that I have an actor that takes a long time to run
     @dramatiq.actor