[Pyamqp] Pyampq debug build Linkedin

sdk/eventhub/azure-eventhub/CHANGELOG.md

@@ -1,14 +1,16 @@
 # Release History
 
-## 5.8.0b5 (Unreleased)
+## 5.8.0b5 (2022-07-19)
 
 ### Features Added
 
 ### Breaking Changes
 
 ### Bugs Fixed
-
+- Bug fixes that would prevent token refresh at regular intervals
+- Pass in the proper kwarg, so that debug levels will be output when requested
 ### Other Changes
+- Specific logging added in to track error scenarios
 
 ## 5.8.0a4 (2022-06-07)
 

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 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.info('update_status In refresh required or OK.')
             is_expired, is_refresh_required = check_expiration_and_refresh_status(self._expires_on, self._refresh_window)
+            _LOGGER.info('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.info('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,12 @@ def close(self):
     def update_token(self):
         self.auth_state = CbsAuthState.IN_PROGRESS
         access_token = self._auth.get_token()
+        if 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/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

@@ -9,15 +9,18 @@
 Examples to show sending events with different options to an Event Hub partition.
 """
 
+import logging
+import sys
 import time
 import os
+from tkinter.tix import Tree
 from azure.eventhub import EventHubProducerClient, EventData
 from azure.eventhub.exceptions import EventHubError
+from azure.identity import DefaultAzureCredential
 
 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 +85,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)
@@ -96,4 +101,4 @@ def send_event_data_list(producer):
     send_event_data_batch_with_properties(producer)
     send_event_data_list(producer)
 
-print("Send messages in {} seconds.".format(time.time() - start_time))
+print("Send messages in {} seconds.".format(time.time() - start_time))