Implement Onesettings control plane - part 1 (#42360)

Author: lzchen

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

@@ -14,6 +14,8 @@
 
 ### Features Added
 
+- Configuration manager/worker fetch via OneSettings part 1
+  ([#42360] https://github.com/Azure/azure-sdk-for-python/pull/42360)
 - Added RateLimited Sampler 
   ([#41954](https://github.com/Azure/azure-sdk-for-python/pull/41954))
 - Refactored Application Insights Sampler Code

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

@@ -0,0 +1,125 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+from typing import Dict, Optional
+import logging
+from threading import Lock
+
+from azure.monitor.opentelemetry.exporter._constants import (
+    _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS,
+    _ONE_SETTINGS_PYTHON_KEY,
+    _ONE_SETTINGS_CHANGE_URL,
+)
+from azure.monitor.opentelemetry.exporter._configuration._utils import make_onesettings_request
+
+# Set up logger
+logger = logging.getLogger(__name__)
+
+
+class _ConfigurationManager:
+    """Singleton class to manage configuration settings."""
+
+    _instance = None
+    _configuration_worker = None
+    _instance_lock = Lock()
+    _config_lock = Lock()
+    _settings_lock = Lock()
+    _version_lock = Lock()
+    _etag = None
+    _refresh_interval = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
+    _settings_cache: Dict[str, str] = {}
+    _version_cache = 0
+
+    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
+
+    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
+        self._configuration_worker = _ConfigurationWorker(self._refresh_interval)
+
+    def get_configuration_and_refresh_interval(self, query_dict: Optional[Dict[str, str]] = None) -> int:
+        """Fetch configuration from OneSettings and update local cache.
+        
+        This method performs a conditional HTTP request to OneSettings using the
+        current ETag for efficient caching. It updates the local configuration
+        cache with any new settings and manages version tracking for change detection.
+        
+        The method handles version comparison logic:
+        - Version increase: New configuration available, cache updated
+        - Version same: No changes, cache remains unchanged  
+        - Version decrease: Unexpected state, logged as warning
+
+        :param query_dict: Optional query parameters to include
+            in the OneSettings request. Commonly used for targeting specific
+            configuration namespaces or environments.
+        :type query_dict: Optional[Dict[str, str]]
+
+        :return: Updated refresh interval in seconds for the next configuration check.
+        :rtype: int
+        
+        Thread Safety:
+            This method is thread-safe and uses multiple locks to ensure consistent
+            state across concurrent access to configuration data.
+        
+        Note:
+            The method automatically handles ETag-based conditional requests to
+            minimize unnecessary data transfer when configuration hasn't changed.
+        """
+        query_dict = query_dict or {}
+        headers = {}
+
+        # Prepare headers with current etag and refresh interval
+        with self._config_lock:
+            if self._etag:
+                headers["If-None-Match"] = self._etag
+            if self._refresh_interval:
+                headers["x-ms-onesetinterval"] = str(self._refresh_interval)
+
+        # Make the OneSettings request
+        response = make_onesettings_request(_ONE_SETTINGS_CHANGE_URL, query_dict, headers)
+
+        # Update configuration state based on response
+        with self._config_lock:
+            self._etag = response.etag
+            self._refresh_interval = response.refresh_interval
+
+        # Evaluate CONFIG_VERSION to see if we need to fetch new config
+        if response.settings:
+            with self._version_lock:
+                if response.version is not None:
+                    # New config published successfully, make a call to config endpoint
+                    if response.version > self._version_cache:
+                        # TODO: Call config endpoint to pull new config
+                        # Update latest version
+                        self._version_cache = response.version
+                    elif response.version == self._version_cache:
+                        # No new config has been published, do nothing
+                        pass
+                    else:
+                        # Erroneous state, should not occur under normal circumstances
+                        logger.warning(
+                            "Latest `CHANGE_VERSION` is less than the current stored version," \
+                            " no configurations updated."
+                        )
+        return self._refresh_interval
+
+    def shutdown(self) -> None:
+        """Shutdown the configuration worker."""
+        with self._instance_lock:
+            if self._configuration_worker:
+                self._configuration_worker.shutdown()
+                self._configuration_worker = None
+            self._instance = None
+
+
+def _update_configuration_and_get_refresh_interval() -> int:
+    targeting = {
+        "namespaces": _ONE_SETTINGS_PYTHON_KEY,
+    }
+    return _ConfigurationManager().get_configuration_and_refresh_interval(targeting)

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

@@ -0,0 +1,165 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+from typing import Dict, Optional
+import json
+import logging
+import requests
+
+from azure.monitor.opentelemetry.exporter._constants import (
+    _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS,
+    _ONE_SETTINGS_CHANGE_VERSION_KEY,
+)
+
+logger = logging.getLogger(__name__)
+
+
+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.
+
+    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
+    """
+
+    def __init__(
+        self,
+        etag: Optional[str] = None,
+        refresh_interval: int = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS,
+        settings: Optional[Dict[str, str]] = None,
+        version: Optional[int] = None
+    ):
+        """Initialize OneSettingsResponse with configuration data.
+
+        Args:
+            etag (Optional[str], optional): ETag header value for caching. Defaults to None.
+            refresh_interval (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.
+            version (Optional[int], optional): Configuration version number. Defaults to None.
+        """
+        self.etag = etag
+        self.refresh_interval = refresh_interval
+        self.settings = settings or {}
+        self.version = version
+
+
+def make_onesettings_request(url: str, query_dict: Optional[Dict[str, str]] = None,
+                           headers: Optional[Dict[str, str]] = None) -> OneSettingsResponse:
+    """Make an HTTP request to the OneSettings API and parse the response.
+
+    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
+    - Parsing the response into a structured OneSettingsResponse object
+
+    :param url: The OneSettings API endpoint URL to request
+    :type url: str
+    :param query_dict: Query parameters to include
+        in the request URL. Defaults to None.
+    :type query_dict: Optional[Dict[str, str]]
+    :param headers: HTTP headers to include in the request.
+    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.
+    :rtype: OneSettingsResponse
+
+    Raises:
+        Does not raise exceptions - all errors are caught and logged, returning a
+        default OneSettingsResponse object.
+    """
+    query_dict = query_dict or {}
+    headers = headers or {}
+
+    try:
+        result = requests.get(url, params=query_dict, headers=headers, timeout=10)
+        result.raise_for_status()  # Raises an exception for 4XX/5XX responses
+
+        return _parse_onesettings_response(result)
+    except requests.exceptions.RequestException as ex:
+        logger.warning("Failed to fetch configuration from OneSettings: %s", str(ex))
+        return OneSettingsResponse()
+    except json.JSONDecodeError as ex:
+        logger.warning("Failed to parse OneSettings response: %s", str(ex))
+        return OneSettingsResponse()
+    except Exception as ex:  # pylint: disable=broad-exception-caught
+        logger.warning("Unexpected error while fetching configuration: %s", str(ex))
+        return OneSettingsResponse()
+
+def _parse_onesettings_response(response: requests.Response) -> OneSettingsResponse:
+    """Parse an HTTP response from OneSettings into a structured response object.
+
+    This function processes the OneSettings API response and extracts:
+    - HTTP headers (ETag, refresh interval)
+    - Response body (configuration settings, version)
+    - Status code handling (200, 304, 4xx, 5xx)
+
+    The parser handles different HTTP status codes appropriately:
+    - 200: New configuration data available, parse settings and version
+    - 304: Not modified, configuration unchanged (empty settings)
+    - 400/404/414/500: Various error conditions, logged with warnings
+
+    :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
+        - settings: Configuration key-value pairs (empty for 304/errors)
+        - version: Configuration version number for change tracking
+    :rtype: OneSettingsResponse
+    Note:
+        This function logs warnings for various error conditions but does not
+        raise exceptions, always returning a valid OneSettingsResponse object.
+    """
+    etag = None
+    refresh_interval = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
+    settings: Dict[str, str] = {}
+    status_code = response.status_code
+    version = None
+
+    # Extract headers
+    if response.headers:
+        etag = response.headers.get("ETag")
+        refresh_interval_header = response.headers.get("x-ms-onesetinterval")
+        try:
+            refresh_interval = int(refresh_interval_header) if refresh_interval_header else refresh_interval
+        except (ValueError, TypeError):
+            logger.warning("Invalid refresh interval format: %s", refresh_interval_header)
+            refresh_interval = _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS
+
+    # Handle different status codes
+    if status_code == 304:
+        # 304 Not Modified - cache stays the same
+        pass
+    elif status_code == 200:
+        # 200 OK - parse new settings
+        if response.content:
+            try:
+                decoded_string = response.content.decode("utf-8")
+                config = json.loads(decoded_string)
+                settings = config.get("settings", {})
+                if settings and settings.get(_ONE_SETTINGS_CHANGE_VERSION_KEY) is not None:
+                    version = int(settings.get(_ONE_SETTINGS_CHANGE_VERSION_KEY))  # type: ignore
+            except (UnicodeDecodeError, json.JSONDecodeError) as ex:
+                logger.warning("Failed to decode OneSettings response content: %s", str(ex))
+            except ValueError as ex:
+                logger.warning("Failed to parse OneSettings change version: %s", str(ex))
+    elif status_code == 400:
+        logger.warning("Bad request to OneSettings: %s", response.content)
+    elif status_code == 404:
+        logger.warning("OneSettings configuration not found: %s", response.content)
+    elif status_code == 414:
+        logger.warning("OneSettings request URI too long: %s", response.content)
+    elif status_code == 500:
+        logger.warning("Internal server error from OneSettings: %s", response.content)
+
+    return OneSettingsResponse(etag, refresh_interval, settings, version)