Customer Facing Statsbeat: Added remaining drop codes to base (#42382)

* Added remaining drop codes to base

* Updated CHANGELOG

* Fixed spell check errors

* Addressed comments

* Fixed spell check error

* Changed handling for readonly and exceptions

* Fix cspell error

* Updated LocalFileSotrage and LocalFileBlob

* Modified storage tests

* Linting

* Fixed cspell and lint errors

* cspell

* Addressed feedback

Author: rads-1996

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

@@ -3,6 +3,8 @@
 ## 1.0.0b42 (Unreleased)
 
 ### Features Added
+- Customer Facing Statsbeat: Added remaining drop codes to base
+  ([#42382](https://github.com/Azure/azure-sdk-for-python/pull/42382))
 
 ### Breaking Changes
 

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

@@ -7,9 +7,18 @@
 import os
 import random
 import subprocess
+import errno
+from enum import Enum
 
 from azure.monitor.opentelemetry.exporter._utils import PeriodicTask
 
+from azure.monitor.opentelemetry.exporter.statsbeat._state import (
+    get_local_storage_setup_state_exception,
+    get_local_storage_setup_state_readonly,
+    set_local_storage_setup_state_exception,
+    set_local_storage_setup_state_readonly,
+)
+
 logger = logging.getLogger(__name__)
 
 ICACLS_PATH = os.path.join(
@@ -27,6 +36,10 @@ def _now():
 def _seconds(seconds):
     return datetime.timedelta(seconds=seconds)
 
+class StorageExportResult(Enum):
+    CLIENT_STORAGE_DISABLED = 1
+    CLIENT_PERSISTENCE_CAPACITY_REACHED = 2
+    CLIENT_READONLY = 3
 
 # pylint: disable=broad-except
 class LocalFileBlob:
@@ -48,6 +61,8 @@ def get(self):
         return None
 
     def put(self, data, lease_period=0):
+        #TODO: Modify method to remove the return of self as it is not being used anywhere.
+        # Add typing to method
         try:
             fullpath = self.fullpath + ".tmp"
             with open(fullpath, "w", encoding="utf-8") as file:
@@ -62,9 +77,8 @@ def put(self, data, lease_period=0):
                 self.fullpath += "@{}.lock".format(_fmt(timestamp))
             os.rename(fullpath, self.fullpath)
             return self
-        except Exception:
-            pass  # keep silent
-        return None
+        except Exception as ex:
+            return str(ex)
 
     def lease(self, period):
         timestamp = _now() + _seconds(period)
@@ -96,7 +110,6 @@ def __init__(
         self._max_size = max_size
         self._retention_period = retention_period
         self._write_timeout = write_timeout
-
         self._enabled = self._check_and_set_folder_permissions()
         if self._enabled:
             self._maintenance_routine()
@@ -182,22 +195,34 @@ def get(self):
         return None
 
     def put(self, data, lease_period=None):
-        if not self._enabled:
-            return None
-        if not self._check_storage_size():
-            return None
-        blob = LocalFileBlob(
-            os.path.join(
-                self._path,
-                "{}-{}.blob".format(
-                    _fmt(_now()),
-                    "{:08x}".format(random.getrandbits(32)),  # thread-safe random
-                ),
+        # TODO: Remove the blob.put result as we are not using it anywhere and use StorageExportResult instead,
+        # Should still capture exceptions returned from LocalFileBlob.put
+        # Add typing for method
+        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())
+                return StorageExportResult.CLIENT_STORAGE_DISABLED
+            if not self._check_storage_size():
+                return StorageExportResult.CLIENT_PERSISTENCE_CAPACITY_REACHED
+            blob = LocalFileBlob(
+                os.path.join(
+                    self._path,
+                    "{}-{}.blob".format(
+                        _fmt(_now()),
+                        "{:08x}".format(random.getrandbits(32)),  # thread-safe random
+                    ),
+                )
             )
-        )
-        if lease_period is None:
-            lease_period = self._lease_period
-        return blob.put(data, lease_period=lease_period)
+            if lease_period is None:
+                lease_period = self._lease_period
+            return blob.put(data, lease_period=lease_period)
+        except Exception as ex:
+            return str(ex)
+
 
     def _check_and_set_folder_permissions(self):
         """
@@ -235,8 +260,13 @@ def _check_and_set_folder_permissions(self):
             else:
                 os.chmod(self._path, 0o700)
                 return True
-        except Exception:
-            pass  # keep silent
+        except OSError as error:
+            if getattr(error, 'errno', None) == errno.EROFS:  # cspell:disable-line
+                set_local_storage_setup_state_readonly()
+            else:
+                set_local_storage_setup_state_exception(str(error))
+        except Exception as ex:
+            set_local_storage_setup_state_exception(str(ex))
         return False
 
     def _check_storage_size(self):

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

@@ -46,14 +46,18 @@
 )
 # from azure.monitor.opentelemetry.exporter._configuration import _ConfigurationManager
 from azure.monitor.opentelemetry.exporter._connection_string_parser import ConnectionStringParser
-from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage
+from azure.monitor.opentelemetry.exporter._storage import (
+    LocalFileStorage,
+    StorageExportResult,
+)
 from azure.monitor.opentelemetry.exporter._utils import _get_auth_policy
 from azure.monitor.opentelemetry.exporter.statsbeat._state import (
     get_statsbeat_initial_success,
     get_statsbeat_shutdown,
     increment_and_check_statsbeat_failure_count,
     is_statsbeat_enabled,
     set_statsbeat_initial_success,
+    get_local_storage_setup_state_exception,
 )
 from azure.monitor.opentelemetry.exporter.statsbeat._utils import (
     _update_requests_map,
@@ -196,14 +200,35 @@ def _transmit_from_storage(self) -> None:
                 else:
                     blob.delete()
 
+
     def _handle_transmit_from_storage(self, envelopes: List[TelemetryItem], result: ExportResult) -> None:
         if self.storage:
             if result == ExportResult.FAILED_RETRYABLE:
                 envelopes_to_store = [x.as_dict() for x in envelopes]
-                self.storage.put(envelopes_to_store)
+                result_from_storage_put = self.storage.put(envelopes_to_store)
+                if self._customer_statsbeat_metrics and self._should_collect_customer_statsbeat():
+                    if result_from_storage_put == StorageExportResult.CLIENT_STORAGE_DISABLED:
+                        # Track items that would have been retried but are dropped since client has local storage disabled
+                        _track_dropped_items(self._customer_statsbeat_metrics, envelopes, DropCode.CLIENT_STORAGE_DISABLED)
+                    elif result_from_storage_put == StorageExportResult.CLIENT_READONLY:
+                        # If filesystem is readonly, track dropped items in customer statsbeat
+                        _track_dropped_items(self._customer_statsbeat_metrics, envelopes, DropCode.CLIENT_READONLY)
+                    elif result_from_storage_put == StorageExportResult.CLIENT_PERSISTENCE_CAPACITY_REACHED:
+                        # If data has to be dropped due to persistent storage being full, track dropped items
+                        _track_dropped_items(self._customer_statsbeat_metrics, envelopes, DropCode.CLIENT_PERSISTENCE_CAPACITY)
+                    elif get_local_storage_setup_state_exception() != "":
+                        # For exceptions caught in _check_and_set_folder_permissions during storage setup
+                        _track_dropped_items(self._customer_statsbeat_metrics, envelopes, DropCode.CLIENT_EXCEPTION, result_from_storage_put)
+                    elif isinstance(result_from_storage_put, str):
+                        # For any exceptions occurred in put method of either LocalFileStorage or LocalFileBlob, track dropped item with reason
+                        _track_dropped_items(self._customer_statsbeat_metrics, envelopes, DropCode.CLIENT_EXCEPTION, result_from_storage_put)
+                    else:
+                        # LocalFileBlob.put returns either an exception(failure, handled above) or the file path(success), eventually that will be removed since this value is not being utilized elsewhere
+                        pass
             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():
@@ -453,7 +478,7 @@ def _should_collect_stats(self):
     # 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
+        from azure.monitor.opentelemetry.exporter.statsbeat._state import get_customer_statsbeat_shutdown
 
         env_value = os.environ.get("APPLICATIONINSIGHTS_STATSBEAT_ENABLED_PREVIEW", "")
         is_customer_statsbeat_enabled = env_value.lower() == "true"

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

@@ -37,6 +37,11 @@
 )
 from azure.monitor.opentelemetry.exporter import VERSION
 
+from azure.monitor.opentelemetry.exporter.statsbeat._state import (
+    _CUSTOMER_STATSBEAT_STATE,
+    _CUSTOMER_STATSBEAT_STATE_LOCK,
+)
+
 _CUSTOMER_STATSBEAT_MAP_LOCK = threading.Lock()
 
 class _CustomerStatsbeatTelemetryCounters:
@@ -256,10 +261,6 @@ def collect_customer_statsbeat(exporter):
     if hasattr(exporter, 'storage') and exporter.storage:
         exporter.storage._customer_statsbeat_metrics = _CUSTOMER_STATSBEAT_METRICS
 
-_CUSTOMER_STATSBEAT_STATE = {
-    "SHUTDOWN": False,
-}
-_CUSTOMER_STATSBEAT_STATE_LOCK = threading.Lock()
 
 def shutdown_customer_statsbeat_metrics() -> None:
     global _CUSTOMER_STATSBEAT_METRICS
@@ -276,6 +277,3 @@ def shutdown_customer_statsbeat_metrics() -> None:
         if shutdown_success:
             with _CUSTOMER_STATSBEAT_STATE_LOCK:
                 _CUSTOMER_STATSBEAT_STATE["SHUTDOWN"] = True
-
-def get_customer_statsbeat_shutdown():
-    return _CUSTOMER_STATSBEAT_STATE["SHUTDOWN"]

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

@@ -19,6 +19,17 @@
 _STATSBEAT_STATE_LOCK = threading.Lock()
 _STATSBEAT_FAILURE_COUNT_THRESHOLD = 3
 
+_CUSTOMER_STATSBEAT_STATE = {
+    "SHUTDOWN": False,
+}
+_CUSTOMER_STATSBEAT_STATE_LOCK = threading.Lock()
+
+_LOCAL_STORAGE_SETUP_STATE = {
+    "READONLY": False,
+    "EXCEPTION_OCCURRED": ""
+}
+
+_LOCAL_STORAGE_SETUP_STATE_LOCK = threading.Lock()
 
 def is_statsbeat_enabled():
     disabled = os.environ.get(_APPLICATIONINSIGHTS_STATSBEAT_DISABLED_ALL)
@@ -68,3 +79,20 @@ def get_statsbeat_live_metrics_feature_set():
 def set_statsbeat_live_metrics_feature_set():
     with _STATSBEAT_STATE_LOCK:
         _STATSBEAT_STATE["LIVE_METRICS_FEATURE_SET"] = True
+
+def get_customer_statsbeat_shutdown():
+    return _CUSTOMER_STATSBEAT_STATE["SHUTDOWN"]
+
+def get_local_storage_setup_state_readonly():
+    return _LOCAL_STORAGE_SETUP_STATE["READONLY"]
+
+def set_local_storage_setup_state_readonly():
+    with _LOCAL_STORAGE_SETUP_STATE_LOCK:
+        _LOCAL_STORAGE_SETUP_STATE["READONLY"] = True
+
+def get_local_storage_setup_state_exception():
+    return _LOCAL_STORAGE_SETUP_STATE["EXCEPTION_OCCURRED"]
+
+def set_local_storage_setup_state_exception(value):
+    with _LOCAL_STORAGE_SETUP_STATE_LOCK:
+        _LOCAL_STORAGE_SETUP_STATE["EXCEPTION_OCCURRED"] = value