Dynamically reconfigure sdkstats on OneSettings configuration change (#42877)

Author: lzchen

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

@@ -29,6 +29,8 @@
   ([#42508] https://github.com/Azure/azure-sdk-for-python/pull/42508)
 - Refactoring of statsbeat to use `StatsbeatManager`
   ([#42716] https://github.com/Azure/azure-sdk-for-python/pull/42716)
+- Support sdk stats dynamic change on OneSettings config change
+  ([#42877] https://github.com/Azure/azure-sdk-for-python/pull/42877)
 
 ## 1.0.0b41 (2025-07-31)
 

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

@@ -7,11 +7,11 @@
 
 from azure.monitor.opentelemetry.exporter._constants import (
     _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS,
-    _ONE_SETTINGS_PYTHON_KEY,
     _ONE_SETTINGS_CHANGE_URL,
     _ONE_SETTINGS_CONFIG_URL,
 )
 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__)
@@ -35,34 +35,42 @@ def with_updates(self, **kwargs) -> '_ConfigurationState':  # pylint: disable=C4
         )
 
 
-class _ConfigurationManager:
+class _ConfigurationManager(metaclass=Singleton):
     """Singleton class to manage configuration settings."""
 
-    _instance = None
-    _configuration_worker = None
-    _instance_lock = Lock()
-    _state_lock = Lock()  # Single lock for all state
-    _current_state = _ConfigurationState()
+    def __init__(self):
+        """Initialize the ConfigurationManager instance."""
 
-    def __new__(cls):
-        with cls._instance_lock:
-            if cls._instance is None:
-                cls._instance = super(_ConfigurationManager, cls).__new__(cls)
-                # Initialize the instance here to avoid re-initialization
-                cls._instance._initialize_worker()
-            return cls._instance
+        self._configuration_worker = None
+        self._state_lock = Lock()  # Single lock for all state
+        self._current_state = _ConfigurationState()
+        self._callbacks = []
+        self._initialize_worker()
 
     def _initialize_worker(self):
         """Initialize the ConfigurationManager and start the configuration worker."""
         # Lazy import to avoid circular import
         from azure.monitor.opentelemetry.exporter._configuration._worker import _ConfigurationWorker
 
         # Get initial refresh interval from state
-        with _ConfigurationManager._state_lock:
-            initial_refresh_interval = _ConfigurationManager._current_state.refresh_interval
+        with self._state_lock:
+            initial_refresh_interval = self._current_state.refresh_interval
 
-        self._configuration_worker = _ConfigurationWorker(initial_refresh_interval)
+        self._configuration_worker = _ConfigurationWorker(self, initial_refresh_interval)
 
+    def register_callback(self, callback):
+        # Register a callback to be invoked when configuration changes.
+        self._callbacks.append(callback)
+
+    def _notify_callbacks(self, settings: Dict[str, str]):
+        # Notify all registered callbacks of configuration changes.
+        for cb in self._callbacks:
+            try:
+                cb(settings)
+            except Exception as ex:  # pylint: disable=broad-except
+                logger.warning("Callback failed: %s", ex)
+
+    # pylint: disable=too-many-statements
     def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str, str]] = None) -> int:
         """Fetch configuration from OneSettings and update local cache atomically.
         
@@ -122,8 +130,8 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
         headers = {}
 
         # Read current state atomically
-        with _ConfigurationManager._state_lock:
-            current_state = _ConfigurationManager._current_state
+        with self._state_lock:
+            current_state = self._current_state
             if current_state.etag:
                 headers["If-None-Match"] = current_state.etag
             if current_state.refresh_interval:
@@ -145,8 +153,8 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
         # Handle version and settings updates
         elif response.settings and response.version is not None:
             needs_config_fetch = False
-            with _ConfigurationManager._state_lock:
-                current_state = _ConfigurationManager._current_state
+            with self._state_lock:
+                current_state = self._current_state
 
                 if response.version > current_state.version_cache:
                     # Version increase: new config available
@@ -182,34 +190,41 @@ def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str,
             # No settings or version provided
             logger.warning("No settings or version provided in config response. Config not updated.")
 
+
+        notify_callbacks = False
+        current_refresh_interval = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
+        state_for_callbacks = None
+
         # Atomic state update
-        with _ConfigurationManager._state_lock:
-            latest_state = _ConfigurationManager._current_state  # Always use latest state
-            _ConfigurationManager._current_state = latest_state.with_updates(**new_state_updates)
-            return _ConfigurationManager._current_state.refresh_interval
+        with self._state_lock:
+            latest_state = self._current_state  # Always use latest state
+            self._current_state = latest_state.with_updates(**new_state_updates)
+            current_refresh_interval = self._current_state.refresh_interval
+            if 'settings_cache' in new_state_updates:
+                notify_callbacks = True
+                state_for_callbacks = self._current_state
+
+        # Handle configuration updates throughout the SDK
+        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
 
     def get_settings(self) -> Dict[str, str]:  # pylint: disable=C4741,C4742
         """Get current settings cache."""
-        with _ConfigurationManager._state_lock:
-            return _ConfigurationManager._current_state.settings_cache.copy()
+        with self._state_lock:
+            return self._current_state.settings_cache.copy()  # type: ignore
 
     def get_current_version(self) -> int:  # pylint: disable=C4741,C4742
         """Get current version."""
-        with _ConfigurationManager._state_lock:
-            return _ConfigurationManager._current_state.version_cache
+        with self._state_lock:
+            return self._current_state.version_cache  # type: ignore
 
     def shutdown(self) -> None:
         """Shutdown the configuration worker."""
-        with _ConfigurationManager._instance_lock:
-            if self._configuration_worker:
-                self._configuration_worker.shutdown()
-                self._configuration_worker = None
-            if _ConfigurationManager._instance:
-                _ConfigurationManager._instance = None
-
-
-def _update_configuration_and_get_refresh_interval() -> int:
-    targeting = {
-        "namespaces": _ONE_SETTINGS_PYTHON_KEY,
-    }
-    return _ConfigurationManager().get_configuration_and_refresh_interval(targeting)
+        if self._configuration_worker:
+            self._configuration_worker.shutdown()
+            self._configuration_worker = None
+        # Clear the singleton instance from the metaclass
+        if self.__class__ in Singleton._instances:  # pylint: disable=protected-access
+            del Singleton._instances[self.__class__]  # pylint: disable=protected-access

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

@@ -3,6 +3,7 @@
 from typing import Dict, Optional
 import json
 import logging
+# mypy: disable-error-code="import-untyped"
 import requests
 
 from azure.monitor.opentelemetry.exporter._constants import (

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

@@ -4,7 +4,7 @@
 import logging
 import threading
 import random
-from azure.monitor.opentelemetry.exporter._configuration import _update_configuration_and_get_refresh_interval
+from azure.monitor.opentelemetry.exporter._constants import _ONE_SETTINGS_PYTHON_TARGETING
 
 logger = logging.getLogger(__name__)
 
@@ -28,14 +28,15 @@ class _ConfigurationWorker:
         _running (bool): Flag indicating if the worker is currently running
     """
 
-    def __init__(self, refresh_interval=None) -> None:
+    def __init__(self, configuration_manager, refresh_interval=None) -> None:
         """Initialize and start the configuration worker thread.
 
         Creates and starts a background daemon thread that will periodically refresh
         configuration from OneSettings. The thread starts immediately upon initialization
         with a random startup delay to prevent thundering herd issues.
 
         Args:
+            configuration_manager: The ConfigurationManager instance to update
             refresh_interval (Optional[int]): Initial refresh interval in seconds.
                 If None, defaults to 3600 seconds (1 hour).
 
@@ -44,6 +45,7 @@ def __init__(self, refresh_interval=None) -> None:
             0-15 second startup delay to stagger configuration requests across multiple
             SDK instances during startup or recovery from outages.
         """
+        self._configuration_manager = configuration_manager
         self._default_refresh_interval = 3600  # Default to 60 minutes in seconds
         self._interval_lock = threading.Lock()
         self._state_lock = threading.Lock()
@@ -132,9 +134,9 @@ def _get_configuration(self) -> None:
 
         while not self._shutdown_event.is_set():
             try:
-                # Perform the refresh operation
                 with self._interval_lock:
-                    self._refresh_interval = _update_configuration_and_get_refresh_interval()
+                    self._refresh_interval = \
+                    self._configuration_manager.get_configuration_and_refresh_interval(_ONE_SETTINGS_PYTHON_TARGETING)
             except Exception as ex:  # pylint: disable=broad-exception-caught
                 logger.warning("Configuration refresh failed: %s", ex)
 

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

@@ -14,6 +14,14 @@
 # Specs taken from https://tools.ietf.org/html/rfc4122
 uuid_regex_pattern = re.compile("^[0-9a-f]{8}-" "[0-9a-f]{4}-" + "[0-9a-f]{4}-" "[0-9a-f]{4}-" "[0-9a-f]{12}$")
 
+# Pattern to extract region from ingestion endpoint URL
+# Examples:
+# - https://westeurope-5.in.applicationinsights.azure.com/ -> westeurope
+# - https://westeurope.in.applicationinsights.azure.com/ -> westeurope
+# - https://eastus-1.in.applicationinsights.azure.com/ -> eastus
+# - https://dc.services.visualstudio.com -> None (global endpoint)
+region_from_endpoint_pattern = re.compile(r"https://([a-z0-9]+)(?:-\d+)?\.in\.applicationinsights\.azure\.com")
+
 
 class ConnectionStringParser:
     """ConnectionString parser.
@@ -29,6 +37,7 @@ def __init__(self, connection_string: typing.Optional[str] = None) -> None:
         self.live_endpoint = ""
         self._connection_string = connection_string
         self.aad_audience = ""
+        self.region = ""
         self._initialize()
         self._validate_instrumentation_key()
 
@@ -66,6 +75,24 @@ def _initialize(self) -> None:
             code_cs.get(AAD_AUDIENCE) or env_cs.get(AAD_AUDIENCE)  # type: ignore
         )
 
+        # Extract region information
+        self.region = self._extract_region()  # type: ignore
+
+    def _extract_region(self) -> typing.Optional[str]:
+        """Extract region from endpoint URL.
+
+        :return: Extracted region or None if not found
+        :rtype: typing.Optional[str]
+        """
+        # Try to extract region from the ingestion endpoint URL
+        endpoint = self.endpoint
+        if endpoint:
+            match = region_from_endpoint_pattern.match(endpoint)
+            if match:
+                return match.group(1)
+
+        return None
+
     def _validate_instrumentation_key(self) -> None:
         """Validates the instrumentation key used for Azure Monitor.
 

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

@@ -80,6 +80,7 @@
 
 # ONE SETTINGS
 _ONE_SETTINGS_PYTHON_KEY = "python"
+_ONE_SETTINGS_PYTHON_TARGETING = {"namespaces": _ONE_SETTINGS_PYTHON_KEY}
 _ONE_SETTINGS_CHANGE_VERSION_KEY = "CHANGE_VERSION"
 _ONE_SETTINGS_CNAME = "https://settings.sdk.monitor.azure.com"
 _ONE_SETTINGS_PATH = "/AzMonSDKDynamicConfiguration"
@@ -88,6 +89,11 @@
 _ONE_SETTINGS_CHANGE_URL = _ONE_SETTINGS_CNAME + _ONE_SETTINGS_CHANGE_PATH
 _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS = 3600  # 60 minutes
 
+## ONE SETTINGS CONFIGS
+_ONE_SETTINGS_DEFAULT_STATS_CONNECTION_STRING_KEY = "DEFAULT_STATS_CONNECTION_STRING"
+_ONE_SETTINGS_SUPPORTED_DATA_BOUNDARIES_KEY = "SUPPORTED_DATA_BOUNDARIES"
+_ONE_SETTINGS_FEATURE_LOCAL_STORAGE = "FEATURE_LOCAL_STORAGE"
+
 # Statsbeat
 # (OpenTelemetry metric name, Statsbeat metric name)
 # Note: OpenTelemetry SDK normalizes metric names to lowercase, so first element should be lowercase

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

@@ -14,7 +14,7 @@ def __init__(self):
         self.call_on_emit = hasattr(super(), 'on_emit')
 
     def on_emit(self, log_data: LogData) -> None:  # type: ignore
-        qpm = _QuickpulseManager._instance
+        qpm = _QuickpulseManager()
         if qpm:
             qpm._record_log_record(log_data)
         if self.call_on_emit:
@@ -37,7 +37,7 @@ def force_flush(self, timeout_millis: int = 30000):
 class _QuickpulseSpanProcessor(SpanProcessor):
 
     def on_end(self, span: ReadableSpan) -> None:
-        qpm = _QuickpulseManager._instance
+        qpm = _QuickpulseManager()
         if qpm:
             qpm._record_span(span)
-        return super().on_end(span)
+        return super().on_end(span)  # type: ignore

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

@@ -379,15 +379,22 @@ def _get_scope(aad_audience=None):
 
 
 class Singleton(type):
-    _instance = None
+    """Metaclass for creating thread-safe singleton instances.
+    
+    Supports multiple singleton classes by maintaining a separate instance
+    for each class that uses this metaclass.
+    """
+    _instances = {}  # type: ignore
     _lock = threading.Lock()
 
-    def __call__(cls, *args: Any, **kwargs: Any):
-        if not cls._instance:
+    def __call__(cls, *args: Any, **kwargs: Any) -> Any:
+        if cls not in cls._instances:
             with cls._lock:
-                if not cls._instance:
-                    cls._instance = super(Singleton, cls).__call__(*args, **kwargs)
-        return cls._instance
+                # Double-check pattern to avoid race conditions
+                if cls not in cls._instances:
+                    instance = super().__call__(*args, **kwargs)
+                    cls._instances[cls] = instance  # type: ignore
+        return cls._instances[cls]
 
 def _get_telemetry_type(item: TelemetryItem):
     if hasattr(item, "data") and item.data is not None:

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

@@ -99,6 +99,7 @@ def __init__(self, **kwargs: Any) -> None:
         parsed_connection_string = ConnectionStringParser(kwargs.get("connection_string"))
 
         # TODO: Re-add this once all parts of OneSettings is finished
+        # Initialize the configuration manager to begin fetching settings
         # self._configuration_manager = _ConfigurationManager()
 
         self._api_version = kwargs.get("api_version") or _SERVICE_API_LATEST
@@ -111,6 +112,7 @@ def __init__(self, **kwargs: Any) -> None:
         self._consecutive_redirects = 0  # To prevent circular redirects
         self._disable_offline_storage = kwargs.get("disable_offline_storage", False)
         self._endpoint = parsed_connection_string.endpoint
+        self._region = parsed_connection_string.region
         self._instrumentation_key = parsed_connection_string.instrumentation_key
         self._aad_audience = parsed_connection_string.aad_audience
         self._storage_maintenance_period = kwargs.get(