Implement network statsbeat - success count (Azure#25752)

Author: lzchen

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

@@ -4,6 +4,9 @@
 
 ### Features Added
 
+- Implement success count network statsbeat
+    ([#25752](https://github.com/Azure/azure-sdk-for-python/pull/25752))
+
 ### Breaking Changes
 
 ### Bugs Fixed

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

@@ -14,7 +14,13 @@
 from azure.monitor.opentelemetry.exporter._generated.models import TelemetryItem
 from azure.monitor.opentelemetry.exporter._connection_string_parser import ConnectionStringParser
 from azure.monitor.opentelemetry.exporter._storage import LocalFileStorage
-
+from azure.monitor.opentelemetry.exporter.statsbeat._state import (
+    _REQ_SUCCESS_NAME,
+    _REQUESTS_LOCK,
+    _REQUESTS_MAP,
+    get_statsbeat_shutdown,
+    is_statsbeat_enabled,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -44,6 +50,7 @@ def __init__(self, **kwargs: Any) -> None:
         self._api_version = kwargs.get('api_version') or _SERVICE_API_LATEST
         self._consecutive_redirects = 0  # To prevent circular redirects
         self._enable_local_storage = kwargs.get('enable_local_storage', True)
+        self._endpoint = parsed_connection_string.endpoint
         self._instrumentation_key = parsed_connection_string.instrumentation_key
         self._storage_maintenance_period = kwargs.get('storage_maintenance_period', 60)  # Maintenance interval in seconds.
         self._storage_max_size = kwargs.get('storage_max_size', 50 * 1024 * 1024)  # Maximum size in bytes (default 50MiB)
@@ -56,8 +63,7 @@ def __init__(self, **kwargs: Any) -> None:
         self._storage_retention_period = kwargs.get('storage_retention_period', 7 * 24 * 60 * 60)  # Retention period in seconds
         self._timeout = kwargs.get('timeout', 10.0)  # networking timeout in seconds
 
-        config = AzureMonitorClientConfiguration(
-            parsed_connection_string.endpoint, **kwargs)
+        config = AzureMonitorClientConfiguration(self._endpoint, **kwargs)
         policies = [
             RequestIdPolicy(**kwargs),
             config.headers_policy,
@@ -86,6 +92,11 @@ def __init__(self, **kwargs: Any) -> None:
                 name="{} Storage".format(self.__class__.__name__),
                 lease_period=self._storage_min_retry_interval,
             )
+        # statsbeat initialization
+        if self._should_collect_stats():
+            # Import here to avoid circular dependencies
+            from azure.monitor.opentelemetry.exporter.statsbeat._statsbeat import collect_statsbeat_metrics
+            collect_statsbeat_metrics(self)
 
     def _transmit_from_storage(self) -> None:
         for blob in self.storage.gets():
@@ -121,13 +132,15 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
         if len(envelopes) > 0:
             try:
                 track_response = self.client.track(envelopes)
-                if not track_response.errors:
+                if not track_response.errors:  # 200
                     self._consecutive_redirects = 0
+                    if self._should_collect_stats():
+                        _update_requests_map(_REQ_SUCCESS_NAME[1])
                     logger.info("Transmission succeeded: Item received: %s. Items accepted: %s",
                                 track_response.items_received, track_response.items_accepted)
                     return ExportResult.SUCCESS
                 resend_envelopes = []
-                for error in track_response.errors:
+                for error in track_response.errors:  # 206
                     if _is_retryable_code(error.status_code):
                         resend_envelopes.append(
                             envelopes[error.index]
@@ -147,6 +160,7 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
                     return ExportResult.FAILED_RETRYABLE
                 return ExportResult.FAILED_NOT_RETRYABLE
             except HttpResponseError as response_error:
+                # HttpResponseError is raised when a response is received
                 if _is_retryable_code(response_error.status_code):
                     return ExportResult.FAILED_RETRYABLE
                 if _is_redirect_code(response_error.status_code):
@@ -174,6 +188,7 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
             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
                 )
@@ -187,6 +202,12 @@ def _transmit(self, envelopes: List[TelemetryItem]) -> ExportResult:
         self._consecutive_redirects = 0
         return ExportResult.SUCCESS
 
+    # check to see whether its the case of stats collection
+    def _should_collect_stats(self):
+        return is_statsbeat_enabled() and \
+            not get_statsbeat_shutdown() and \
+            self.__class__.__name__ != "_StatsBeatExporter"
+
 
 def _is_redirect_code(response_code: int) -> bool:
     """
@@ -212,3 +233,8 @@ def _is_retryable_code(response_code: int) -> bool:
         503,  # Service Unavailable
         504,  # Gateway timeout
     ))
+
+
+def _update_requests_map(type_name):
+    with _REQUESTS_LOCK:
+        _REQUESTS_MAP[type_name] = _REQUESTS_MAP.get(type_name, 0) + 1

sdk/monitor/azure-monitor-opentelemetry-exporter/azure/monitor/opentelemetry/exporter/export/metrics/_exporter.py

@@ -81,12 +81,12 @@ def export(
     def force_flush(
         self,
         timeout_millis: float = 10_000,
-    ) -> None:
+    ) -> bool:
         """
         Ensure that export of any metrics currently received by the exporter
         are completed as soon as possible.
         """
-        # TODO
+        return True
 
     def shutdown(
         self,

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

@@ -0,0 +1,34 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import logging
+
+from typing import Optional
+from opentelemetry.sdk.metrics.export import DataPointT
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.util.instrumentation import InstrumentationScope
+
+from azure.monitor.opentelemetry.exporter._generated.models import TelemetryItem
+from azure.monitor.opentelemetry.exporter import AzureMonitorMetricExporter
+from azure.monitor.opentelemetry.exporter.statsbeat._state import _STATSBEAT_METRIC_NAME_MAPPINGS
+
+_logger = logging.getLogger(__name__)
+
+
+class _StatsBeatExporter(AzureMonitorMetricExporter):
+
+    # pylint: disable=protected-access
+    def _point_to_envelope(
+        self,
+        point: DataPointT,
+        name: str,
+        resource: Optional[Resource] = None,
+        scope: Optional[InstrumentationScope] = None
+    ) -> TelemetryItem:
+        # map statsbeat name from OpenTelemetry name
+        name = _STATSBEAT_METRIC_NAME_MAPPINGS.get(name, "")
+        return super()._point_to_envelope(
+            point,
+            name,
+            resource,
+            scope,
+        )

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

@@ -0,0 +1,56 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import os
+import threading
+
+_REQUESTS_LOCK = threading.Lock()
+_REQUESTS_MAP = {}
+
+_STATSBEAT_STATE = {
+    "INITIAL_FAILURE_COUNT": 0,
+    "INITIAL_SUCCESS": False,
+    "SHUTDOWN": False,
+}
+_STATSBEAT_STATE_LOCK = threading.Lock()
+
+# (OpenTelemetry metric name, Statsbeat metric name)
+_ATTACH_METRIC_NAME = "Attach"
+_FEATURE_METRIC_NAME = "Feature"
+_REQ_SUCCESS_NAME = ("statsbeat_success_count", "Request Success Count")
+_REQ_FAILURE_NAME = "Request Failure Count"
+_REQ_DURATION_NAME = "Request Duration"
+_REQ_RETRY_NAME = "Retry Count"
+_REQ_THROTTLE_NAME = "Throttle Count"
+_REQ_EXCEPTION_NAME = "Exception Count"
+
+_STATSBEAT_METRIC_NAME_MAPPINGS = dict(
+    [
+        _REQ_SUCCESS_NAME,
+    ]
+)
+
+
+def is_statsbeat_enabled():
+    return not os.environ.get("APPLICATIONINSIGHTS_STATSBEAT_DISABLED_ALL")
+
+
+def increment_statsbeat_initial_failure_count():
+    with _STATSBEAT_STATE_LOCK:
+        _STATSBEAT_STATE["INITIAL_FAILURE_COUNT"] += 1
+
+
+def get_statsbeat_initial_failure_count():
+    return _STATSBEAT_STATE["INITIAL_FAILURE_COUNT"]
+
+
+def set_statsbeat_initial_success(success):
+    with _STATSBEAT_STATE_LOCK:
+        _STATSBEAT_STATE["INITIAL_SUCCESS"] = success
+
+
+def get_statsbeat_initial_success():
+    return _STATSBEAT_STATE["INITIAL_SUCCESS"]
+
+
+def get_statsbeat_shutdown():
+    return _STATSBEAT_STATE["SHUTDOWN"]

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

@@ -0,0 +1,107 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import os
+import threading
+
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+
+from azure.monitor.opentelemetry.exporter.statsbeat._exporter import _StatsBeatExporter
+from azure.monitor.opentelemetry.exporter.statsbeat._statsbeat_metrics import _StatsbeatMetrics
+
+# pylint: disable=line-too-long
+_DEFAULT_NON_EU_STATS_CONNECTION_STRING = "InstrumentationKey=c4a29126-a7cb-47e5-b348-11414998b11e;IngestionEndpoint=https://westus-0.in.applicationinsights.azure.com/"
+_DEFAULT_EU_STATS_CONNECTION_STRING = "InstrumentationKey=7dc56bab-3c0c-4e9f-9ebb-d1acadee8d0f;IngestionEndpoint=https://westeurope-5.in.applicationinsights.azure.com/"
+_DEFAULT_STATS_SHORT_EXPORT_INTERVAL = 900  # 15 minutes
+_DEFAULT_STATS_LONG_EXPORT_INTERVAL = 86400  # 24 hours
+_EU_ENDPOINTS = [
+    "westeurope",
+    "northeurope",
+    "francecentral",
+    "francesouth",
+    "germanywestcentral",
+    "norwayeast",
+    "norwaywest",
+    "swedencentral",
+    "switzerlandnorth",
+    "switzerlandwest",
+]
+
+_STATSBEAT_METER_PROVIDER = None
+_STATSBEAT_LOCK = threading.Lock()
+
+# pylint: disable=global-statement
+# pylint: disable=protected-access
+def collect_statsbeat_metrics(exporter) -> None:
+    global _STATSBEAT_METER_PROVIDER
+    # Only start statsbeat if did not exist before
+    if _STATSBEAT_METER_PROVIDER is None:
+        with _STATSBEAT_LOCK:
+            statsbeat_exporter = _StatsBeatExporter(
+                connection_string=_get_stats_connection_string(exporter._endpoint),
+            )
+            reader = PeriodicExportingMetricReader(
+                statsbeat_exporter,
+                export_interval_millis=_get_stats_short_export_interval() * 1000,  # 15m by default
+            )
+            _STATSBEAT_METER_PROVIDER = MeterProvider(metric_readers=[reader])
+            _StatsbeatMetrics(
+                _STATSBEAT_METER_PROVIDER,
+                exporter._instrumentation_key,
+                exporter._endpoint,
+            )
+            # Export some initial stats on program start
+            # TODO: initial stats
+            # TODO: set context
+            # execution_context.set_is_exporter(True)
+            # exporter.export_metrics(_STATSBEAT_METRICS.get_initial_metrics())
+            # execution_context.set_is_exporter(False)
+        # TODO: state
+        # with _STATSBEAT_STATE_LOCK:
+        #     _STATSBEAT_STATE["INITIAL_FAILURE_COUNT"] = 0
+        #     _STATSBEAT_STATE["INITIAL_SUCCESS"] = 0
+        #     _STATSBEAT_STATE["SHUTDOWN"] = False
+
+# TODO
+# def shutdown_statsbeat_metrics() -> None:
+#     global _STATSBEAT_METER_PROVIDER
+#     shutdown_success = False
+#     if _STATSBEAT_METER_PROVIDER is not None:
+#         with _STATSBEAT_LOCK:
+#             try:
+#                 _STATSBEAT_METER_PROVIDER.shutdown()
+#                 _STATSBEAT_METER_PROVIDER = None
+#                 shutdown_success = True
+#             except:
+#                 pass
+#         if shutdown_success:
+#             # with _STATSBEAT_STATE_LOCK:
+#             #     _STATSBEAT_STATE["SHUTDOWN"] = True
+#             pass
+
+
+def _get_stats_connection_string(endpoint: str) -> str:
+    cs_env = os.environ.get("APPLICATION_INSIGHTS_STATS_CONNECTION_STRING")
+    if cs_env:
+        return cs_env
+    for endpoint_location in _EU_ENDPOINTS:
+        if endpoint_location in endpoint:
+            # Use statsbeat EU endpoint if user is in EU region
+            return _DEFAULT_EU_STATS_CONNECTION_STRING
+    return _DEFAULT_NON_EU_STATS_CONNECTION_STRING
+
+
+# seconds
+def _get_stats_short_export_interval() -> float:
+    ei_env = os.environ.get("APPLICATION_INSIGHTS_STATS_SHORT_EXPORT_INTERVAL")
+    if ei_env:
+        return int(ei_env)
+    return _DEFAULT_STATS_SHORT_EXPORT_INTERVAL
+
+
+# seconds
+def _get_stats_long_export_interval() -> float:
+    ei_env = os.environ.get("APPLICATION_INSIGHTS_STATS_LONG_EXPORT_INTERVAL")
+    if ei_env:
+        return int(ei_env)
+    return _DEFAULT_STATS_LONG_EXPORT_INTERVAL