diff --git a/sdk/eventhub/azure-eventhub/CHANGELOG.md b/sdk/eventhub/azure-eventhub/CHANGELOG.md index 026bc3170242..10ef4c81b221 100644 --- a/sdk/eventhub/azure-eventhub/CHANGELOG.md +++ b/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 diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_client_base.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_client_base.py index 4fd65f126086..73b833b54873 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_client_base.py +++ b/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(), diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_producer_client.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_producer_client.py index 6a6cfb181c1a..84b5ea1a5056 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_producer_client.py +++ b/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]: diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_cbs_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_cbs_async.py index c7f4e8c94b59..7757db7f2024 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/aio/_cbs_async.py +++ b/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) diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/cbs.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/cbs.py index 5813475f050b..267283e01759 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_pyamqp/cbs.py +++ b/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: diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py b/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py index 144ff61d3dc6..02ba7f8b776f 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/_version.py +++ b/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" diff --git a/sdk/eventhub/azure-eventhub/azure/eventhub/aio/_client_base_async.py b/sdk/eventhub/azure-eventhub/azure/eventhub/aio/_client_base_async.py index dc6fb2e79054..90a8a9e594ca 100644 --- a/sdk/eventhub/azure-eventhub/azure/eventhub/aio/_client_base_async.py +++ b/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(), diff --git a/sdk/eventhub/azure-eventhub/samples/sync_samples/send.py b/sdk/eventhub/azure-eventhub/samples/sync_samples/send.py index 0173ba5b8e0f..129d429d6473 100644 --- a/sdk/eventhub/azure-eventhub/samples/sync_samples/send.py +++ b/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)