diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md index c35d26408038..90d8e897b560 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/CHANGELOG.md @@ -14,6 +14,10 @@ - Live Metrics now honors the `APPLICATIONINSIGHTS_AUTHENTICATION_STRING` environment variable for AAD authentication as a fallback when no explicit credential is supplied and local authentication is disabled. ([#48284](https://github.com/Azure/azure-sdk-for-python/pull/48284)) +- Fix a memory leak where exporters registered as OneSettings configuration callbacks were retained for the + process lifetime; bound-method callbacks are now held via weak references so discarded exporters can be + garbage collected. + ([#48379](https://github.com/Azure/azure-sdk-for-python/pull/48379)) ### Other Changes @@ -24,6 +28,8 @@ ([#48027](https://github.com/Azure/azure-sdk-for-python/pull/48027)) - Align OneSettings feature-flag evaluation with the control-plane schema: use full-name `os`/`rp`/`attach` values, add `ikey` and `region` conditions, require exact single-value matches (removing list and version-range support), and only honor a `ver` condition when a matching `component` is also present ([#48059](https://github.com/Azure/azure-sdk-for-python/pull/48059)) +- Support remote toggling of local (offline) storage via the OneSettings `FEATURE_LOCAL_STORAGE` feature flag: the control plane can disable or re-enable disk-backed retry storage at runtime, but never overrides an explicit `disable_offline_storage=True` user opt-out. Statsbeat storage is decoupled from the user setting (always off), while customer-sdkstats honors the user setting and follows the remote toggle. + ([#48379](https://github.com/Azure/azure-sdk-for-python/pull/48379)) ## 1.0.0b55 (2026-07-01) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/__init__.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/__init__.py index 682e19bc88f3..89c9aea01846 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/__init__.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/__init__.py @@ -2,7 +2,9 @@ # Licensed under the MIT License. from dataclasses import dataclass, field from typing import Any, Dict, Optional +import inspect import logging +import weakref from threading import Lock from azure.monitor.opentelemetry.exporter._constants import ( @@ -73,7 +75,47 @@ def register_callback(self, callback): # 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) + # + # Bound methods are stored via weakref.WeakMethod so a discarded exporter (the callback's + # __self__) can be garbage collected instead of being pinned for the process lifetime by this + # singleton. This fixes the leak even when the exporter is dropped without calling shutdown(). + # Plain functions (module-level SDK callbacks) are stored as-is: they live for the process + # anyway and are not weakref-friendly. + if inspect.ismethod(callback): + stored = weakref.WeakMethod(callback) + else: + stored = callback + self._callbacks.append(stored) + + # Replay the currently cached configuration to the just-registered callback. Notifications + # fire only on a config *change*, so a callback registered after a non-default value was + # already cached would otherwise stay stale until the next change (which may never come). + # Centralizing the replay here keeps callbacks free of any feature-specific "apply cached + # state" logic and makes registration order-independent. An empty cache is a no-op. + cached_settings = self.get_settings() + if cached_settings: + self._invoke_callback(stored, cached_settings) + + def _invoke_callback(self, callback, settings: Dict[str, str]) -> bool: + # Resolve a stored callback (a plain function or a weakref.WeakMethod) and invoke it with the + # given settings, isolating any exception so one bad callback cannot break the others (or the + # exporter constructing during a registration replay). Returns True when the callback is a + # dead WeakMethod whose owner has been garbage collected, so the caller can prune it. + if isinstance(callback, weakref.WeakMethod): + # Only bound-method callbacks (e.g. the per-exporter local-storage callback) are stored as + # WeakMethod, so they must be resolved back to a live bound method here. Module-level + # function callbacks (live metrics, sdkstats) are stored directly and hit the else branch. + resolved = callback() + if resolved is None: + return True + target = resolved + else: + target = callback + try: + target(settings) + except Exception as ex: # pylint: disable=broad-except + logger.debug("Callback failed: %s", ex) + return False def _notify_callbacks(self, settings: Dict[str, str]): # Notify all registered callbacks of configuration changes. @@ -81,11 +123,15 @@ def _notify_callbacks(self, settings: Dict[str, str]): # "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: + # Invoke each callback; _invoke_callback reports back any dead weak references to prune. + dead = [cb for cb in callbacks if self._invoke_callback(cb, settings)] + # Prune dead weak references so _callbacks does not grow unbounded under exporter churn. + # list.remove is atomic under the GIL; swallow ValueError if another thread already removed it. + for cb in dead: try: - cb(settings) - except Exception as ex: # pylint: disable=broad-except - logger.debug("Callback failed: %s", ex) + self._callbacks.remove(cb) + except ValueError: + pass def _is_transient_error(self, response: OneSettingsResponse) -> bool: """Check if the response indicates a transient error. diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/_utils.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/_utils.py index db5202db54f5..02c504b848c6 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/_utils.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_configuration/_utils.py @@ -12,7 +12,6 @@ _ONE_SETTINGS_DEFAULT_REFRESH_INTERVAL_SECONDS, ) - logger = logging.getLogger(__name__) @@ -267,7 +266,11 @@ def evaluate_feature(feature_key: str, settings: Dict[str, Any]) -> Optional[boo if not isinstance(feature_config, dict): return None - default_state = feature_config.get("default", "disabled").lower() == "enabled" + # Coerce the raw value with str() before .lower(): the OneSettings payload is only JSON-decoded, + # so a malformed "default" (e.g. a JSON boolean true instead of the string "enabled") would + # otherwise raise AttributeError here. str() keeps a well-formed string unchanged while making a + # non-string safely fall through to the default-disabled state instead of crashing the caller. + default_state = str(feature_config.get("default", "disabled")).lower() == "enabled" override_list = feature_config.get("override", []) # If no override conditions, return default state diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_storage.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_storage.py index c950384a69b3..d15f5df50397 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_storage.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/_storage.py @@ -119,6 +119,10 @@ def __init__( self._max_size = max_size self._retention_period = retention_period self._write_timeout = write_timeout + # Remote (OneSettings) on/off toggle, independent of _enabled (folder permissions). + # When False, put()/gets() no-op but the instance and its maintenance thread stay alive, + # so the FEATURE_LOCAL_STORAGE kill-switch can flip storage on/off without teardown. + self._active = True self._enabled = self._check_and_set_folder_permissions() if self._enabled: self._maintenance_routine() @@ -138,6 +142,16 @@ def close(self) -> None: self._maintenance_task.cancel() self._maintenance_task.join() + def enable(self) -> None: + # Turn the remote toggle on; put()/gets() resume. No-op if folder permissions were denied. + self._active = True + + def disable(self) -> None: + # Turn the remote toggle off; put()/gets() become no-ops. The instance and maintenance + # thread are left running so storage can be re-enabled later without reconstruction. Any + # telemetry already persisted to disk is left in place for retry once re-enabled. + self._active = False + def __enter__(self) -> "LocalFileStorage": return self @@ -157,7 +171,7 @@ def _maintenance_routine(self) -> None: # pylint: disable=too-many-nested-blocks def gets(self) -> Generator[LocalFileBlob, None, None]: - if self._enabled: + if self._enabled and self._active: now = _now() lease_deadline = _fmt(now) retention_deadline = _fmt(now - _seconds(self._retention_period)) @@ -196,6 +210,7 @@ def gets(self) -> Generator[LocalFileBlob, None, None]: pass def get(self) -> Optional[LocalFileBlob]: + # gets() already gates on _enabled and _active, so no need to re-check _active here. if not self._enabled: return None cursor = self.gets() @@ -207,12 +222,15 @@ def get(self) -> Optional[LocalFileBlob]: def put(self, data: List[Any], lease_period: Optional[int] = None) -> Union[StorageExportResult, str]: try: - if not self._enabled: - if get_local_storage_setup_state_readonly(): - return StorageExportResult.CLIENT_READONLY - if get_local_storage_setup_state_exception() != "": - # Type conversion has been done to match the return type of this function - return str(get_local_storage_setup_state_exception()) + # Storage is unavailable when remotely disabled (_active) or never set up (_enabled). + if not self._active or not self._enabled: + # Only report the specific setup-failure reason when not remotely disabled. + if self._active: + if get_local_storage_setup_state_readonly(): + return StorageExportResult.CLIENT_READONLY + if get_local_storage_setup_state_exception() != "": + # Type conversion has been done to match the return type of this function + return str(get_local_storage_setup_state_exception()) return StorageExportResult.CLIENT_STORAGE_DISABLED if not self._check_storage_size(): return StorageExportResult.CLIENT_PERSISTENCE_CAPACITY_REACHED diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py index b3d7402672c2..fdcc9da12ab2 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/_base.py @@ -8,7 +8,7 @@ import sys from pathlib import Path from enum import Enum -from typing import List, Optional, Any +from typing import Dict, List, Optional, Any from urllib.parse import urlparse import psutil @@ -42,6 +42,7 @@ _REQ_THROTTLE_NAME, _RETRYABLE_STATUS_CODES, _THROTTLE_STATUS_CODES, + _ONE_SETTINGS_FEATURE_LOCAL_STORAGE, DropCode, _exception_categories, ) @@ -49,6 +50,12 @@ ConnectionStringParser, ) from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage +from azure.monitor.opentelemetry.exporter._configuration._state import ( + get_configuration_manager, +) +from azure.monitor.opentelemetry.exporter._configuration._utils import ( + evaluate_feature, +) from azure.monitor.opentelemetry.exporter._utils import ( _get_auth_policy, _get_sha256_hash, @@ -215,14 +222,24 @@ def __init__(self, **kwargs: Any) -> None: # ) self.storage: Optional[LocalFileStorage] = None if not self._disable_offline_storage: - self.storage = LocalFileStorage( # pyright: ignore - path=self._storage_directory, # type: ignore - max_size=self._storage_max_size, - maintenance_period=self._storage_maintenance_period, - retention_period=self._storage_retention_period, - name="{} Storage".format(self.__class__.__name__), - lease_period=self._storage_min_retry_interval, - ) + self._enable_local_storage() + + # Register a OneSettings callback so local (offline) storage can be toggled remotely via the + # FEATURE_LOCAL_STORAGE feature flag. The customer-sdkstats exporter participates because it + # writes to the same on-disk folder as the main exporter (keyed on the customer's ikey), so it + # must follow the same remote kill-switch; its own manager still applies the user's static + # disable_offline_storage, and the callback's hard gate never re-enables a user opt-out. The + # statsbeat exporter is skipped: it never persists to disk (isolated Microsoft-ikey folder) and + # does not participate in the remote toggle. + # register_callback is a NoOp if the control plane worker never starts, and + # get_configuration_manager() returns None when the control plane is disabled via env var. + if not self._is_stats_exporter(): + config_manager = get_configuration_manager() + if config_manager: + # register_callback also replays any already-cached configuration to this callback, + # so a late-created exporter immediately honors an existing kill-switch. The replay is + # centralized in the configuration manager; no feature-specific handling is needed here. + config_manager.register_callback(self._local_storage_configuration_callback) # statsbeat initialization if self._should_collect_stats(): @@ -253,10 +270,13 @@ def __init__(self, **kwargs: Any) -> None: _MAX_STORAGE_DRAIN_BATCH = 10 def _transmit_from_storage(self) -> None: - if not self.storage: + # self.storage is None only when the user opted out of offline storage (it is constructed + # once at init and never nulled by the remote toggle, which only flips it active/inactive). + storage = self.storage + if not storage: return drained = 0 - for blob in self.storage.gets(): + for blob in storage.gets(): if drained >= self._MAX_STORAGE_DRAIN_BATCH: break # give a few more seconds for blob lease operation @@ -280,15 +300,18 @@ def _transmit_from_storage(self) -> None: drained += 1 def _handle_transmit_from_storage(self, envelopes: List[TelemetryItem], result: ExportResult) -> None: - if self.storage: + # self.storage is None only when the user opted out of offline storage (it is constructed + # once at init and never nulled by the remote toggle, which only flips it active/inactive). + storage = self.storage + if storage: if result == ExportResult.FAILED_RETRYABLE: envelopes_to_store = [x.as_dict() for x in envelopes] if self._retry_after_delay_seconds is not None: - result_from_storage_put = self.storage.put( + result_from_storage_put = storage.put( envelopes_to_store, lease_period=self._retry_after_delay_seconds ) else: - result_from_storage_put = self.storage.put(envelopes_to_store) + result_from_storage_put = storage.put(envelopes_to_store) if self._should_collect_customer_sdkstats(): track_dropped_items_from_storage(result_from_storage_put, envelopes) self._retry_after_delay_seconds = None @@ -350,8 +373,9 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal granted + len(overflow), len(overflow), ) - if self.storage: - self.storage.put([x.as_dict() for x in overflow]) + storage = self.storage + if storage: + storage.put([x.as_dict() for x in overflow]) else: logger.warning( "Rate limiter deferred %d envelopes but offline " @@ -435,14 +459,15 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal error.message, (envelopes[error.index] if error.index is not None else ""), ) - if self.storage and resend_envelopes: + storage = self.storage + if storage and resend_envelopes: envelopes_to_store = [x.as_dict() for x in resend_envelopes] lease_period = ( retry_after_delay_seconds if retry_after_delay_seconds is not None else self._storage_min_retry_interval ) - result_from_storage = self.storage.put(envelopes_to_store, lease_period) + result_from_storage = storage.put(envelopes_to_store, lease_period) if self._should_collect_customer_sdkstats(): track_dropped_items_from_storage(result_from_storage, resend_envelopes) self._consecutive_redirects = 0 @@ -667,6 +692,55 @@ def _transmit(self, envelopes: List[TelemetryItem], _skip_rate_limit: bool = Fal self._consecutive_redirects = 0 return ExportResult.SUCCESS + # OneSettings configuration-change callback: toggles local (offline) storage. + def _local_storage_configuration_callback(self, settings: Dict[str, str]) -> None: + """Toggle local (offline) storage in response to a OneSettings configuration change. + + OneSettings acts as a remote kill-switch via the FEATURE_LOCAL_STORAGE feature flag: + it can force storage off, and re-enable it only when the user did not explicitly opt out + with ``disable_offline_storage=True``. A user opt-out is a hard gate that OneSettings can + never override. + + :param settings: Configuration settings from OneSettings. + :type settings: dict[str, str] + """ + # The user's explicit opt-out is a hard gate - never override it. + if self._disable_offline_storage: + return + local_storage_enabled = evaluate_feature(_ONE_SETTINGS_FEATURE_LOCAL_STORAGE, settings) + # None means the flag is absent/invalid; leave the current storage state unchanged. + if local_storage_enabled is None: + return + if local_storage_enabled: + self._enable_local_storage() + else: + self._disable_local_storage() + + def _enable_local_storage(self) -> None: + # Construct the local file storage once (on first enable / exporter init), then keep the + # instance for the exporter's lifetime. A remote re-enable just flips the toggle back on + # rather than recreating storage or its maintenance thread. + if self.storage is not None: + self.storage.enable() + return + if self._storage_directory is None: + self._storage_directory = _get_storage_directory(self._instrumentation_key or "") + self.storage = LocalFileStorage( # pyright: ignore + path=self._storage_directory, # type: ignore + max_size=self._storage_max_size, + maintenance_period=self._storage_maintenance_period, + retention_period=self._storage_retention_period, + name="{} Storage".format(self.__class__.__name__), + lease_period=self._storage_min_retry_interval, + ) + + def _disable_local_storage(self) -> None: + # Flip the remote toggle off so put()/gets() no-op. The storage instance and its maintenance + # thread are left running (torn down only at exporter shutdown), and any telemetry already + # persisted to disk is left in place so it can still be retried if storage is re-enabled. + if self.storage is not None: + self.storage.disable() + # check to see whether its the case of stats collection def _should_collect_stats(self): return ( diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/_manager.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/_manager.py index 3fea11c08c6c..974e0f30ff24 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/_manager.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/_manager.py @@ -47,6 +47,9 @@ def __init__( self.instrumentation_key = instrumentation_key # features + # ``disable_offline_storage`` mirrors the user's setting and is used ONLY to report the + # DISK_RETRY statsbeat feature bit (telemetry about the customer's config). Statsbeat's own + # exporter never persists to disk, independent of the user (see _do_initialize). self.disable_offline_storage = disable_offline_storage self.credential = credential self.distro_version = distro_version @@ -83,6 +86,8 @@ def from_exporter(cls, exporter: Any) -> Optional["StatsbeatConfig"]: endpoint=exporter._endpoint, region=exporter._region, instrumentation_key=exporter._instrumentation_key, + # Carry the user's setting only to report the DISK_RETRY feature bit. Statsbeat's own + # exporter never persists to disk (see _do_initialize), regardless of the user's setting. disable_offline_storage=exporter._disable_offline_storage, credential=exporter._credential, distro_version=exporter._distro_version, @@ -92,8 +97,10 @@ def from_exporter(cls, exporter: Any) -> Optional["StatsbeatConfig"]: def from_config(cls, base_config: "StatsbeatConfig", config_dict: Dict[str, str]) -> Optional["StatsbeatConfig"]: """Update configuration from a dictionary. Used in conjunction with OneSettings control plane. - Creates a new StatsbeatConfig instance with the same base configuration but updated - `connection_string` and `disable_offline_storage` from the provided dictionary. + Creates a new StatsbeatConfig instance with the same base configuration but an updated + `connection_string` from the provided dictionary. The customer's `disable_offline_storage` + setting is preserved for DISK_RETRY reporting; sdkstats's own storage is always disabled + (it never persists to disk) and is not controlled by OneSettings. :param base_config: Base configuration to update :type base_config: StatsbeatConfig @@ -118,17 +125,12 @@ def from_config(cls, base_config: "StatsbeatConfig", config_dict: Dict[str, str] # If something went wrong in fetching connection string, fall back to the original connection_string = base_config.connection_string - # TODO: Add support for disable_offline_storage from config_dict once supported in control plane - disable_offline_storage = config_dict.get("disable_offline_storage") - disable_offline_storage_config = ( - isinstance(disable_offline_storage, str) and disable_offline_storage.lower() == "true" - ) - return cls( endpoint=base_config.endpoint, region=base_config.region, instrumentation_key=base_config.instrumentation_key, - disable_offline_storage=disable_offline_storage_config, # TODO: Use config value once supported + # Preserve the customer's setting across reconfigures (used only for DISK_RETRY reporting). + disable_offline_storage=base_config.disable_offline_storage, credential=base_config.credential, distro_version=base_config.distro_version, connection_string=connection_string, @@ -243,7 +245,11 @@ def _do_initialize(self, config: StatsbeatConfig) -> bool: statsbeat_exporter = AzureMonitorMetricExporter( connection_string=config.connection_string, - disable_offline_storage=config.disable_offline_storage, + # Statsbeat never persists its own envelopes to disk. It is best-effort internal + # diagnostics, so it does not need disk-backed retry, and disabling storage avoids any + # disk writes for users who opted out. config.disable_offline_storage reflects the + # customer's config and is used only for DISK_RETRY reporting below. + disable_offline_storage=True, is_sdkstats=True, ) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_customer_sdkstats.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_customer_sdkstats.py index e8809c7a4768..2b17cab1d814 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_customer_sdkstats.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_customer_sdkstats.py @@ -22,6 +22,7 @@ def collect_customer_sdkstats(exporter: "BaseExporter") -> None: # type: ignore customer_stats.initialize( connection_string=exporter._connection_string, # type: ignore credential=exporter._credential, # type: ignore + disable_offline_storage=exporter._disable_offline_storage, # type: ignore ) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_manager.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_manager.py index 6597a109cf00..a36793fb3ce6 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_manager.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/customer/_manager.py @@ -128,13 +128,18 @@ def is_shutdown(self) -> bool: """ return self._status == CustomerSdkStatsStatus.SHUTDOWN # type: ignore - def initialize(self, connection_string: str, credential: Optional[Any] = None) -> bool: + def initialize( + self, connection_string: str, credential: Optional[Any] = None, disable_offline_storage: bool = False + ) -> bool: """Initialize Customer SDKStats collection with the provided connection string. :param connection_string: Azure Monitor connection string :type connection_string: str :param credential: Token credential for AAD authentication. Defaults to None. :type credential: ~azure.core.credentials.TokenCredential or None + :param disable_offline_storage: Whether local (offline) storage is disabled. Mirrors the + user's setting on the originating exporter, whose storage directory it shares. Defaults to False. + :type disable_offline_storage: bool :return: True if initialization was successful, False otherwise :rtype: bool @@ -150,15 +155,21 @@ def initialize(self, connection_string: str, credential: Optional[Any] = None) - # Already initialized, return True return True - return self._do_initialize(connection_string, credential=credential) + return self._do_initialize( + connection_string, credential=credential, disable_offline_storage=disable_offline_storage + ) - def _do_initialize(self, connection_string: str, credential: Optional[Any] = None) -> bool: + def _do_initialize( + self, connection_string: str, credential: Optional[Any] = None, disable_offline_storage: bool = False + ) -> bool: """Internal initialization method. :param connection_string: Azure Monitor connection string :type connection_string: str :param credential: Token credential for AAD authentication. Defaults to None. :type credential: ~azure.core.credentials.TokenCredential or None + :param disable_offline_storage: Whether local (offline) storage is disabled. Defaults to False. + :type disable_offline_storage: bool :return: True if initialization was successful, False otherwise :rtype: bool @@ -170,6 +181,7 @@ def _do_initialize(self, connection_string: str, credential: Optional[Any] = Non exporter_kwargs: Dict[str, Any] = { "connection_string": connection_string, "is_customer_sdkstats": True, + "disable_offline_storage": disable_offline_storage, } if credential is not None: exporter_kwargs["credential"] = credential @@ -211,6 +223,8 @@ def _do_initialize(self, connection_string: str, credential: Optional[Any] = Non def _cleanup(self) -> None: """Clean up resources on initialization failure.""" + # TODO: shut down _customer_sdkstats_exporter before nulling it, else a failed + # init orphans its LocalFileStorage maintenance thread. self._customer_sdkstats_exporter = None self._customer_sdkstats_metric_reader = None self._customer_sdkstats_meter_provider = None diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/configuration/test_manager.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/configuration/test_manager.py index 675ae5fd3002..c2311d2dc980 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/configuration/test_manager.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/configuration/test_manager.py @@ -180,6 +180,140 @@ def test_register_callback_after_init(self, mock_worker_class): self.assertIn(callback, manager._callbacks) + @patch("azure.monitor.opentelemetry.exporter._configuration._worker._ConfigurationWorker") + def test_register_callback_replays_cached_settings(self, mock_worker_class): + """Registering a callback immediately replays the currently cached settings to it, so a + callback registered after a value was already cached is not left stale until the next change.""" + manager = _ConfigurationManager() + manager._current_state = manager._current_state.with_updates(settings_cache={"FEATURE": "enabled"}) + callback = Mock() + + manager.register_callback(callback) + + callback.assert_called_once_with({"FEATURE": "enabled"}) + + @patch("azure.monitor.opentelemetry.exporter._configuration._worker._ConfigurationWorker") + def test_register_callback_empty_cache_no_replay(self, mock_worker_class): + """When the cache is empty (worker has not fetched yet), registration does not invoke the + callback (the replay is a no-op).""" + manager = _ConfigurationManager() + callback = Mock() + + manager.register_callback(callback) + + callback.assert_not_called() + + @patch("azure.monitor.opentelemetry.exporter._configuration._worker._ConfigurationWorker") + def test_register_callback_replay_isolates_exception(self, mock_worker_class): + """A callback that raises during the registration replay is isolated: registration still + succeeds and the callback is stored (matching notification-time exception isolation).""" + manager = _ConfigurationManager() + manager._current_state = manager._current_state.with_updates(settings_cache={"FEATURE": "enabled"}) + callback = Mock(side_effect=ValueError("boom")) + + manager.register_callback(callback) + + callback.assert_called_once_with({"FEATURE": "enabled"}) + self.assertIn(callback, manager._callbacks) + + @patch("azure.monitor.opentelemetry.exporter._configuration._worker._ConfigurationWorker") + def test_register_callback_replays_to_bound_method(self, mock_worker_class): + """The replay resolves a WeakMethod-stored bound method and invokes it with cached settings.""" + sink = [] + + class _Holder: + def __init__(self, sink): + self._sink = sink + + def on_settings(self, settings): + self._sink.append(settings) + + manager = _ConfigurationManager() + manager._current_state = manager._current_state.with_updates(settings_cache={"FEATURE": "enabled"}) + holder = _Holder(sink) + + manager.register_callback(holder.on_settings) + + self.assertEqual(sink, [{"FEATURE": "enabled"}]) + + def test_bound_method_callback_stored_weakly(self): + """A bound-method callback is wrapped in weakref.WeakMethod so its owner is not pinned.""" + import weakref + + class _Holder: + def on_settings(self, settings): + pass + + holder = _Holder() + manager = _ConfigurationManager() + manager.register_callback(holder.on_settings) + + self.assertEqual(len(manager._callbacks), 1) + self.assertIsInstance(manager._callbacks[0], weakref.WeakMethod) + + def test_function_callback_stored_directly(self): + """A plain function callback is stored as-is (not weakly): it lives for the process anyway.""" + import weakref + + def on_settings(settings): + pass + + manager = _ConfigurationManager() + manager.register_callback(on_settings) + + self.assertEqual(len(manager._callbacks), 1) + self.assertNotIsInstance(manager._callbacks[0], weakref.WeakMethod) + self.assertIs(manager._callbacks[0], on_settings) + + def test_dead_bound_method_not_notified_and_pruned(self): + """When a bound method's owner is garbage collected, _notify_callbacks skips it and prunes + the dead weak reference, so a discarded exporter is not retained or re-invoked.""" + import gc + + sink = [] + + class _Holder: + def __init__(self, sink): + self._sink = sink + + def on_settings(self, settings): + self._sink.append(settings) + + manager = _ConfigurationManager() + holder = _Holder(sink) + manager.register_callback(holder.on_settings) + self.assertEqual(len(manager._callbacks), 1) + + # Drop the only strong reference to the owner and force collection. + del holder + gc.collect() + + manager._notify_callbacks({"FEATURE": "x"}) + + # The dead callback was neither invoked nor left behind. + self.assertEqual(sink, []) + self.assertEqual(manager._callbacks, []) + + def test_live_bound_method_still_notified(self): + """A bound method whose owner is still alive is resolved and invoked normally.""" + sink = [] + + class _Holder: + def __init__(self, sink): + self._sink = sink + + def on_settings(self, settings): + self._sink.append(settings) + + manager = _ConfigurationManager() + holder = _Holder(sink) + manager.register_callback(holder.on_settings) + + manager._notify_callbacks({"FEATURE": "x"}) + + self.assertEqual(sink, [{"FEATURE": "x"}]) + self.assertEqual(len(manager._callbacks), 1) + @patch("azure.monitor.opentelemetry.exporter._configuration._worker._ConfigurationWorker") def test_worker_initialization(self, mock_worker_class): """Test that ConfigurationWorker is initialized properly.""" diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_customer_sdkstats.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_customer_sdkstats.py index 83889032c43e..3588faf0248c 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_customer_sdkstats.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_customer_sdkstats.py @@ -62,12 +62,14 @@ def test_collect_customer_sdkstats_passes_credential(self, mock_get_manager): mock_exporter = mock.Mock() mock_exporter._connection_string = "InstrumentationKey=12345678-1234-5678-abcd-12345678abcd" mock_exporter._credential = mock.Mock() + mock_exporter._disable_offline_storage = False collect_customer_sdkstats(mock_exporter) mock_manager.initialize.assert_called_once_with( connection_string=mock_exporter._connection_string, credential=mock_exporter._credential, + disable_offline_storage=mock_exporter._disable_offline_storage, ) def test_collect_customer_sdkstats_multiple_calls(self): diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_manager.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_manager.py index 24fede682c32..d0a90630cc81 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_manager.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/customer_sdk_stats/test_manager.py @@ -106,6 +106,7 @@ def test_initialize_success(self, mock_meter_provider, mock_metric_reader, mock_ connection_string=connection_string, is_customer_sdkstats=True, credential=self.mock_credential, + disable_offline_storage=False, ) mock_metric_reader.assert_called_once() mock_meter_provider.assert_called_once() @@ -157,6 +158,7 @@ def test_initialize_with_credential( # pylint: disable=unused-argument connection_string=connection_string, is_customer_sdkstats=True, credential=mock_credential, + disable_offline_storage=False, ) @patch("azure.monitor.opentelemetry.exporter.export.metrics._exporter.AzureMonitorMetricExporter") @@ -179,6 +181,30 @@ def test_initialize_without_credential( # pylint: disable=unused-argument mock_exporter.assert_called_once_with( connection_string=connection_string, is_customer_sdkstats=True, + disable_offline_storage=False, + ) + + @patch("azure.monitor.opentelemetry.exporter.export.metrics._exporter.AzureMonitorMetricExporter") + @patch("azure.monitor.opentelemetry.exporter.statsbeat.customer._manager.PeriodicExportingMetricReader") + @patch("azure.monitor.opentelemetry.exporter.statsbeat.customer._manager.MeterProvider") + def test_initialize_honors_disable_offline_storage( # pylint: disable=unused-argument + self, mock_meter_provider, mock_metric_reader, mock_exporter + ): + """Customer sdkstats shares the user's storage directory, so it honors the user's opt-out.""" + mock_meter = Mock() + mock_meter_provider_instance = Mock() + mock_meter_provider_instance.get_meter.return_value = mock_meter + mock_meter_provider.return_value = mock_meter_provider_instance + + connection_string = "InstrumentationKey=12345678-1234-5678-abcd-12345678abcd" + + result = self.manager.initialize(connection_string, disable_offline_storage=True) + + self.assertTrue(result) + mock_exporter.assert_called_once_with( + connection_string=connection_string, + is_customer_sdkstats=True, + disable_offline_storage=True, ) def test_initialize_multiple_calls(self): diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/statsbeat/test_manager.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/statsbeat/test_manager.py index 9baef3379c32..e3c03938fc64 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/statsbeat/test_manager.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/statsbeat/test_manager.py @@ -96,6 +96,8 @@ def test_from_exporter_valid(self): self.assertEqual(config.endpoint, "https://westus-1.in.applicationinsights.azure.com/") self.assertEqual(config.region, "westus") self.assertEqual(config.instrumentation_key, "test-key") + # config.disable_offline_storage mirrors the user's setting (used only for DISK_RETRY + # reporting); statsbeat's own exporter never persists to disk regardless (see _do_initialize). self.assertTrue(config.disable_offline_storage) self.assertIsNotNone(config.credential) self.assertEqual(config.distro_version, "1.0.0") @@ -140,11 +142,13 @@ def test_from_config_valid(self, mock_get_cs_for_region): region="westus", instrumentation_key="test-key", credential="test_credential", - disable_offline_storage=False, + disable_offline_storage=True, distro_version="1.0.0", ) - config_dict = {"disable_offline_storage": "true"} + # A conflicting value in the dict must be ignored: the customer's setting is preserved from + # base_config, and statsbeat's own storage is never controlled by OneSettings. + config_dict = {"disable_offline_storage": "false"} new_config = StatsbeatConfig.from_config(base_config, config_dict) diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py index 1f2499e1d4be..943b13251182 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_base_exporter.py @@ -311,6 +311,194 @@ def test_constructor_disable_offline_storage_with_storage_directory(self, mock_g self.assertEqual(base._storage_directory, "test/path") mock_get_temp_dir.assert_not_called() + # ======================================================================== + # ONESETTINGS LOCAL STORAGE TOGGLE TESTS + # ======================================================================== + + def _make_local_storage_settings(self, enabled): + state = "enabled" if enabled else "disabled" + return {"FEATURE_LOCAL_STORAGE": {"default": state}} + + def test_configuration_callback_disables_storage(self): + """OneSettings FEATURE_LOCAL_STORAGE=disabled turns off active local storage (put/gets no-op).""" + base = BaseExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + ) + self.assertIsNotNone(base.storage) + try: + base._local_storage_configuration_callback(self._make_local_storage_settings(False)) + # Flag model: the instance stays alive but is toggled off. + self.assertIsNotNone(base.storage) + self.assertFalse(base.storage._active) + finally: + if base.storage is not None: + clean_folder(base.storage._path) + + def test_configuration_callback_reenables_storage(self): + """After a disable, FEATURE_LOCAL_STORAGE=enabled toggles the same storage instance back on.""" + base = BaseExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + ) + base._local_storage_configuration_callback(self._make_local_storage_settings(False)) + self.assertIsNotNone(base.storage) + self.assertFalse(base.storage._active) + storage_instance = base.storage + try: + base._local_storage_configuration_callback(self._make_local_storage_settings(True)) + # Same instance is reused (not reconstructed) and toggled back on. + self.assertIs(base.storage, storage_instance) + self.assertTrue(base.storage._active) + finally: + if base.storage is not None: + clean_folder(base.storage._path) + + def test_configuration_callback_respects_user_optout(self): + """A user's explicit disable_offline_storage=True is a hard gate OneSettings cannot override.""" + base = BaseExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=True, + ) + self.assertIsNone(base.storage) + base._local_storage_configuration_callback(self._make_local_storage_settings(True)) + self.assertIsNone(base.storage) + + def test_configuration_callback_flag_absent_no_change(self): + """When FEATURE_LOCAL_STORAGE is absent, storage state is left unchanged.""" + base = BaseExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + ) + try: + base._local_storage_configuration_callback({}) + self.assertIsNotNone(base.storage) + finally: + if base.storage is not None: + clean_folder(base.storage._path) + + @mock.patch("azure.monitor.opentelemetry.exporter.export._base.get_configuration_manager") + def test_constructor_registers_local_storage_callback(self, mock_get_config_manager): + """A normal exporter registers its local storage callback with the config manager.""" + mock_manager = mock.Mock() + mock_manager.get_settings.return_value = {} + mock_get_config_manager.return_value = mock_manager + base = BaseExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + ) + try: + mock_manager.register_callback.assert_called_once_with(base._local_storage_configuration_callback) + finally: + if base.storage is not None: + clean_folder(base.storage._path) + + @mock.patch("azure.monitor.opentelemetry.exporter.export._base.get_configuration_manager") + def test_constructor_no_callback_when_control_plane_disabled(self, mock_get_config_manager): + """When the control plane is disabled (manager is None), no callback is registered.""" + mock_get_config_manager.return_value = None + base = BaseExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + ) + try: + # No exception means the None manager was handled gracefully. + self.assertIsNotNone(base) + finally: + if base.storage is not None: + clean_folder(base.storage._path) + + @mock.patch("azure.monitor.opentelemetry.exporter.export._base.get_configuration_manager") + def test_stats_exporter_does_not_register_local_storage_callback(self, mock_get_config_manager): + """The statsbeat exporter manages its own storage and must not register the base callback.""" + mock_manager = mock.Mock() + mock_get_config_manager.return_value = mock_manager + base = AzureMonitorMetricExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + is_sdkstats=True, + ) + try: + mock_manager.register_callback.assert_not_called() + finally: + if base.storage is not None: + clean_folder(base.storage._path) + + @mock.patch("azure.monitor.opentelemetry.exporter.export._base.get_configuration_manager") + def test_customer_sdkstats_exporter_registers_local_storage_callback(self, mock_get_config_manager): + """The customer-sdkstats exporter shares the customer's storage folder, so it participates + in the remote FEATURE_LOCAL_STORAGE toggle and must register the base callback.""" + mock_manager = mock.Mock() + mock_manager.get_settings.return_value = {} + mock_get_config_manager.return_value = mock_manager + base = AzureMonitorMetricExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + is_customer_sdkstats=True, + ) + try: + mock_manager.register_callback.assert_called_once_with(base._local_storage_configuration_callback) + finally: + if base.storage is not None: + clean_folder(base.storage._path) + + def test_late_exporter_applies_cached_disabled_settings(self): + """An exporter created after FEATURE_LOCAL_STORAGE=disabled was cached applies that cached + state at registration time (via the manager's centralized replay, no config change needed) + and starts with storage toggled off.""" + from azure.monitor.opentelemetry.exporter._configuration import _ConfigurationManager + from azure.monitor.opentelemetry.exporter._utils import Singleton + + Singleton._instances.pop(_ConfigurationManager, None) + manager = _ConfigurationManager() + manager._current_state = manager._current_state.with_updates( + settings_cache=self._make_local_storage_settings(False) + ) + try: + with mock.patch( + "azure.monitor.opentelemetry.exporter.export._base.get_configuration_manager", + return_value=manager, + ): + base = BaseExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + ) + try: + # Flag model: instance exists but is toggled off per the cached kill-switch. + self.assertIsNotNone(base.storage) + self.assertFalse(base.storage._active) + finally: + if base.storage is not None: + clean_folder(base.storage._path) + finally: + Singleton._instances.pop(_ConfigurationManager, None) + + def test_late_exporter_empty_cache_leaves_storage_active(self): + """When the cached settings are empty (worker has not fetched yet), the replay is a no-op and + storage remains active per the user's disable_offline_storage setting.""" + from azure.monitor.opentelemetry.exporter._configuration import _ConfigurationManager + from azure.monitor.opentelemetry.exporter._utils import Singleton + + Singleton._instances.pop(_ConfigurationManager, None) + manager = _ConfigurationManager() + try: + with mock.patch( + "azure.monitor.opentelemetry.exporter.export._base.get_configuration_manager", + return_value=manager, + ): + base = BaseExporter( + connection_string="InstrumentationKey=4321abcd-5678-4efa-8abc-1234567890ab;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/", + disable_offline_storage=False, + ) + try: + self.assertIsNotNone(base.storage) + self.assertTrue(base.storage._active) + finally: + if base.storage is not None: + clean_folder(base.storage._path) + finally: + Singleton._instances.pop(_ConfigurationManager, None) + def test_normal_exporter_includes_http_logging_policy(self): from azure.core.pipeline.policies import HttpLoggingPolicy diff --git a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_storage.py b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_storage.py index 025dfe02c919..0a076ae3eb6f 100644 --- a/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_storage.py +++ b/sdk/monitor/azure-monitor-opentelemetry-exporter/tests/test_storage.py @@ -347,6 +347,27 @@ def test_get(self): self.assertIsNone(stor.get()) self.assertIsNone(stor.get()) + def test_toggle_disable_makes_put_and_gets_noop(self): + """disable() turns put()/gets() into no-ops without tearing down the instance.""" + with LocalFileStorage(os.path.join(TEST_FOLDER, "toggle")) as stor: + self.assertTrue(stor._active) + stor.disable() + self.assertFalse(stor._active) + result = stor.put((1, 2, 3)) + self.assertEqual(result, StorageExportResult.CLIENT_STORAGE_DISABLED) + self.assertIsNone(stor.get()) + self.assertEqual(list(stor.gets()), []) + + def test_toggle_reenable_resumes_put_and_gets(self): + """enable() after disable() resumes persistence on the same instance and drains prior blobs.""" + with LocalFileStorage(os.path.join(TEST_FOLDER, "toggle2")) as stor: + stor.disable() + self.assertEqual(stor.put((1, 2, 3)), StorageExportResult.CLIENT_STORAGE_DISABLED) + stor.enable() + self.assertTrue(stor._active) + self.assertEqual(stor.put((1, 2, 3), 0), StorageExportResult.LOCAL_FILE_BLOB_SUCCESS) + self.assertEqual(stor.get().get(), (1, 2, 3)) + def test_put(self): test_input = (1, 2, 3) with LocalFileStorage(os.path.join(TEST_FOLDER, "bar")) as stor: