Updates to handle automatic logging to a rabbitmq exchange log. The c…

python/idsse_common/idsse/common/log_util.py

@@ -38,7 +38,7 @@ def set_corr_id_context_var(
     Args:
         originator (str): Function, class, service name, etc. that is using logging module
         key (Optional[uuid.UUID]): a UUID. Default: randomly generated UUID.
-        issue_dt (Optional[Union[str, datettime]]): Datetime when a relevant forecast was issued
+        issue_dt (Optional[Union[str, datetime]]): Datetime when a relevant forecast was issued
     """
     if not key:
         key = uuid.uuid4()
@@ -88,7 +88,7 @@ class UTCFormatter(logging.Formatter):
 
 def get_default_log_config(level: str, with_corr_id=True):
     """
-    Get standardized python logging config (formatters, handlers directing to stdout, etc.)
+    Get standardized python logging config (formatters, handlers directing to stdout, rabbitmq etc.)
     as a dictionary. This dictionary can be passed directly to logging.config.dictConfig:
 
     import logging
@@ -137,11 +137,20 @@ def get_default_log_config(level: str, with_corr_id=True):
                 'formatter': 'standard',
                 'filters': filter_list,
             },
+            'rabbit': {
+                'class': 'python_logging_rabbitmq.RabbitMQHandler',
+                'host': 'localhost',
+                'port': 5672,
+                'formatter': 'standard',
+                'filters': filter_list,
+                'level': 'ERROR',
+                'declare_exchange': True,
+            },
         },
         'loggers': {
             '': {
                 'level': level,
-                'handlers': ['default', ],
+                'handlers': ['default', 'rabbit'],
             },
         }
     }

python/idsse_common/idsse/common/publish_confirm.py

@@ -63,6 +63,7 @@ class PublishConfirm:
     be closed, which usually are tied to permission related issues or
     socket timeouts.
     """
+
     def __init__(self, conn: Conn, exchange: Exch, queue: Queue):
         """Setup the example publisher object, passing in the RabbitMqUtils we will use to
         connect to RabbitMQ.
@@ -74,7 +75,7 @@ def __init__(self, conn: Conn, exchange: Exch, queue: Queue):
                 a "private queue", i.e. not intended for consumers, and all published messages
                 will have a 10-second TTL.
         """
-        self._thread = Thread(name=f'PublishConfirm-{randint(0,9)}',
+        self._thread = Thread(name=f'PublishConfirm-{randint(0, 9)}',
                               daemon=True,
                               target=self._run)
 
@@ -89,7 +90,7 @@ def __init__(self, conn: Conn, exchange: Exch, queue: Queue):
 
     def publish_message(self,
                         message: Dict,
-                        routing_key = '',
+                        routing_key='',
                         corr_id: Optional[str] = None) -> bool:
         """If the class is not stopping, publish a message to RabbitMQ,
         appending a list of deliveries with the message number that was sent.
@@ -113,8 +114,8 @@ def publish_message(self,
         # We expect a JSON message format, do a check here...
         try:
             properties = BasicProperties(content_type='application/json',
-                                                content_encoding='utf-8',
-                                                correlation_id=corr_id)
+                                         content_encoding='utf-8',
+                                         correlation_id=corr_id)
 
             logger.info('Publishing message to queue %s, message length: %d',
                         self._rmq_params.queue.name, len(json.dumps(message)))
@@ -124,8 +125,8 @@ def publish_message(self,
             self._records.message_number += 1
             self._records.deliveries[self._records.message_number] = message
             logger.debug('Published message # %i to exchange %s, queue %s, routing_key %s',
-                        self._records.message_number, self._rmq_params.exchange.name,
-                        self._rmq_params.queue.name, routing_key)
+                         self._records.message_number, self._rmq_params.exchange.name,
+                         self._rmq_params.queue.name, routing_key)
             return True
 
         except Exception as e:  # pylint: disable=broad-exception-caught
@@ -269,14 +270,16 @@ def _on_channel_open(self, channel: Channel):
         logger.debug('Adding channel close callback')
         self._channel.add_on_close_callback(self._on_channel_closed)
 
