[Pyamqp] Pyampq debug build Linkedin

sdk/eventhub/azure-eventhub/CHANGELOG.md

@@ -1,15 +1,16 @@
 # Release History
 
-## 5.8.0b5 (Unreleased)
-
-### Features Added
-
-### Breaking Changes
+## 5.8.0a5 (2022-07-19)
 
 ### Bugs Fixed
 
+- Fixed bug that prevented token refresh at regular intervals.
+- Fixed bug that was improperly passing the debug keyword argument, so that network trace debug logs are output when requested.
+
 ### Other Changes
 
+- Added logging added in to track proper token refreshes & fetches, output exception reason for producer init failure.
+
 ## 5.8.0a4 (2022-06-07)
 
 ### Features Added

sdk/eventhub/azure-eventhub/azure/eventhub/_client_base.py

@@ -10,7 +10,7 @@
 import functools
 import collections
 from typing import Any, Dict, Tuple, List, Optional, TYPE_CHECKING, cast, Union
-from datetime import timedelta
+from datetime import timedelta, datetime
 from urllib.parse import urlparse
 import six
 
@@ -386,7 +386,19 @@ def _management_request(self, mgmt_msg, op_type):
                 mgmt_client.open()
                 while not mgmt_client.client_ready():
                     time.sleep(0.05)
-                mgmt_msg.application_properties["security_token"] = mgmt_auth.get_token()
+                access_token = mgmt_auth.get_token()
+
+                if not access_token:
+                    _LOGGER.info("Management client received an empty access token object")
+
+                elif not access_token.token:
+                    _LOGGER.info("Management client received an empty token")
+
+                else:
+                    _LOGGER.info(f"Management client token expires on: {datetime.fromtimestamp(access_token.expires_on)}")
+
+                mgmt_msg.application_properties["security_token"] = access_token.token
+
                 response = mgmt_client.mgmt_request(
                     mgmt_msg,
                     operation=READ_OPERATION.decode(),

sdk/eventhub/azure-eventhub/azure/eventhub/_producer_client.py

@@ -117,6 +117,7 @@ def __exit__(self, *args):
     def _get_partitions(self):
         # type: () -> None
         if not self._partition_ids:
+            _LOGGER.info("Populating partition IDs so producers can be started.")
             self._partition_ids = self.get_partition_ids()  # type: ignore
             for p_id in cast(List[str], self._partition_ids):
                 self._producers[p_id] = None
@@ -311,7 +312,9 @@ def send_batch(self, event_data_batch, **kwargs):
             cast(EventHubProducer, self._producers[partition_id]).send(
                 to_send_batch, timeout=send_timeout
             )
-        except (KeyError, AttributeError, EventHubError):
+        except (KeyError, AttributeError, EventHubError) as e:
+            _LOGGER.info(
+                "Producer for partition ID '{}' not available: {}. Rebuilding new producer.".format(partition_id, e))
             self._start_producer(partition_id, send_timeout)
             cast(EventHubProducer, self._producers[partition_id]).send(
                 to_send_batch, timeout=send_timeout
@@ -431,6 +434,7 @@ def close(self):
                 :caption: Close down the client.
 
         """
+        _LOGGER.info("Closing ProducerClient")
         with self._lock:
             for pid in self._producers:
                 if self._producers[pid]:

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_cbs_async.py

@@ -175,6 +175,8 @@ async def close(self):
     async def update_token(self):
         self.auth_state = CbsAuthState.IN_PROGRESS
         access_token = await self._auth.get_token()
+        if not access_token.token:
+            _LOGGER.debug("update_token received an empty token")
         self._expires_on = access_token.expires_on
         expires_in = self._expires_on - int(utc_now().timestamp())
         self._refresh_window = int(float(expires_in) * 0.1)

sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/cbs.py

@@ -152,12 +152,15 @@ def _on_execute_operation_complete(
 
     def _update_status(self):
         if self.auth_state == CbsAuthState.OK or self.auth_state == CbsAuthState.REFRESH_REQUIRED:
+            _LOGGER.debug('update_status In refresh required or OK.')
             is_expired, is_refresh_required = check_expiration_and_refresh_status(self._expires_on, self._refresh_window)
+            _LOGGER.debug('is expired == %r, is refresh required == %r', is_expired, is_refresh_required)
             if is_expired:
                 self.auth_state = CbsAuthState.EXPIRED
             elif is_refresh_required:
                 self.auth_state = CbsAuthState.REFRESH_REQUIRED
         elif self.auth_state == CbsAuthState.IN_PROGRESS:
+            _LOGGER.debug('In update status, in progress. token put time: %r', self._token_put_time)
             put_timeout = check_put_timeout_status(self._auth_timeout, self._token_put_time)
             if put_timeout:
                 self.auth_state = CbsAuthState.TIMEOUT
@@ -186,7 +189,14 @@ def close(self):
     def update_token(self):
         self.auth_state = CbsAuthState.IN_PROGRESS
         access_token = self._auth.get_token()
+        if not access_token:
+            _LOGGER.info("Update_token received an empty token object")
+        elif not access_token.token:
+            _LOGGER.info("Update_token received an empty token")
         self._expires_on = access_token.expires_on
+        _LOGGER.info('Update_token after token has been updated')
+        _LOGGER.info('Current time: %r', datetime.now())
+        _LOGGER.info('Token expiry: %r', datetime.fromtimestamp(self._expires_on))
         expires_in = self._expires_on - int(utc_now().timestamp())
         self._refresh_window = int(float(expires_in) * 0.1)
         try:

sdk/eventhub/azure-eventhub/azure/eventhub/_version.py

@@ -3,4 +3,4 @@
 # Licensed under the MIT License.
 # ------------------------------------
 
-VERSION = "5.8.0b5"
+VERSION = "5.8.0a5"

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

@@ -265,7 +265,12 @@ async def _management_request_async(self, mgmt_msg: Message, op_type: bytes) ->
                 await mgmt_client.open_async()
                 while not (await mgmt_client.client_ready_async()):
                     await asyncio.sleep(0.05)
-                mgmt_msg.application_properties["security_token"] = await mgmt_auth.get_token()
+                access_token = await mgmt_auth.get_token()
+                mgmt_msg.application_properties["security_token"] = access_token.token
+
+                if not access_token.token:
+                    _LOGGER.info("update_token received an empty token")
+
                 response = await mgmt_client.mgmt_request_async(
                     mgmt_msg,
                     operation=READ_OPERATION.decode(),

sdk/eventhub/azure-eventhub/samples/sync_samples/send.py

@@ -8,7 +8,6 @@
 """
 Examples to show sending events with different options to an Event Hub partition.
 """
-
 import time
 import os
 from azure.eventhub import EventHubProducerClient, EventData
@@ -17,7 +16,6 @@
 CONNECTION_STR = os.environ['EVENT_HUB_CONN_STR']
 EVENTHUB_NAME = os.environ['EVENT_HUB_NAME']
 
-
 def send_event_data_batch(producer):
     # Without specifying partition_id or partition_key
     # the events will be distributed to available partitions via round-robin.
@@ -82,11 +80,13 @@ def send_event_data_list(producer):
         print("Sending error: ", eh_err)
 
 
+
 producer = EventHubProducerClient.from_connection_string(
     conn_str=CONNECTION_STR,
     eventhub_name=EVENTHUB_NAME
 )
 
+
 start_time = time.time()
 with producer:
     send_event_data_batch(producer)