Read from global state and add a 15s warmup delay

Author: rads-1996

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

@@ -264,8 +264,8 @@ def _check_and_set_folder_permissions(self) -> bool:
             # Unix
             else:
                 open_flags = (
-                    os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW  # cspell:disable-line
-                )  # pylint: disable=no-member
+                    os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW  # pylint: disable=no-member  # cspell:disable-line
+                )
                 dir_fd = os.open(self._path, open_flags)
                 try:
                     dir_stat = os.fstat(dir_fd)

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

@@ -24,6 +24,8 @@
 
 logger = logging.getLogger(__name__)
 
+_STATSBEAT_INITIAL_EXPORT_WARMUP_SECONDS = 15  # 15 second warmup delay
+
 
 class StatsbeatConfig:
     """Configuration class for Statsbeat metrics collection."""
@@ -241,8 +243,8 @@ def _do_initialize(self, config: StatsbeatConfig) -> bool:
                 config.distro_version,
             )
 
-            # Force initial flush and initialize non-initial metrics
-            self._meter_provider.force_flush()
+            # Schedule initial statsbeat flush after warmup delay to allow feature bits to settle.
+            self._schedule_initial_export_flush()
             self._metrics.init_non_initial_metrics()
 
             self._config = config
@@ -258,6 +260,22 @@ def _do_initialize(self, config: StatsbeatConfig) -> bool:
             self._cleanup()
             return False
 
+    def _schedule_initial_export_flush(self) -> None:
+        def _flush() -> None:
+            meter_provider = self._meter_provider
+            if not self._initialized or meter_provider is None:
+                return
+            try:
+                meter_provider.force_flush()
+            except Exception as e:  # pylint: disable=broad-except
+                logger.warning(  # pylint: disable=do-not-log-exceptions-if-not-debug
+                    "Failed to force flush statsbeat after warmup: %s", e
+                )
+
+        timer = threading.Timer(_STATSBEAT_INITIAL_EXPORT_WARMUP_SECONDS, _flush)
+        timer.daemon = True
+        timer.start()
+
     def _cleanup(self, shutdown_meter_provider: bool = True) -> None:
         # Clean up resources with optional meter provider shutdown
         if shutdown_meter_provider and self._meter_provider:

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/_statsbeat_metrics.py

