Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,12 @@

### Bugs Fixed

- Fixed a resource leak where replica clients that were no longer part of the auto-failover set were not closed during client refresh.
- Fixed auto-failover replica discovery so that a DNS SRV lookup timeout is distinguished from an empty replica list. A timeout now correctly triggers the longer fallback refresh interval, while an empty result refreshes at the normal interval.
- Fixed a thread-safety issue where iterating, checking membership, or getting the length of the provider's configuration mapping was not guarded by the update lock.
- Fixed `refresh_on` handling so that a single-string watched setting is treated as a key with the default (no) label instead of being incorrectly unpacked character-by-character.
- Fixed a `KeyError` when loading with an endpoint and credential (no connection string).

### Other Changes

- Bumped minimum dependency on `azure-core` to `>=1.31.0`.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ def __init__(self, **kwargs: Any) -> None:
self._startup_timeout: int = kwargs.pop("startup_timeout", DEFAULT_STARTUP_TIMEOUT)

self._replica_client_manager = ConfigurationClientManager(
connection_string=kwargs.pop("connection_string"),
connection_string=kwargs.pop("connection_string", None),
endpoint=kwargs.pop("endpoint"),
credential=kwargs.pop("credential", None),
user_agent=user_agent,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -70,11 +70,13 @@ def is_json_content_type(content_type: str) -> bool:


def _build_watched_setting(setting: Union[str, Tuple[str, str]]) -> Tuple[str, str]:
try:
key, label = setting # type:ignore
except (IndexError, ValueError):
key = str(setting) # Ensure key is a string
label = NULL_CHAR
if isinstance(setting, str):
key, label = setting, NULL_CHAR
else:
try:
key, label = setting
except (TypeError, ValueError):
key, label = str(setting), NULL_CHAR
if "*" in key or "*" in label:
raise ValueError("Wildcard key or label filters are not supported for refresh.")
return key, label
Expand Down Expand Up @@ -264,17 +266,20 @@ def __getitem__(self, key: str) -> Any:
return self._dict[key]

def __iter__(self) -> Iterator[str]:
return self._dict.__iter__()
with self._update_lock:
return self._dict.__iter__()

def __len__(self) -> int:
return len(self._dict)
with self._update_lock:
return len(self._dict)

def __contains__(self, __x: object) -> bool:
# pylint:disable=docstring-missing-param,docstring-missing-return,docstring-missing-rtype
"""
Returns True if the configuration settings contains the specified key.
"""
return self._dict.__contains__(__x)
with self._update_lock:
return self._dict.__contains__(__x)

def keys(self) -> KeysView[str]:
"""
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -132,7 +132,7 @@ def _check_configuration_setting(
self.LOGGER.debug("Refresh all triggered by key: %s label %s.", key, label)
return True, None
else:
raise e
raise
return False, None

@distributed_trace
Expand Down Expand Up @@ -343,7 +343,7 @@ def _validate_snapshot(self, snapshot_name: str) -> bool:
if e.status_code == 404:
self.LOGGER.warning("Snapshot '%s' not found when resolving snapshot.", snapshot_name)
return False
raise e
raise
if snapshot.composition_type != SnapshotComposition.KEY:
raise ValueError(f"Composition type for '{snapshot_name}' must be 'key'.")
return True
Expand Down Expand Up @@ -483,10 +483,10 @@ def refresh_clients(self):
if self._next_update_time and self._next_update_time > time.time():
return

failover_endpoints = find_auto_failover_endpoints(self._original_endpoint, self._replica_discovery_enabled)

if failover_endpoints is None:
# SRV record not found, so we should refresh after a longer interval
try:
failover_endpoints = find_auto_failover_endpoints(self._original_endpoint, self._replica_discovery_enabled)
except TimeoutError:
# SRV record resolution timed out, so we should refresh after a longer interval
self._next_update_time = time.time() + FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL
return

Expand Down Expand Up @@ -530,6 +530,12 @@ def refresh_clients(self):
)
)
self._next_update_time = time.time() + MINIMAL_CLIENT_REFRESH_INTERVAL
# Close any replica clients that are no longer part of the failover.
retained_endpoints = {self._original_client.endpoint}
retained_endpoints.update(client.endpoint for client in discovered_clients)
for client in self._replica_clients:
if client.endpoint not in retained_endpoints:
client.close()
if not self._load_balancing_enabled:
random.shuffle(discovered_clients)
self._replica_clients = [self._original_client] + discovered_clients
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -40,8 +40,8 @@ def find_auto_failover_endpoints(endpoint: str, replica_discovery_enabled: bool)

replicas = _find_replicas(origin.target)

if not replicas:
return None # Timeout
if replicas is None:
raise TimeoutError("Timed out while resolving auto-failover replica endpoints.")
Comment on lines +43 to +44

srv_records = [origin] + replicas
endpoints = []
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ def __init__(self, **kwargs: Any) -> None:
)

if kwargs.get("secret_refresh_interval", 60) < 1:
raise ValueError("Secret refresh interval must be greater than 1 second.")
raise ValueError("Secret refresh interval must be at least 1 second.")

self.secret_refresh_timer: Optional[_RefreshTimer] = (
_RefreshTimer(refresh_interval=kwargs.pop("secret_refresh_interval", 60))
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,7 @@ async def _check_configuration_setting(
self.LOGGER.debug("Refresh all triggered by key: %s label %s.", key, label)
return True, None
else:
raise e
raise
return False, None

@distributed_trace
Expand Down Expand Up @@ -345,7 +345,7 @@ async def _validate_snapshot(self, snapshot_name: str) -> bool:
if e.status_code == 404:
self.LOGGER.warning("Snapshot '%s' not found when resolving snapshot.", snapshot_name)
return False
raise e
raise
if snapshot.composition_type != SnapshotComposition.KEY:
raise ValueError(f"Composition type for '{snapshot_name}' must be 'key'.")
return True
Expand Down Expand Up @@ -485,12 +485,12 @@ async def refresh_clients(self):
if self._next_update_time and self._next_update_time > time.time():
return

failover_endpoints = await find_auto_failover_endpoints(
self._original_endpoint, self._replica_discovery_enabled
)

if failover_endpoints is None:
# SRV record not found, so we should refresh after a longer interval
try:
failover_endpoints = await find_auto_failover_endpoints(
self._original_endpoint, self._replica_discovery_enabled
)
except TimeoutError:
# SRV record resolution timed out, so we should refresh after a longer interval
self._next_update_time = time.time() + FALLBACK_CLIENT_REFRESH_EXPIRED_INTERVAL
return

Expand Down Expand Up @@ -534,6 +534,12 @@ async def refresh_clients(self):
)
)
self._next_update_time = time.time() + MINIMAL_CLIENT_REFRESH_INTERVAL
# Close any replica clients that are no longer part of the failover.
retained_endpoints = {self._original_client.endpoint}
retained_endpoints.update(client.endpoint for client in discovered_clients)
for client in self._replica_clients:
if client.endpoint not in retained_endpoints:
await client.close()
if not self._load_balancing_enabled:
random.shuffle(discovered_clients)
self._replica_clients = [self._original_client] + discovered_clients
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,8 +25,8 @@ async def find_auto_failover_endpoints(endpoint: str, replica_discovery_enabled:

replicas = await _find_replicas(origin.target)

if not replicas:
return None # Timeout
if replicas is None:
raise TimeoutError("Timed out while resolving auto-failover replica endpoints.")

srv_records = [origin] + replicas
endpoints = []
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,9 @@ def __init__(self, endpoint, connection_string, credential, retry_total, retry_b
self.retry_total = retry_total
self.retry_backoff = retry_backoff

async def close(self):
pass


@pytest.mark.usefixtures("caplog")
class TestAsyncConfigurationClientManager:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,9 @@ def __init__(self, endpoint, connection_string, credential, retry_total, retry_b
self.retry_total = retry_total
self.retry_backoff = retry_backoff

async def close(self):
pass


@pytest.mark.usefixtures("caplog")
class TestConfigurationAsyncClientManagerLoadBalance:
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,9 @@ def __init__(self, endpoint, connection_string, credential, retry_total, retry_b
self.retry_total = retry_total
self.retry_backoff = retry_backoff

def close(self):
pass


@pytest.mark.usefixtures("caplog")
class TestConfigurationClientManager(unittest.TestCase):
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,9 @@ def __init__(self, endpoint, connection_string, credential, retry_total, retry_b
self.retry_total = retry_total
self.retry_backoff = retry_backoff

def close(self):
pass


@pytest.mark.usefixtures("caplog")
class TestConfigurationClientManagerLoadBalance(unittest.TestCase):
Expand Down
Loading