Implement exponential backoff + killswitch + sdkstats feature control (#43147)

Author: lzchen

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

@@ -7,6 +7,8 @@
   ([#43032](https://github.com/Azure/azure-sdk-for-python/pull/43032))
 - Adding customer sdkstats to feature statsbeat
   ([#43066](https://github.com/Azure/azure-sdk-for-python/pull/43066))
+- OneSettings control plane: Add killswitch + exponential  backoff + sdkstats feature control
+  ([#43147](https://github.com/Azure/azure-sdk-for-python/pull/43147))
 
 ### Breaking Changes
 
@@ -41,7 +43,6 @@
   ([#42655](https://github.com/Azure/azure-sdk-for-python/pull/42655))
 - Customer Facing SDKStats: Added telemetry_success field to dropped items as per [Spec] - https://github.com/aep-health-and-standards/Telemetry-Collection-Spec/pull/606
   ([#42846](https://github.com/Azure/azure-sdk-for-python/pull/42846))
-### Breaking Changes
 
 ### Bugs Fixed
 - Customer Facing SDKStats: Refactor to use `Manager` and `Singleton` pattern

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

@@ -9,11 +9,14 @@
     _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS,
     _ONE_SETTINGS_CHANGE_URL,
     _ONE_SETTINGS_CONFIG_URL,
+    _ONE_SETTINGS_MAX_REFRESH_INTERVAL_SECONDS,
+    _RETRYABLE_STATUS_CODES,
 )
-from azure.monitor.opentelemetry.exporter._configuration._utils import _ConfigurationProfile
+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__)
 
@@ -79,29 +82,49 @@ def _notify_callbacks(self, settings: Dict[str, str]):
             except Exception as ex:  # pylint: disable=broad-except
                 logger.warning("Callback failed: %s", ex)
 
-    # pylint: disable=too-many-statements
+    def _is_transient_error(self, response: OneSettingsResponse) -> bool:
+        """Check if the response indicates a transient error.
+
+        :param response: OneSettingsResponse object from OneSettings request
+        :type response: OneSettingsResponse
+        :return: True if the error is transient and refresh interval should be increased
+        :rtype: bool
+        """
+        # Check for exception indicator or retryable HTTP status codes
+        return response.has_exception or response.status_code in _RETRYABLE_STATUS_CODES
+
+    # pylint: disable=too-many-statements, too-many-branches
     def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str, str]] = None) -> int:
         """Fetch configuration from OneSettings and update local cache atomically.
         
         This method performs a conditional HTTP request to OneSettings using the
         current ETag for efficient caching. It atomically updates the local configuration
         state with any new settings and manages version tracking for change detection.
         
+        When transient errors are encountered (timeouts, network exceptions, or HTTP status
+        codes 429, 500-504) from the CHANGE endpoint, the method doubles the current refresh 
+        interval to reduce load on the failing service and returns immediately. The refresh 
+        interval is capped at 24 hours (86,400 seconds) to prevent excessively long delays.
+        
         The method implements a check-and-set pattern for thread safety:
         1. Reads current state atomically to prepare request headers
         2. Makes HTTP request to OneSettings CHANGE endpoint outside locks
-        3. Re-reads current state to make version comparison decisions
-        4. Conditionally fetches from CONFIG endpoint if version increased
-        5. Updates all state fields atomically in a single operation
+        3. If transient error (including timeouts/exceptions), doubles refresh interval
+        (capped at 24 hours) and returns immediately
+        4. Re-reads current state to make version comparison decisions
+        5. Conditionally fetches from CONFIG endpoint if version increased
+        6. Updates all state fields atomically in a single operation
         
         Version comparison logic:
         - Version increase: New configuration available, fetches and caches new settings
         - Version same: No changes detected, ETag and refresh interval updated safely
         - Version decrease: Unexpected rollback state, logged as warning, no updates applied
         
         Error handling:
+        - Transient errors (timeouts, exceptions, retryable HTTP codes) from CHANGE endpoint:
+        Refresh interval doubled (capped), immediate return
         - CONFIG endpoint failure: ETag not updated to preserve retry capability on next call
-        - Network failures: Handled by make_onesettings_request, returns default values
+        - Network failures: Handled by make_onesettings_request with error indicators
         - Missing settings/version: Logged as warning, only ETag and refresh interval updated
 
         :param query_dict: Optional query parameters to include in the OneSettings request.
@@ -110,8 +133,9 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
         :type query_dict: Optional[Dict[str, str]]
 
         :return: Updated refresh interval in seconds for the next configuration check.
-            This value comes from the OneSettings response and determines how frequently
-            the background worker should call this method.
+            This value comes from the OneSettings response or is doubled (capped at 24 hours)
+            if transient errors are encountered from the CHANGE endpoint, determining how 
+            frequently the background worker should call this method.
         :rtype: int
         
         Thread Safety:
@@ -134,6 +158,12 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
             All configuration state (ETag, refresh interval, version, settings) is updated
             atomically using immutable state objects. This prevents race conditions where
             different threads might observe inconsistent combinations of these values.
+            
+        Transient Error Handling:
+            When transient errors are detected from the CHANGE endpoint (including timeouts,
+            network exceptions, or retryable HTTP status codes), the refresh interval is
+            doubled and the method returns immediately, preserving current state for retry.
+            The refresh interval is capped at 24 hours to ensure eventual recovery attempts.
         """
         query_dict = query_dict or {}
         headers = {}
@@ -149,11 +179,27 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
         # Make the OneSettings request
         response = make_onesettings_request(_ONE_SETTINGS_CHANGE_URL, query_dict, headers)
 
+        # Check for transient errors from CHANGE endpoint - return immediately if found
+        if self._is_transient_error(response):
+            with self._state_lock:
+                # Double the refresh interval and cap it at 24 hours
+                doubled_interval = self._current_state.refresh_interval * 2
+                current_refresh_interval = min(doubled_interval, _ONE_SETTINGS_MAX_REFRESH_INTERVAL_SECONDS)
+
+            # Create appropriate log message based on error type
+            if response.has_exception:
+                error_description = "network error"
+            else:
+                error_description = f"HTTP {response.status_code}"
+
+            logger.warning("OneSettings CHANGE request failed with transient error (%s). Retrying. ", error_description)
+            return current_refresh_interval  # type: ignore
+
         # Prepare new state updates
         new_state_updates = {}
         if response.etag is not None:
             new_state_updates['etag'] = response.etag
-        if response.refresh_interval and response.refresh_interval > 0:
+        if response.refresh_interval and response.refresh_interval > 0:  # type: ignore
             new_state_updates['refresh_interval'] = response.refresh_interval  # type: ignore
 
         if response.status_code == 304:
@@ -217,14 +263,14 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
         if notify_callbacks and state_for_callbacks is not None and state_for_callbacks.settings_cache:
             self._notify_callbacks(state_for_callbacks.settings_cache)
 
-        return current_refresh_interval
+        return current_refresh_interval  # type: ignore
 
     def get_settings(self) -> Dict[str, str]:  # pylint: disable=C4741,C4742
         """Get current settings cache."""
         with self._state_lock:
             return self._current_state.settings_cache.copy()  # type: ignore
 
-    def get_current_version(self) -> int:  # pylint: disable=C4741,C4742
+    def get_current_version(self) -> int:    # type: ignore # pylint: disable=C4741,C4742
         """Get current version."""
         with self._state_lock:
             return self._current_state.version_cache  # type: ignore

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

@@ -4,23 +4,29 @@
 
 This module provides global access functions for the Configuration Manager singleton.
 """
+import os
+from typing import Optional, TYPE_CHECKING
 
-from typing import TYPE_CHECKING
+from azure.monitor.opentelemetry.exporter._constants import _APPLICATIONINSIGHTS_CONTROLPLANE_DISABLED
 
 if TYPE_CHECKING:
     from azure.monitor.opentelemetry.exporter._configuration import _ConfigurationManager
 
 # Global singleton instance for easy access throughout the codebase
 _configuration_manager = None
 
-def get_configuration_manager() -> "_ConfigurationManager":
+def get_configuration_manager() -> Optional["_ConfigurationManager"]:
     """Get the global Configuration Manager singleton instance.
 
     This provides a single access point to the manager and handles lazy initialization.
+    Returns None if control plane functionality is disabled via environment variable.
 
-    :return: The singleton Configuration Manager instance
-    :rtype: _ConfigurationManager
+    :return: The singleton Configuration Manager instance, or None if disabled
+    :rtype: Optional[_ConfigurationManager]
     """
+    disabled = os.environ.get(_APPLICATIONINSIGHTS_CONTROLPLANE_DISABLED)
+    if disabled is not None and disabled.lower() == "true":
+        return None
     global _configuration_manager  # pylint: disable=global-statement
     if _configuration_manager is None:
         from azure.monitor.opentelemetry.exporter._configuration import _ConfigurationManager

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

@@ -46,14 +46,15 @@ class OneSettingsResponse:
     """Response object containing OneSettings API response data.
 
     This class encapsulates the parsed response from a OneSettings API call,
-    including configuration settings, version information, and metadata.
+    including configuration settings, version information, error indicators and metadata.
 
     Attributes:
         etag (Optional[str]): ETag header value for caching and conditional requests
         refresh_interval (int): Interval in seconds for the next configuration refresh
         settings (Dict[str, str]): Dictionary of configuration key-value pairs
         version (Optional[int]): Configuration version number for change tracking
         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.)
     """
 
     def __init__(
@@ -62,7 +63,8 @@ def __init__(
         refresh_interval: int = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS,
         settings: Optional[Dict[str, str]] = None,
         version: Optional[int] = None,
-        status_code: int = 200
+        status_code: int = 200,
+        has_exception: bool = False
     ):
         """Initialize OneSettingsResponse with configuration data.
 
@@ -74,12 +76,14 @@ def __init__(
                 Defaults to empty dict if None.
             version (Optional[int], optional): Configuration version number. Defaults to 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.settings = settings or {}
         self.version = version
         self.status_code = status_code
+        self.has_exception = has_exception
 
 
 def make_onesettings_request(url: str, query_dict: Optional[Dict[str, str]] = None,
@@ -88,7 +92,7 @@ def make_onesettings_request(url: str, query_dict: Optional[Dict[str, str]] = No
 
     This function handles the complete OneSettings request lifecycle including:
     - Making the HTTP GET request with optional query parameters and headers
-    - Error handling for network, HTTP, and JSON parsing errors
+    - Error handling for network, HTTP, timeout, and JSON parsing errors
     - Parsing the response into a structured OneSettingsResponse object
 
     :param url: The OneSettings API endpoint URL to request
@@ -100,13 +104,13 @@ def make_onesettings_request(url: str, query_dict: Optional[Dict[str, str]] = No
     Common headers include 'If-None-Match' for ETag caching. Defaults to None.
     :type headers: Optional[Dict[str, str]]
 
-    :return: Parsed response containing configuration data and metadata.
-            Returns a default response object if the request fails.
+    :return: Parsed response containing configuration data and metadata, including
+            error indicators for exceptions and timeouts.
     :rtype: OneSettingsResponse
 
     Raises:
         Does not raise exceptions - all errors are caught and logged, returning a
-        default OneSettingsResponse object.
+        OneSettingsResponse object with appropriate error indicators set.
     """
     query_dict = query_dict or {}
     headers = headers or {}
@@ -116,15 +120,18 @@ def make_onesettings_request(url: str, query_dict: Optional[Dict[str, str]] = No
         result.raise_for_status()  # Raises an exception for 4XX/5XX responses
 
         return _parse_onesettings_response(result)
+    except requests.exceptions.Timeout as ex:
+        logger.warning("OneSettings request timed out: %s", str(ex))
+        return OneSettingsResponse(has_exception=True)
     except requests.exceptions.RequestException as ex:
         logger.warning("Failed to fetch configuration from OneSettings: %s", str(ex))
-        return OneSettingsResponse()
+        return OneSettingsResponse(has_exception=True)
     except json.JSONDecodeError as ex:
         logger.warning("Failed to parse OneSettings response: %s", str(ex))
-        return OneSettingsResponse()
+        return OneSettingsResponse(has_exception=True)
     except Exception as ex:  # pylint: disable=broad-exception-caught
         logger.warning("Unexpected error while fetching configuration: %s", str(ex))
-        return OneSettingsResponse()
+        return OneSettingsResponse(has_exception=True)
 
 
 def _parse_onesettings_response(response: requests.Response) -> OneSettingsResponse:

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_constants.py

@@ -79,6 +79,7 @@
 _MICROSOFT_CUSTOM_EVENT_NAME = "microsoft.custom_event.name"
 
 # ONE SETTINGS
+_APPLICATIONINSIGHTS_CONTROLPLANE_DISABLED = "APPLICATIONINSIGHTS_CONTROLPLANE_DISABLED"
 _ONE_SETTINGS_PYTHON_KEY = "python"
 _ONE_SETTINGS_PYTHON_TARGETING = {"namespaces": _ONE_SETTINGS_PYTHON_KEY}
 _ONE_SETTINGS_CHANGE_VERSION_KEY = "CHANGE_VERSION"
@@ -94,6 +95,9 @@
 _ONE_SETTINGS_SUPPORTED_DATA_BOUNDARIES_KEY = "SUPPORTED_DATA_BOUNDARIES"
 _ONE_SETTINGS_FEATURE_LOCAL_STORAGE = "FEATURE_LOCAL_STORAGE"
 _ONE_SETTINGS_FEATURE_LIVE_METRICS = "FEATURE_LIVE_METRICS"
+_ONE_SETTINGS_FEATURE_SDK_STATS = "FEATURE_SDK_STATS"
+# Maximum refresh interval cap (24 hours in seconds)
+_ONE_SETTINGS_MAX_REFRESH_INTERVAL_SECONDS = 24 * 60 * 60  # 86,400 seconds
 
 # Statsbeat
 # (OpenTelemetry metric name, Statsbeat metric name)

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_live_metrics.py

@@ -41,7 +41,9 @@ def enable_live_metrics(**kwargs: Any) -> None:  # pylint: disable=C4758
         # Will only be added if QuickpulseManager is initialized successfully
         # Is a NoOp if _ConfigurationManager not initialized
         config_manager = get_configuration_manager()
-        config_manager.register_callback(get_quickpulse_configuration_callback)
+        # config_manager would be `None` if control plane is disabled
+        if config_manager:
+            config_manager.register_callback(get_quickpulse_configuration_callback)
 
     # We can detect feature usage for statsbeat since we are in an opt-in model currently
     # Once we move to live metrics on-by-default, we will have to check for both explicit usage

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_quickpulse/_manager.py

@@ -274,9 +274,6 @@ def shutdown(self) -> bool:
 
             if shutdown_success:
                 _set_global_quickpulse_state(_QuickpulseState.OFFLINE)
-                # Clear the singleton instance from the metaclass
-                if self.__class__ in _QuickpulseManager._instances:  # pylint: disable=protected-access
-                    del _QuickpulseManager._instances[self.__class__]  # pylint: disable=protected-access
 
             return shutdown_success
 

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py

@@ -173,14 +173,15 @@ def __init__(self, **kwargs: Any) -> None:
         self.client: AzureMonitorClient = AzureMonitorClient(
             host=self._endpoint, connection_timeout=self._timeout, policies=policies, **kwargs
         )
-        self._configuration_manager.initialize(
-            os=_get_os(),
-            rp=_get_rp(),
-            attach=_get_attach_type(),
-            component="ext",
-            version=ext_version,
-            region=self._region,
-        )
+        if self._configuration_manager:
+            self._configuration_manager.initialize(
+                os=_get_os(),
+                rp=_get_rp(),
+                attach=_get_attach_type(),
+                component="ext",
+                version=ext_version,
+                region=self._region,
+            )
         self.storage: Optional[LocalFileStorage] = None
         if not self._disable_offline_storage:
             self.storage = LocalFileStorage(  # pyright: ignore