[ServiceBus/EventHub] add service specific message annotations to receiver logs (Azure#38090)

* [ServiceBus] add service specific message annotations to receiver logs

* add to eventhub + update readme logging to include thread formatting

* only log service annotations header props

* move logging to sdk layer

* move msg logging into eh consumer

* remove sb pyamqp recvr logs

* black

Author: swathipil

sdk/eventhub/azure-eventhub/CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Other Changes
 
+- Added logging to track received messages.
+
 ## 5.12.2 (2024-10-02)
 
 ### Bugs Fixed

sdk/eventhub/azure-eventhub/README.md

@@ -453,6 +453,8 @@ import logging
 import sys
 
 handler = logging.StreamHandler(stream=sys.stdout)
+log_fmt = logging.Formatter(fmt="%(asctime)s | %(threadName)s | %(levelname)s | %(name)s | %(message)s")
+handler.setFormatter(log_fmt)
 logger = logging.getLogger('azure.eventhub')
 logger.setLevel(logging.DEBUG)
 logger.addHandler(handler)

sdk/eventhub/azure-eventhub/azure/eventhub/_consumer.py

@@ -164,6 +164,13 @@ def _next_message_in_buffer(self):
         # pylint:disable=protected-access
         message = self._message_buffer.popleft()
         event_data = EventData._from_message(message)
+        if self._client._config.network_tracing and _LOGGER.isEnabledFor(logging.DEBUG):
+            _LOGGER.debug(
+                "Received message: seq-num: %r, offset: %r, partition-key: %r",
+                event_data.sequence_number,
+                event_data.offset,
+                event_data.partition_key,
+            )
         if self._amqp_transport.KIND == "uamqp":
             event_data._uamqp_message = message
         self._last_received_event = event_data

sdk/eventhub/azure-eventhub/azure/eventhub/aio/_consumer_async.py

@@ -159,6 +159,13 @@ def _next_message_in_buffer(self):
         # pylint:disable=protected-access
         message = self._message_buffer.popleft()
         event_data = EventData._from_message(message)
+        if self._client._config.network_tracing and _LOGGER.isEnabledFor(logging.DEBUG):
+            _LOGGER.debug(
+                "Received message: seq-num: %r, offset: %r, partition-key: %r",
+                event_data.sequence_number,
+                event_data.offset,
+                event_data.partition_key,
+            )
         if self._amqp_transport.KIND == "uamqp":
             event_data._uamqp_message = message
         self._last_received_event = event_data

sdk/servicebus/azure-servicebus/CHANGELOG.md

@@ -12,6 +12,8 @@
 
 ### Other Changes
 
+- Added logging to track received messages.
+
 ## 7.12.3 (2024-09-19)
 
 ### Bugs Fixed

sdk/servicebus/azure-servicebus/README.md

@@ -424,6 +424,8 @@ import logging
 import sys
 
 handler = logging.StreamHandler(stream=sys.stdout)
+log_fmt = logging.Formatter(fmt="%(asctime)s | %(threadName)s | %(levelname)s | %(name)s | %(message)s")
+handler.setFormatter(log_fmt)
 logger = logging.getLogger('azure.servicebus')
 logger.setLevel(logging.DEBUG)
 logger.addHandler(handler)

sdk/servicebus/azure-servicebus/azure/servicebus/_transport/_pyamqp_transport.py

@@ -10,6 +10,7 @@
 import datetime
 from datetime import timezone
 from typing import Optional, Tuple, cast, List, TYPE_CHECKING, Any, Callable, Dict, Union, Iterator, Type
+import logging
 
 from .._pyamqp import (
     utils,
@@ -109,6 +110,8 @@
     from .._pyamqp.client import AMQPClient
     from .._pyamqp.message import MessageDict
 
+_LOGGER = logging.getLogger(__name__)
+
 
 class _ServiceBusErrorPolicy(RetryPolicy):
 
@@ -720,6 +723,15 @@ def build_received_message(
             amqp_transport=receiver._amqp_transport,
         )
         receiver._last_received_sequenced_number = message.sequence_number
+        if receiver._config.logging_enable and _LOGGER.isEnabledFor(logging.DEBUG):
+            _LOGGER.debug(
+                "Received message: seq-num: %r, enqd-utc: %r, lockd-til-utc: %r, ttl: %r, dlvry-cnt: %r",
+                message.sequence_number,
+                message.enqueued_time_utc,
+                message.locked_until_utc,
+                message.time_to_live,
+                message.delivery_count,
+            )
         return message
 
     @staticmethod