-        # Decleare exchange on our new channel
-        exch_name, exch_type = self._rmq_params.exchange
+        # Declare exchange on our new channel
+        exch_name, exch_type, exch_durable = self._rmq_params.exchange  # pylint: disable=unused-variable
         logger.debug('Declaring exchange %s', exch_name)
 
         # Note: using functools.partial is not required, it is demonstrating
         # how arbitrary data can be passed to the callback when it is called
         cb = functools.partial(self._on_exchange_declareok, userdata=exch_name)
-        self._channel.exchange_declare(exchange=exch_name, exchange_type=exch_type, callback=cb)
+        self._channel.exchange_declare(exchange=exch_name,
+                                       exchange_type=exch_type,
+                                       callback=cb)
 
     def _on_channel_closed(self, channel: Channel, reason: Exception):
         """Invoked by pika when RabbitMQ unexpectedly closes the channel.
@@ -378,7 +381,7 @@ def _on_delivery_confirmation(self, method_frame: Method):
         delivery_tag = method.delivery_tag
 
         logger.info('Received %s for delivery tag: %i (multiple: %s)',
-                     confirmation_type, delivery_tag, ack_multiple)
+                    confirmation_type, delivery_tag, ack_multiple)
 
         if confirmation_type == 'ack':
             self._records.acked += 1

python/idsse_common/idsse/common/rabbitmq_utils.py

@@ -58,6 +58,7 @@ class Exch(NamedTuple):
     """An internal data class for holding the RabbitMQ exchange info"""
     name: str
     type: str
+    durable: bool = True
 
 
 class Queue(NamedTuple):
@@ -89,7 +90,7 @@ def _initialize_exchange_and_queue(
 
     # Do not try to declare the default exchange. It already exists
     if exch.name != '':
-        channel.exchange_declare(exchange=exch.name, exchange_type=exch.type)
+        channel.exchange_declare(exchange=exch.name, exchange_type=exch.type, durable=exch.durable)
 
     # Do not try to declare or bind built-in queues. They are pseudo-queues that already exist
     if queue.name.startswith('amq.rabbitmq.'):
@@ -121,7 +122,7 @@ def subscribe_to_queue(
     queue if needed, and invoking the provided callback when a message is received.
 
     If an existing BlockingConnection or BlockingChannel are passed, they are used to
-    setup the subscription, but by default a new connection and channel will be established and
+    set up the subscription, but by default a new connection and channel will be established and
     returned, which the caller can immediately begin doing RabbitMQ operations with.
 
     For example: start a blocking consume of messages with channel.start_consuming(), or
@@ -131,7 +132,7 @@ def subscribe_to_queue(
         connection (Union[Conn, BlockingConnection]): connection parameters to establish new
             RabbitMQ connection, or existing RabbitMQ connection to reuse for this consumer.
         params (RabbitMqParams): parameters for the RabbitMQ exchange and queue from which to
-            to consume messages.
+            consume messages.
         on_message_callback (Callable[
             [BlockingChannel, Basic.Deliver, BasicProperties, bytes], None]):
             function to handle messages that are received over the subscribed exchange and queue.
@@ -172,4 +173,4 @@ def subscribe_to_queue(
     _channel.basic_qos(prefetch_count=1)
     _channel.basic_consume(queue=queue_name, on_message_callback=on_message_callback,
                            auto_ack=auto_ack)
-    return (_connection, _channel)
+    return _connection, _channel

python/idsse_common/setup.py

@@ -18,7 +18,8 @@
         'pint',
         'importlib_metadata',
         'pika',
-        'jsonschema'
+        'jsonschema',
+	    'python-logging-rabbitmq'
       ],
       extras_require={
         'develop': [

python/idsse_common/test/test_rabbitmq_utils.py

@@ -24,7 +24,7 @@
 # Example data objects
 CONN = Conn('localhost', '/', port=5672, username='user', password='password')
 RMQ_PARAMS = RabbitMqParams(
-    Exch('ims_data', 'topic'),
+    Exch('ims_data', 'topic', True),
     Queue('ims_data', '', True, False, True)
 )
 
@@ -91,7 +91,8 @@ def test_connection_params_works(monkeypatch: MonkeyPatch, mock_connection: Mock
     # assert exchange was declared
     _channel.exchange_declare.assert_called_once_with(
         exchange=RMQ_PARAMS.exchange.name,
-        exchange_type=RMQ_PARAMS.exchange.type
+        exchange_type=RMQ_PARAMS.exchange.type,
+        durable=RMQ_PARAMS.exchange.durable
     )
 
     # assert queue was declared and bound