Onesettings local storage toggle

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 <!-- cspell:ignore ikey -->
   ([#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)
 

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,19 +75,63 @@ 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.
         # Snapshot the list first so a concurrent register_callback on another thread can't trigger
         # "list changed size during iteration". list.append and list(...) are individually atomic
         # under the GIL, so no lock is needed here (and _state_lock stays scoped to config state).
         callbacks = list(self._callbacks)
-        for cb in callbacks:
+        # 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.

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

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

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,13 +42,20 @@
     _REQ_THROTTLE_NAME,
     _RETRYABLE_STATUS_CODES,
     _THROTTLE_STATUS_CODES,
+    _ONE_SETTINGS_FEATURE_LOCAL_STORAGE,
     DropCode,
     _exception_categories,
 )
 from azure.monitor.opentelemetry.exporter._connection_string_parser import (
     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 (