@@ -133,7 +133,7 @@ def __init__(
             _StatsbeatMetrics._COMMON_ATTRIBUTES["version"] = _get_version()
 
         self._ikey = instrumentation_key
-        self._feature = _StatsbeatFeature.NONE
+        self._feature = _StatsbeatMetrics._FEATURE_ATTRIBUTES["feature"] or _StatsbeatFeature.NONE
         if not disable_offline_storage:
             self._feature |= _StatsbeatFeature.DISK_RETRY
         if has_credential:
@@ -264,6 +264,10 @@ def _get_feature_metric(self, options: CallbackOptions) -> Iterable[Observation]
             return observations
         # Feature metric
         # Check if any features were enabled during runtime
+        feature_bits = int(_StatsbeatMetrics._FEATURE_ATTRIBUTES.get("feature") or 0)
+        if feature_bits:
+            self._feature |= feature_bits
+            _StatsbeatMetrics._FEATURE_ATTRIBUTES["feature"] = self._feature
         if get_statsbeat_custom_events_feature_set():
             self._feature |= _StatsbeatFeature.CUSTOM_EVENTS_EXTENSION
             _StatsbeatMetrics._FEATURE_ATTRIBUTES["feature"] = self._feature

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/statsbeat/test_manager.py

@@ -366,9 +366,16 @@ def test_initialize_invalid_config(self, mock_is_enabled):
     @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.PeriodicExportingMetricReader")
     @patch("azure.monitor.opentelemetry.exporter.export.metrics._exporter.AzureMonitorMetricExporter")
     @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager._StatsbeatMetrics")
+    @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.threading.Timer")
     @patch("azure.monitor.opentelemetry.exporter.statsbeat._state.is_statsbeat_enabled")
     def test_initialize_success(
-        self, mock_is_enabled, mock_statsbeat_metrics, mock_exporter_class, mock_reader_class, mock_meter_provider_class
+        self,
+        mock_is_enabled,
+        mock_timer_class,
+        mock_statsbeat_metrics,
+        mock_exporter_class,
+        mock_reader_class,
+        mock_meter_provider_class,
     ):
         """Test successful initialization."""
         mock_is_enabled.return_value = True
@@ -388,6 +395,7 @@ def test_initialize_success(
         # Mock the statsbeat metrics
         mock_metrics = Mock()
         mock_statsbeat_metrics.return_value = mock_metrics
+        mock_timer_class.return_value = Mock()
 
         config = self._create_valid_config()
 
@@ -404,9 +412,47 @@ def test_initialize_success(
         mock_reader_class.assert_called_once()
         mock_meter_provider_class.assert_called_once()
         mock_statsbeat_metrics.assert_called_once()
-        mock_meter_provider.force_flush.assert_called_once()
+        mock_timer_class.assert_called_once()
+        mock_timer_class.return_value.start.assert_called_once()
+        mock_meter_provider.force_flush.assert_not_called()
         mock_metrics.init_non_initial_metrics.assert_called_once()
 
+    @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.threading.Timer")
+    @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.MeterProvider")
+    @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.PeriodicExportingMetricReader")
+    @patch("azure.monitor.opentelemetry.exporter.export.metrics._exporter.AzureMonitorMetricExporter")
+    @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager._StatsbeatMetrics")
+    @patch("azure.monitor.opentelemetry.exporter.statsbeat._state.is_statsbeat_enabled")
+    def test_initialize_with_delay_schedules_non_blocking_flush(
+        self,
+        mock_is_enabled,
+        mock_statsbeat_metrics,
+        mock_exporter_class,
+        mock_reader_class,
+        mock_meter_provider_class,
+        mock_timer_class,
+    ):
+        """Test delayed initial statsbeat flush is scheduled asynchronously."""
+        mock_is_enabled.return_value = True
+
+        mock_exporter_class.return_value = Mock()
+        mock_reader_class.return_value = Mock()
+        mock_meter_provider = Mock()
+        mock_meter_provider_class.return_value = mock_meter_provider
+        mock_metrics = Mock()
+        mock_statsbeat_metrics.return_value = mock_metrics
+
+        mock_timer = Mock()
+        mock_timer_class.return_value = mock_timer
+
+        result = self.manager.initialize(self._create_valid_config())
+
+        self.assertTrue(result)
+        mock_timer_class.assert_called_once()
+        self.assertEqual(mock_timer_class.call_args[0][0], 15)
+        mock_timer.start.assert_called_once()
+        mock_meter_provider.force_flush.assert_not_called()
+
     @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.MeterProvider")
     @patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.PeriodicExportingMetricReader")
     @patch("azure.monitor.opentelemetry.exporter.export.metrics._exporter.AzureMonitorMetricExporter")

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/statsbeat/test_metrics.py

@@ -716,6 +716,7 @@ def test_get_feature_metric_instrumentation(self):
             False,
         )
         metric._feature = _StatsbeatFeature.NONE
+        _StatsbeatMetrics._FEATURE_ATTRIBUTES["feature"] = _StatsbeatFeature.NONE
         attributes = dict(_StatsbeatMetrics._COMMON_ATTRIBUTES)
         attributes.update(_StatsbeatMetrics._INSTRUMENTATION_ATTRIBUTES)
         self.assertEqual(attributes["type"], _FEATURE_TYPES.INSTRUMENTATION)
@@ -747,6 +748,7 @@ def test_get_feature_metric_instrumentation_none(self):
             False,
         )
         metric._feature = _StatsbeatFeature.NONE
+        _StatsbeatMetrics._FEATURE_ATTRIBUTES["feature"] = _StatsbeatFeature.NONE
         self.assertEqual(_StatsbeatMetrics._INSTRUMENTATION_ATTRIBUTES["feature"], 0)
         with mock.patch("azure.monitor.opentelemetry.exporter._utils.get_instrumentations") as instrumentations:
             instrumentations.return_value = 0

sdk/monitor/azure-monitor-opentelemetry-exporter/tests/statsbeat/test_statsbeat.py

@@ -10,7 +10,7 @@
 )
 from azure.monitor.opentelemetry.exporter.statsbeat import StatsbeatConfig, _statsbeat
 from azure.monitor.opentelemetry.exporter.statsbeat._manager import StatsbeatManager
