Customer Facing Statsbeat: Updated base and storage to include the metrices for customer statsbeat (#42123)

* Updated base and storage to include the metrices for customer statsbeat

* Updated readonly dropped items and CHANGELOG

* Update sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/statsbeat/_customer_statsbeat.py

Co-authored-by: Copilot <[email protected]>

* Fixed linting errors

* Updated base and exporter with retry and dropped items tracking

* Resolved spell check errors

* Resolved cspell errors

* Removed changes from storage

* Updated CHANGELOG

* Added ending line to test_storage

* Addressed feedback and modifed track_dropped_items

---------

Co-authored-by: Copilot <[email protected]>

Author: rads-1996

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

@@ -38,6 +38,9 @@
   ([#41971](https://github.com/Azure/azure-sdk-for-python/pull/41971))
 - Support AI Foundry by Handling GEN_AI_SYSTEM Attributes with [Spec](https://github.com/aep-health-and-standards/Telemetry-Collection-Spec/blob/main/ApplicationInsights/genai_semconv_mapping.md) ([#41705](https://github.com/Azure/azure-sdk-for-python/pull/41705))
 - Distinguish Azure AI SDKs in statsbeats ([#42016](https://github.com/Azure/azure-sdk-for-python/pull/42016))
+- Customer Facing Statsbeat: Updated base to include the metrics for customer statsbeat
+  ([#42123](https://github.com/Azure/azure-sdk-for-python/pull/42123))
+
 
 ## 1.0.0b39 (2025-06-25)
 

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

@@ -144,8 +144,8 @@
 class DropCode(str, Enum, metaclass=CaseInsensitiveEnumMeta):
     CLIENT_READONLY = "CLIENT_READONLY"
     CLIENT_EXCEPTION = "CLIENT_EXCEPTION"
-    CLIENT_STALE_DATA = "CLIENT_STALE_DATA"
     CLIENT_PERSISTENCE_CAPACITY = "CLIENT_PERSISTENCE_CAPACITY"
+    CLIENT_STORAGE_DISABLED = "CLIENT_STORAGE_DISABLED"
     UNKNOWN = "UNKNOWN"
 
 DropCodeType = Union[DropCode, int]

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

@@ -42,6 +42,7 @@
     _REQ_THROTTLE_NAME,
     _RETRYABLE_STATUS_CODES,
     _THROTTLE_STATUS_CODES,
+    DropCode,
 )
 from azure.monitor.opentelemetry.exporter._connection_string_parser import ConnectionStringParser
 from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage
@@ -53,7 +54,13 @@
     is_statsbeat_enabled,
     set_statsbeat_initial_success,
 )
-from azure.monitor.opentelemetry.exporter.statsbeat._utils import _update_requests_map
+from azure.monitor.opentelemetry.exporter.statsbeat._utils import (
+    _update_requests_map,
+    _track_dropped_items,
+    _track_retry_items,
+    _track_successful_items,
+)
+
 
 logger = logging.getLogger(__name__)
 
@@ -162,6 +169,14 @@ def __init__(self, **kwargs: Any) -> None:
 
             collect_statsbeat_metrics(self)
 
+        # Initialize customer statsbeat if enabled
+        self._customer_statsbeat_metrics = None
+        if self._should_collect_customer_statsbeat():
+
+            from azure.monitor.opentelemetry.exporter.statsbeat._customer_statsbeat import collect_customer_statsbeat
+            # Collect customer statsbeat metrics
+            collect_customer_statsbeat(self)
+
     def _transmit_from_storage(self) -> None:
         if not self.storage:
             return
@@ -184,6 +199,10 @@ def _handle_transmit_from_storage(self, envelopes: List[TelemetryItem], result:
             elif result == ExportResult.SUCCESS:
                 # Try to send any cached events
                 self._transmit_from_storage()
+        else:
+            # Track items that would have been retried but are dropped since client has local storage disabled
+            if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                _track_dropped_items(self._customer_statsbeat_metrics, envelopes, DropCode.CLIENT_STORAGE_DISABLED)
 
     # pylint: disable=too-many-branches
     # pylint: disable=too-many-nested-blocks
@@ -219,14 +238,25 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                         _update_requests_map(_REQ_SUCCESS_NAME[1], 1)
                     reach_ingestion = True
                     result = ExportResult.SUCCESS
+
+                    # Track successful items in customer statsbeat
+                    if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                        _track_successful_items(self._customer_statsbeat_metrics, envelopes)
                 else:  # 206
                     reach_ingestion = True
                     resend_envelopes = []
                     for error in track_response.errors:
                         if _is_retryable_code(error.status_code):
                             resend_envelopes.append(envelopes[error.index])  # type: ignore
+                            # Track retried items in customer statsbeat
+                            if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                                _track_retry_items(self._customer_statsbeat_metrics, resend_envelopes, error)
                         else:
                             if not self._is_stats_exporter():
+                                # Track dropped items in customer statsbeat, non-retryable scenario
+                                if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                                    if error is not None and hasattr(error, "index") and error.index is not None:
+                                        _track_dropped_items(self._customer_statsbeat_metrics, [envelopes[error.index]], error.status_code)
                                 logger.error(
                                     "Data drop %s: %s %s.",
                                     error.status_code,
@@ -237,6 +267,10 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                         envelopes_to_store = [x.as_dict() for x in resend_envelopes]
                         self.storage.put(envelopes_to_store, 0)
                         self._consecutive_redirects = 0
+                    elif resend_envelopes:
+                        # Track items that would have been retried but are dropped since client has local storage disabled
+                        if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                            _track_dropped_items(self._customer_statsbeat_metrics, resend_envelopes, DropCode.CLIENT_STORAGE_DISABLED)
                     # Mark as not retryable because we already write to storage here
                     result = ExportResult.FAILED_NOT_RETRYABLE
             except HttpResponseError as response_error:
@@ -249,6 +283,8 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                     result = ExportResult.FAILED_RETRYABLE
                     # Log error for 401: Unauthorized, 403: Forbidden to assist with customer troubleshooting
                     if not self._is_stats_exporter():
+                        if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                            _track_retry_items(self._customer_statsbeat_metrics, envelopes, response_error)
                         if response_error.status_code == 401:
                             logger.error(
                                 "Retryable server side error: %s. " \
@@ -257,6 +293,8 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                                 response_error.message,
                             )
                         elif response_error.status_code == 403:
+                            if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                                _track_retry_items(self._customer_statsbeat_metrics, envelopes, response_error)
                             logger.error(
                                 "Retryable server side error: %s. " \
                                 "Your application may be configured with a token credential " \
@@ -269,6 +307,10 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                     if self._should_collect_stats():
                         _update_requests_map(_REQ_THROTTLE_NAME[1], value=response_error.status_code)
                     result = ExportResult.FAILED_NOT_RETRYABLE
+
+                    if not self._is_stats_exporter():
+                        if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                            _track_dropped_items(self._customer_statsbeat_metrics, envelopes, response_error.status_code)
                 elif _is_redirect_code(response_error.status_code):
                     self._consecutive_redirects = self._consecutive_redirects + 1
                     # pylint: disable=W0212
@@ -286,12 +328,22 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                             result = self._transmit(envelopes)
                         else:
                             if not self._is_stats_exporter():
+                                if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                                    _track_dropped_items(self._customer_statsbeat_metrics, envelopes, DropCode.CLIENT_EXCEPTION, "Error parsing redirect information.")
                                 logger.error(
                                     "Error parsing redirect information.",
                                 )
                             result = ExportResult.FAILED_NOT_RETRYABLE
                     else:
                         if not self._is_stats_exporter():
+                            # Track dropped items in customer statsbeat, non-retryable scenario
+                            if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                                _track_dropped_items(
+                                    self._customer_statsbeat_metrics,
+                                    envelopes,
+                                    DropCode.CLIENT_EXCEPTION,
+                                    "Error sending telemetry because of circular redirects. Please check the integrity of your connection string."
+                                )
                             logger.error(
                                 "Error sending telemetry because of circular redirects. "
                                 "Please check the integrity of your connection string."
@@ -311,20 +363,33 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                             "Non-retryable server side error: %s.",
                             response_error.message,
                         )
+                        # Track dropped items in customer statsbeat, non-retryable scenario
+                        if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                            _track_dropped_items(self._customer_statsbeat_metrics, envelopes, response_error.status_code)
                         if _is_invalid_code(response_error.status_code):
                             # Shutdown statsbeat on invalid code from customer endpoint
                             # Import here to avoid circular dependencies
                             from azure.monitor.opentelemetry.exporter.statsbeat._statsbeat import (
                                 shutdown_statsbeat_metrics,
                             )
+                            from azure.monitor.opentelemetry.exporter.statsbeat._customer_statsbeat import (
+                                shutdown_customer_statsbeat_metrics,
+                            )
 
                             shutdown_statsbeat_metrics()
+                            # Also shutdown customer statsbeat on invalid code
+                            shutdown_customer_statsbeat_metrics()
                     result = ExportResult.FAILED_NOT_RETRYABLE
             except ServiceRequestError as request_error:
                 # Errors when we're fairly sure that the server did not receive the
                 # request, so it should be safe to retry.
                 # ServiceRequestError is raised by azure.core for these cases
                 logger.warning("Retrying due to server request error: %s.", request_error.message)
+
+                # Track retry items in customer statsbeat for client-side exceptions
+                if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                    _track_retry_items(self._customer_statsbeat_metrics, envelopes, request_error)
+
                 if self._should_collect_stats():
                     exc_type = request_error.exc_type
                     if exc_type is None or exc_type is type(None):
@@ -333,6 +398,11 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                 result = ExportResult.FAILED_RETRYABLE
             except Exception as ex:
                 logger.exception("Envelopes could not be exported and are not retryable: %s.", ex)  # pylint: disable=C4769
+
+                # Track dropped items in customer statsbeat for general exceptions
+                if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                    _track_dropped_items(self._customer_statsbeat_metrics, envelopes, DropCode.CLIENT_EXCEPTION, str(ex))
+
                 if self._should_collect_stats():
                     _update_requests_map(_REQ_EXCEPTION_NAME[1], value=ex.__class__.__name__)
                 result = ExportResult.FAILED_NOT_RETRYABLE
@@ -355,6 +425,7 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                             )
 
                             shutdown_statsbeat_metrics()
+
                             # pylint: disable=lost-exception
                             return ExportResult.FAILED_NOT_RETRYABLE  # pylint: disable=W0134
                 # pylint: disable=lost-exception
@@ -373,13 +444,31 @@ def _should_collect_stats(self):
             and not self._instrumentation_collection
         )
 
+
+    # check to see whether its the case of customer stats collection
+    def _should_collect_customer_statsbeat(self):
+        # Import here to avoid circular dependencies
+        from azure.monitor.opentelemetry.exporter.statsbeat._customer_statsbeat import get_customer_statsbeat_shutdown
+
+        env_value = os.environ.get("APPLICATIONINSIGHTS_STATSBEAT_ENABLED_PREVIEW", "")
+        is_customer_statsbeat_enabled = env_value.lower() == "true"
+        # Don't collect customer statsbeat for instrumentation collection or customer statsbeat exporters
+        return (
+            is_customer_statsbeat_enabled
+            and not get_customer_statsbeat_shutdown()
+            and not self._is_customer_stats_exporter()
+            and not self._instrumentation_collection
+        )
+
     # check to see if statsbeat is in "attempting to be initialized" state
     def _is_statsbeat_initializing_state(self):
         return self._is_stats_exporter() and not get_statsbeat_shutdown() and not get_statsbeat_initial_success()
 
     def _is_stats_exporter(self):
         return self.__class__.__name__ == "_StatsBeatExporter"
 
+    def _is_customer_stats_exporter(self):
+        return getattr(self, '_is_customer_statsbeat', False)
 
 def _is_invalid_code(response_code: Optional[int]) -> bool:
     """Determine if response is a invalid response.