Various Onesettings implementation changes (#48027)

Author: lzchen

sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md

@@ -15,6 +15,8 @@
 - Simplify OneSettings change detection to use ETag-based mechanism instead of change version tracking to reflect spec update
 - Change OneSettings log messages from warning to debug level to reduce noise for users with firewalls
   ([#47949](https://github.com/Azure/azure-sdk-for-python/pull/47949))
+- Harden OneSettings configuration manager and worker: handle non-retryable HTTP errors by slow-polling instead of retrying, fix worker holding its lock across network I/O, make shutdown a soft reset that leaves the singleton reusable, and make callback registration thread-safe and initialization-independent
+  ([#48027](https://github.com/Azure/azure-sdk-for-python/pull/48027))
 
 ## 1.0.0b55 (2026-07-01)
 

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/__init__.py

@@ -10,13 +10,13 @@
     _ONE_SETTINGS_CHANGE_URL,
     _ONE_SETTINGS_CONFIG_URL,
     _ONE_SETTINGS_MAX_REFRESH_INTERVAL_SECONDS,
+    _ONE_SETTINGS_BACKOFF_BASE_SECONDS,
     _RETRYABLE_STATUS_CODES,
 )
 from azure.monitor.opentelemetry.exporter._configuration._utils import _ConfigurationProfile, OneSettingsResponse
 from azure.monitor.opentelemetry.exporter._configuration._utils import make_onesettings_request
 from azure.monitor.opentelemetry.exporter._utils import Singleton
 
-
 # Set up logger
 logger = logging.getLogger(__name__)
 
@@ -26,14 +26,14 @@ class _ConfigurationState:
     """Immutable state object for configuration data."""
 
     etag: str = ""
-    refresh_interval: int = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
+    refresh_interval_s: int = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
     settings_cache: Dict[str, str] = field(default_factory=dict)
 
     def with_updates(self, **kwargs) -> "_ConfigurationState":  # pylint: disable=C4741,C4742
         """Create a new state object with updated values."""
         return _ConfigurationState(
             etag=kwargs.get("etag", self.etag),
-            refresh_interval=kwargs.get("refresh_interval", self.refresh_interval),
+            refresh_interval_s=kwargs.get("refresh_interval_s", self.refresh_interval_s),
             settings_cache=kwargs.get("settings_cache", self.settings_cache.copy()),
         )
 
@@ -46,6 +46,8 @@ def __init__(self):
         self._configuration_worker = None
         self._state_lock = Lock()  # Single lock for all state
         self._current_state = _ConfigurationState()
+        # Consecutive transient-error count for exponential backoff on change-detection (e2) calls
+        self._backoff_attempts = 0
         self._callbacks = []
         self._initialized = False
 
@@ -62,20 +64,24 @@ def initialize(self, **kwargs):
             from azure.monitor.opentelemetry.exporter._configuration._worker import _ConfigurationWorker
 
             # Get initial refresh interval from current state
-            initial_refresh_interval = self._current_state.refresh_interval
+            initial_refresh_interval = self._current_state.refresh_interval_s
 
             self._configuration_worker = _ConfigurationWorker(self, initial_refresh_interval)
             self._initialized = True
 
     def register_callback(self, callback):
-        # Register a callback to be invoked when configuration changes.
-        if not self._initialized:
-            return
+        # Register a callback to be invoked when configuration changes. Registration is independent of
+        # initialize(): the callback simply sits in the list until the worker fires a config change, so
+        # there is no need to guard on initialization state.
         self._callbacks.append(callback)
 
     def _notify_callbacks(self, settings: Dict[str, str]):
         # Notify all registered callbacks of configuration changes.
-        for cb in self._callbacks:
+        # Snapshot the list first so a concurrent register_callback on another thread can't trigger
+        # "list changed size during iteration". list.append and list(...) are individually atomic
+        # under the GIL, so no lock is needed here (and _state_lock stays scoped to config state).
+        callbacks = list(self._callbacks)
+        for cb in callbacks:
             try:
                 cb(settings)
             except Exception as ex:  # pylint: disable=broad-except
@@ -98,12 +104,18 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
 
         This method implements the change detection mechanism per the OneSettings spec:
         - Polls CHANGE endpoint (e2) with cached ETag via if-none-match header.
-          - If 304 Not Modified: no changes, update refresh interval only.
+          - If 304 Not Modified: no settings fetched; the refresh interval (and echoed ETag)
+            are updated from the response headers.
           - If 200 (new ETag or no cached ETag): fetch from CONFIG endpoint (e1) for settings.
+            If the e1 fetch fails, the new ETag is dropped so the change is re-attempted on the
+            next poll rather than being silently marked as applied.
 
         When transient errors are encountered (timeouts, network exceptions, or retryable
-        HTTP status codes) from the CHANGE endpoint, the method doubles the current refresh
-        interval (capped at 24 hours) and returns immediately.
+        HTTP status codes) from the CHANGE endpoint, the method applies progressive exponential
+        backoff from a fixed base (3600s, then 7200s, 14400s, ... capped at 24 hours), tracked
+        via a consecutive-failure counter that resets on the next non-transient response, and
+        returns immediately. Non-retryable HTTP errors keep the cached configuration, do not
+        advance the ETag, and slow-poll at the max interval (24 hours).
 
         :param query_dict: Optional query parameters to include in the OneSettings request.
         :type query_dict: Optional[Dict[str, str]]
@@ -119,32 +131,48 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
             current_state = self._current_state
             if current_state.etag:
                 headers["If-None-Match"] = current_state.etag
-            if current_state.refresh_interval:
-                headers["x-ms-onesetinterval"] = str(current_state.refresh_interval)
+            if current_state.refresh_interval_s:
+                # refresh_interval_s is stored in seconds internally; the header expects minutes.
+                headers["x-ms-onesetinterval"] = str(current_state.refresh_interval_s // 60)
 
         # Poll CHANGE endpoint (e2)
         response = make_onesettings_request(_ONE_SETTINGS_CHANGE_URL, query_dict, headers)
 
-        # Check for transient errors - double interval and return
+        # Check for transient errors - apply exponential backoff and return
         if self._is_transient_error(response):
             with self._state_lock:
-                doubled_interval = self._current_state.refresh_interval * 2
-                current_refresh_interval = min(doubled_interval, _ONE_SETTINGS_MAX_REFRESH_INTERVAL_SECONDS)
+                # Progressive exponential backoff from a fixed base, capped at the max interval.
+                # Schedule (per spec): 1st failure -> 3600s, then 7200s, 14400s, ... up to 86400s.
+                self._backoff_attempts += 1
+                backoff_interval = min(
+                    _ONE_SETTINGS_BACKOFF_BASE_SECONDS * int(2 ** (self._backoff_attempts - 1)),
+                    _ONE_SETTINGS_MAX_REFRESH_INTERVAL_SECONDS,
+                )
 
             if response.has_exception:
                 error_description = "network error"
             else:
                 error_description = f"HTTP {response.status_code}"
 
-            logger.debug("OneSettings CHANGE request failed with transient error (%s). Retrying. ", error_description)
-            return current_refresh_interval  # type: ignore
+            logger.debug(
+                "OneSettings CHANGE request failed with transient error (%s). "
+                "Backing off (attempt %d) for %d seconds.",
+                error_description,
+                self._backoff_attempts,
+                backoff_interval,
+            )
+            return backoff_interval
+
+        # Non-transient response from the CHANGE endpoint counts as a success: reset backoff counter
+        with self._state_lock:
+            self._backoff_attempts = 0
 
         # Prepare new state updates
         new_state_updates: Dict[str, Any] = {}
         if response.etag is not None:
             new_state_updates["etag"] = response.etag
-        if response.refresh_interval and response.refresh_interval > 0:  # type: ignore
-            new_state_updates["refresh_interval"] = response.refresh_interval  # type: ignore
+        if response.refresh_interval_s and response.refresh_interval_s > 0:  # type: ignore
+            new_state_updates["refresh_interval_s"] = response.refresh_interval_s  # type: ignore
 
         if response.status_code == 304:
             # Not modified: no configuration changes published
@@ -159,8 +187,14 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
                 # Do not update etag to allow retry on next call
                 new_state_updates.pop("etag", None)
         else:
-            # Unexpected non-transient status code
-            logger.debug("Unexpected response status from CHANGE endpoint: %d", response.status_code)
+            # Non-retryable HTTP error from the CHANGE endpoint (e.g. 400/404/414). Retryable errors
+            # (network/timeout and _RETRYABLE_STATUS_CODES) are already handled as transient above.
+            # These remaining errors are effectively permanent from the SDK's perspective and won't be
+            # resolved by retrying at the normal cadence, so keep the current cached configuration, do
+            # not advance the ETag, and slow-poll at the max interval. Config fetching is internal, so
+            # this stays silent (debug only) and never surfaces to users.
+            logger.debug("Non-retryable response status from CHANGE endpoint: %d", response.status_code)
+            return _ONE_SETTINGS_MAX_REFRESH_INTERVAL_SECONDS
 
         notify_callbacks = False
         current_refresh_interval = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
@@ -170,7 +204,7 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
         with self._state_lock:
             latest_state = self._current_state
             self._current_state = latest_state.with_updates(**new_state_updates)
-            current_refresh_interval = self._current_state.refresh_interval
+            current_refresh_interval = self._current_state.refresh_interval_s
             if "settings_cache" in new_state_updates:
                 notify_callbacks = True
                 state_for_callbacks = self._current_state
@@ -187,12 +221,18 @@ def get_settings(self) -> Dict[str, str]:  # pylint: disable=C4741,C4742
             return self._current_state.settings_cache.copy()  # type: ignore
 
     def shutdown(self) -> None:
-        """Shutdown the configuration worker."""
+        """Shutdown the configuration worker.
+
+        This is a soft reset (matching the QuickpulseManager convention): the worker is stopped and
+        transient state is cleared, but the singleton instance is left intact and reusable so a later
+        initialize() can restart polling. The cached configuration (_current_state) is intentionally
+        preserved across shutdown so the next initialize() resumes from the cached ETag.
+        """
         if self._configuration_worker:
             self._configuration_worker.shutdown()
             self._configuration_worker = None
+        # Worker thread is now joined, so no callback notifications are in flight and no other thread
+        # is registering callbacks, so we can clear this state directly. Callbacks are cleared so a
+        # subsequent initialize()/re-registration does not accumulate duplicates.
         self._initialized = False
         self._callbacks.clear()
-        # Clear the singleton instance from the metaclass
-        if self.__class__ in _ConfigurationManager._instances:  # pylint: disable=protected-access
-            del _ConfigurationManager._instances[self.__class__]  # pylint: disable=protected-access

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/_state.py

@@ -4,6 +4,7 @@
 
 This module provides global access functions for the Configuration Manager singleton.
 """
+import logging
 import os
 from typing import Optional, TYPE_CHECKING
 
@@ -12,6 +13,8 @@
 if TYPE_CHECKING:
     from azure.monitor.opentelemetry.exporter._configuration import _ConfigurationManager
 
+logger = logging.getLogger(__name__)
+
 # Global singleton instance for easy access throughout the codebase
 _configuration_manager = None
 
@@ -27,6 +30,10 @@ def get_configuration_manager() -> Optional["_ConfigurationManager"]:
     """
     disabled = os.environ.get(_APPLICATIONINSIGHTS_CONTROLPLANE_DISABLED)
     if disabled is not None and disabled.lower() == "true":
+        logger.debug(
+            "OneSettings control plane disabled via %s; using built-in SDK default configuration.",
+            _APPLICATIONINSIGHTS_CONTROLPLANE_DISABLED,
+        )
         return None
     global _configuration_manager  # pylint: disable=global-statement
     if _configuration_manager is None:

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/_utils.py

@@ -51,7 +51,7 @@ class OneSettingsResponse:
 
     Attributes:
         etag (Optional[str]): ETag header value for caching and conditional requests
-        refresh_interval (int): Interval in seconds for the next configuration refresh
+        refresh_interval_s (int): Interval in seconds for the next configuration refresh
         settings (Dict[str, str]): Dictionary of configuration key-value pairs
         status_code (int): HTTP status code from the response
         has_exception (bool): True if the request resulted in a transient error (network error, timeout, etc.)
@@ -60,7 +60,7 @@ class OneSettingsResponse:
     def __init__(
         self,
         etag: Optional[str] = None,
-        refresh_interval: int = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS,
+        refresh_interval_s: int = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS,
         settings: Optional[Dict[str, str]] = None,
         status_code: int = 200,
         has_exception: bool = False,
@@ -69,15 +69,15 @@ def __init__(
 
         Args:
             etag (Optional[str], optional): ETag header value for caching. Defaults to None.
-            refresh_interval (int, optional): Refresh interval in seconds.
+            refresh_interval_s (int, optional): Refresh interval in seconds.
                 Defaults to _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS.
             settings (Optional[Dict[str, str]], optional): Configuration settings dictionary.
                 Defaults to empty dict if None.
             status_code (int, optional): HTTP status code. Defaults to 200.
             has_exception (bool, optional): Indicates if request failed with a transient error. Defaults to False.
         """
         self.etag = etag
-        self.refresh_interval = refresh_interval
+        self.refresh_interval_s = refresh_interval_s
         self.settings = settings or {}
         self.status_code = status_code
         self.has_exception = has_exception
@@ -115,19 +115,20 @@ def make_onesettings_request(
 
     try:
         result = requests.get(url, params=query_dict, headers=headers, timeout=10)
-        result.raise_for_status()  # Raises an exception for 4XX/5XX responses
-
+        # Do NOT call raise_for_status(): HTTP error codes (4xx/5xx) are handled by the parser so
+        # the real status_code is preserved. This lets callers distinguish retryable errors
+        # (see _RETRYABLE_STATUS_CODES) from non-retryable client errors (400/404/414). Only genuine
+        # network/timeout failures below are surfaced as has_exception=True (transient).
         return _parse_onesettings_response(result)
     except requests.exceptions.Timeout as ex:
         logger.debug("OneSettings request timed out: %s", str(ex))
         return OneSettingsResponse(has_exception=True)
     except requests.exceptions.RequestException as ex:
         logger.debug("Failed to fetch configuration from OneSettings: %s", str(ex))
         return OneSettingsResponse(has_exception=True)
-    except json.JSONDecodeError as ex:
-        logger.debug("Failed to parse OneSettings response: %s", str(ex))
-        return OneSettingsResponse(has_exception=True)
     except Exception as ex:  # pylint: disable=broad-exception-caught
+        # _parse_onesettings_response already swallows JSON/decode errors internally, so nothing
+        # here raises json.JSONDecodeError; this catch-all covers any other unexpected failure.
         logger.debug("Unexpected error while fetching configuration: %s", str(ex))
         return OneSettingsResponse(has_exception=True)
 
@@ -143,24 +144,24 @@ def _parse_onesettings_response(response: requests.Response) -> OneSettingsRespo
     The parser handles different HTTP status codes appropriately:
     - 200: New configuration data available, parse settings
     - 304: Not modified, configuration unchanged (empty settings)
-    - 400/404/414/500: Various error conditions, logged with warnings
+    - 400/404/414/500: Various error conditions, logged at debug
 
     :param response: HTTP response object from the requests library containing
         the OneSettings API response with headers, status code, and content.
     :type response: requests.Response
 
     :return: Structured response object containing:
         - etag: ETag header value for conditional requests
-        - refresh_interval: Next refresh interval from headers
+        - refresh_interval_s: Next refresh interval from headers
         - settings: Configuration key-value pairs (empty for 304/errors)
         - status_code: HTTP status code of the response
     :rtype: OneSettingsResponse
     Note:
-        This function logs warnings for various error conditions but does not
-        raise exceptions, always returning a valid OneSettingsResponse object.
+        This function logs various error conditions at debug level (config fetching is internal)
+        but does not raise exceptions, always returning a valid OneSettingsResponse object.
     """
     etag = None
-    refresh_interval = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
+    refresh_interval_s = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
     settings: Dict[str, str] = {}
     status_code = response.status_code
 
@@ -169,12 +170,12 @@ def _parse_onesettings_response(response: requests.Response) -> OneSettingsRespo
         etag = response.headers.get("ETag")
         refresh_interval_header = response.headers.get("x-ms-onesetinterval")
         try:
-            # Note: OneSettings refresh interval is in minutes, convert to seconds
+            # Note: OneSettings refresh interval returned is in minutes, convert to seconds
             if refresh_interval_header:
-                refresh_interval = int(refresh_interval_header) * 60
+                refresh_interval_s = int(refresh_interval_header) * 60
         except (ValueError, TypeError):
             logger.debug("Invalid refresh interval format: %s", refresh_interval_header)
-            refresh_interval = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
+            refresh_interval_s = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
 
     # Handle different status codes
     if status_code == 304:
@@ -198,7 +199,7 @@ def _parse_onesettings_response(response: requests.Response) -> OneSettingsRespo
     elif status_code == 500:
         logger.debug("Internal server error from OneSettings: %s", response.content)
 
-    return OneSettingsResponse(etag, refresh_interval, settings, status_code)
+    return OneSettingsResponse(etag, refresh_interval_s, settings, status_code)
 
 
 # mypy: disable-error-code="no-any-return"