-from azure.monitor.opentelemetry.exporter.statsbeat._statsbeat_metrics import _StatsbeatFeature
+from azure.monitor.opentelemetry.exporter.statsbeat._statsbeat_metrics import _StatsbeatFeature, _StatsbeatMetrics
 from azure.monitor.opentelemetry.exporter.statsbeat._state import (
     _STATSBEAT_STATE,
     _STATSBEAT_STATE_LOCK,
@@ -24,6 +24,8 @@
 
 # cSpell:disable
 
+_StatsbeatMetrics_FEATURE_ATTRIBUTES = dict(_StatsbeatMetrics._FEATURE_ATTRIBUTES)
+
 
 # pylint: disable=protected-access, unused-argument
 class TestStatsbeat(unittest.TestCase):
@@ -41,6 +43,7 @@ def setUp(self):
             _STATSBEAT_STATE["CUSTOM_EVENTS_FEATURE_SET"] = False
             _STATSBEAT_STATE["LIVE_METRICS_FEATURE_SET"] = False
             _STATSBEAT_STATE["CUSTOMER_SDKSTATS_FEATURE_SET"] = False
+        _StatsbeatMetrics._FEATURE_ATTRIBUTES = dict(_StatsbeatMetrics_FEATURE_ATTRIBUTES)
 
     def tearDown(self):
         """Clean up after tests."""
@@ -56,7 +59,10 @@ def tearDown(self):
     @mock.patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.MeterProvider")
     @mock.patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.PeriodicExportingMetricReader")
     @mock.patch("azure.monitor.opentelemetry.exporter.AzureMonitorMetricExporter")
-    def test_collect_statsbeat_metrics(self, mock_exporter, mock_reader, mock_meter_provider, mock_statsbeat_metrics):
+    @mock.patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.threading.Timer")
+    def test_collect_statsbeat_metrics(
+        self, mock_timer_class, mock_exporter, mock_reader, mock_meter_provider, mock_statsbeat_metrics
+    ):
         """Test that collect_statsbeat_metrics properly initializes statsbeat collection."""
         # Arrange
         exporter = mock.Mock()
@@ -78,6 +84,9 @@ def test_collect_statsbeat_metrics(self, mock_exporter, mock_reader, mock_meter_
         flush_mock = mock.Mock()
         mock_meter_provider_instance.force_flush = flush_mock
 
+        mock_timer = mock.Mock()
+        mock_timer_class.return_value = mock_timer
+
         mock_statsbeat_metrics_instance = mock.Mock()
         mock_statsbeat_metrics.return_value = mock_statsbeat_metrics_instance
 
@@ -87,6 +96,10 @@ def test_collect_statsbeat_metrics(self, mock_exporter, mock_reader, mock_meter_
         # Act
         _statsbeat.collect_statsbeat_metrics(exporter)
 
+        # Simulate warmup timer callback execution.
+        timer_callback = mock_timer_class.call_args[0][1]
+        timer_callback()
+
         # Assert - verify manager is initialized
         self.assertTrue(manager._initialized)
         self.assertEqual(manager._metrics, mock_statsbeat_metrics_instance)
@@ -329,8 +342,15 @@ def test_collect_statsbeat_metrics_no_callback_when_init_fails(
     @mock.patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.MeterProvider")
     @mock.patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.PeriodicExportingMetricReader")
     @mock.patch("azure.monitor.opentelemetry.exporter.export.metrics._exporter.AzureMonitorMetricExporter")
+    @mock.patch("azure.monitor.opentelemetry.exporter.statsbeat._manager.threading.Timer")
     def test_collect_statsbeat_metrics_exists(
-        self, mock_exporter, mock_reader, mock_meter_provider, mock_statsbeat_metrics, mock_get_manager
+        self,
+        mock_timer_class,
+        mock_exporter,
+        mock_reader,
+        mock_meter_provider,
+        mock_statsbeat_metrics,
+        mock_get_manager,
     ):
         """Test that collect_statsbeat_metrics reuses existing configuration when called multiple times with same config."""  # pylint: disable=line-too-long
         # Arrange
@@ -353,6 +373,9 @@ def test_collect_statsbeat_metrics_exists(
         flush_mock = mock.Mock()
         mock_meter_provider_instance.force_flush = flush_mock
 
+        mock_timer = mock.Mock()
+        mock_timer_class.return_value = mock_timer
+
         mock_statsbeat_metrics_instance = mock.Mock()
         mock_statsbeat_metrics.return_value = mock_statsbeat_metrics_instance
 
@@ -362,6 +385,8 @@ def test_collect_statsbeat_metrics_exists(
 
         # Act - Initialize first time
         _statsbeat.collect_statsbeat_metrics(exporter)
+        timer_callback = mock_timer_class.call_args[0][1]
+        timer_callback()
         first_metrics = manager._metrics
         self.assertTrue(manager._initialized)
         self.assertEqual(first_metrics, mock_statsbeat_metrics_instance)