Implement Customer Facing Statsbeat (#41669)

* Implementation for customer facing statsbeat

* Updated test cases

* Updated CHANGELOG with PR number

* Changes for successful items

* Deleted obselete test files

* Fixed _init_.py and reverted changes

* Commented out tests

* Fixed spelling error

* Disabled spell check

* Removed storage changes for now

* Updated CHANGELOG

* Fixed test failures

* Fixed failing tests

* Fixed linting errors

* Fixed newline lint error

* Addressed review comments

* Added new line in CHANGELOG

Author: rads-1996

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

@@ -6,6 +6,8 @@
 
 - Detect synthetically created telemetry based on the user-agent header
   ([#41733](https://github.com/Azure/azure-sdk-for-python/pull/41733))
+- Added customer-facing statsbeat preview.
+  ([#41669](https://github.com/Azure/azure-sdk-for-python/pull/41669))
 
 ### Breaking Changes
 

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

@@ -2,11 +2,13 @@
 # Licensed under the MIT License.
 # cSpell:disable
 
+from enum import Enum
 from opentelemetry.semconv.metrics import MetricInstruments
 from opentelemetry.semconv.metrics.http_metrics import (
     HTTP_CLIENT_REQUEST_DURATION,
     HTTP_SERVER_REQUEST_DURATION,
 )
+from azure.core import CaseInsensitiveEnumMeta
 
 # Environment variables
 
@@ -62,14 +64,18 @@
 _MESSAGE_ENVELOPE_NAME = "Microsoft.ApplicationInsights.Message"
 _REQUEST_ENVELOPE_NAME = "Microsoft.ApplicationInsights.Request"
 _REMOTE_DEPENDENCY_ENVELOPE_NAME = "Microsoft.ApplicationInsights.RemoteDependency"
+_REMOTE_DEPENDENCY_ENVELOPE_DATA = "Microsoft.ApplicationInsights.RemoteDependencyData"
+_EVENT_ENVELOPE_NAME = "Microsoft.ApplicationInsights.Event"
+_PAGE_VIEW_ENVELOPE_NAME = "Microsoft.ApplicationInsights.PageView"
+_PERFORMANCE_COUNTER_ENVELOPE_NAME = "Microsoft.ApplicationInsights.PerformanceCounter"
+_AVAILABILITY_ENVELOPE_NAME = "Microsoft.ApplicationInsights.Availability"
 
 # Feature constants
 _APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE = "APPLICATION_INSIGHTS_EVENT_MARKER_ATTRIBUTE"
 _AZURE_MONITOR_DISTRO_VERSION_ARG = "distro_version"
 _MICROSOFT_CUSTOM_EVENT_NAME = "microsoft.custom_event.name"
 
 # Statsbeat
-
 # (OpenTelemetry metric name, Statsbeat metric name)
 _ATTACH_METRIC_NAME = ("attach", "Attach")
 _FEATURE_METRIC_NAME = ("feature", "Feature")
@@ -115,6 +121,69 @@
     "ukwest",
 ]
 
+# Telemetry Types
+_AVAILABILITY = "AVAILABILITY"
+_CUSTOM_EVENT = "CUSTOM_EVENT"
+_CUSTOM_METRIC = "CUSTOM_METRIC"
+_DEPENDENCY = "DEPENDENCY"
+_EXCEPTION = "EXCEPTION"
+_PAGE_VIEW = "PAGE_VIEW"
+_PERFORMANCE_COUNTER = "PERFORMANCE_COUNTER"
+_REQUEST = "REQUEST"
+_TRACE = "TRACE"
+_UNKNOWN = "UNKNOWN"
+
+# Customer Facing Statsbeat
+_APPLICATIONINSIGHTS_STATSBEAT_ENABLED_PREVIEW = "APPLICATIONINSIGHTS_STATSBEAT_ENABLED_PREVIEW"
+
+_CUSTOMER_STATSBEAT_LANGUAGE = "python"
+
+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"
+    UNKNOWN = "UNKNOWN"
+
+class RetryCode(str, Enum, metaclass=CaseInsensitiveEnumMeta):
+    CLIENT_TIMEOUT = "CLIENT_TIMEOUT"
+    UNKNOWN = "UNKNOWN"
+
+class CustomerStatsbeatMetricName(str, Enum, metaclass=CaseInsensitiveEnumMeta):
+    ITEM_SUCCESS_COUNT = "preview.item.success.count"
+    ITEM_DROP_COUNT = "preview.item.dropped.count"
+    ITEM_RETRY_COUNT = "preview.item.retry.count"
+
+class CustomerStatsbeatProperties:
+    language: str
+    version: str
+    compute_type: str
+    def __init__(self, language: str, version: str, compute_type: str):
+        self.language = language
+        self.version = version
+        self.compute_type = compute_type
+
+## Map from Azure Monitor envelope names to TelemetryType
+_TYPE_MAP = {
+                _EVENT_ENVELOPE_NAME: _CUSTOM_EVENT,
+                _METRIC_ENVELOPE_NAME: _CUSTOM_METRIC,
+                _REMOTE_DEPENDENCY_ENVELOPE_DATA: _DEPENDENCY,
+                _EXCEPTION_ENVELOPE_NAME: _EXCEPTION,
+                _PAGE_VIEW_ENVELOPE_NAME: _PAGE_VIEW,
+                _MESSAGE_ENVELOPE_NAME: _TRACE,
+                _REQUEST_ENVELOPE_NAME: _REQUEST,
+                _PERFORMANCE_COUNTER_ENVELOPE_NAME: _PERFORMANCE_COUNTER,
+                _AVAILABILITY_ENVELOPE_NAME: _AVAILABILITY,
+            }
+
+# Map RP names
+class _RP_Names(Enum):
+    APP_SERVICE = "appsvc"
+    FUNCTIONS = "functions"
+    AKS = "aks"
+    VM = "vm"
+    UNKNOWN = "unknown"
+
 # Instrumentations
 
 # Special constant for azure-sdk opentelemetry instrumentation

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

@@ -28,7 +28,11 @@
     _PYTHON_APPLICATIONINSIGHTS_ENABLE_TELEMETRY,
     _WEBSITE_SITE_NAME,
 )
-
+from azure.monitor.opentelemetry.exporter._constants import (
+    _TYPE_MAP,
+    _UNKNOWN,
+    _RP_Names,
+)
 
 opentelemetry_version = ""
 
@@ -380,3 +384,19 @@ def __call__(cls, *args, **kwargs):
         if not cls._instance:
             cls._instance = super(Singleton, cls).__call__(*args, **kwargs)
         return cls._instance
+
+def _get_telemetry_type(item: TelemetryItem):
+    if hasattr(item, "data") and item.data is not None:
+        base_type = getattr(item.data, "base_type", None)
+        if base_type:
+            return _TYPE_MAP.get(base_type, _UNKNOWN)
+    return _UNKNOWN
+
+def get_compute_type():
+    if _is_on_functions():
+        return _RP_Names.FUNCTIONS.value
+    if _is_on_app_service():
+        return _RP_Names.APP_SERVICE.value
+    if _is_on_aks():
+        return _RP_Names.AKS.value
+    return _RP_Names.UNKNOWN.value

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

@@ -0,0 +1,92 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+"""Customer Statsbeat implementation for Azure Monitor OpenTelemetry Exporter.
+
+This module provides the implementation for collecting and reporting customer statsbeat
+metrics that track the usage and performance of the Azure Monitor OpenTelemetry Exporter.
+"""
+
+from typing import List, Dict, Any, Iterable
+import os
+
+from opentelemetry.metrics import CallbackOptions, Observation
+from opentelemetry.sdk.metrics import MeterProvider
+from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader
+
+from azure.monitor.opentelemetry.exporter._constants import (
+    _APPLICATIONINSIGHTS_STATSBEAT_ENABLED_PREVIEW,
+    _DEFAULT_STATS_SHORT_EXPORT_INTERVAL,
+    CustomerStatsbeatProperties,
+    #DropCode,
+    #RetryCode,
+    CustomerStatsbeatMetricName,
+    _CUSTOMER_STATSBEAT_LANGUAGE,
+)
+
+from azure.monitor.opentelemetry.exporter.statsbeat._exporter import AzureMonitorMetricExporter
+from azure.monitor.opentelemetry.exporter._utils import (
+    Singleton,
+    get_compute_type,
+)
+
+from azure.monitor.opentelemetry.exporter import VERSION
+
+class _CustomerStatsbeatTelemetryCounters:
+    def __init__(self):
+        self.total_item_success_count: Dict[str, Any] = {}
+
+class CustomerStatsbeatMetrics(metaclass=Singleton):
+    def __init__(self, options):
+        self._counters = _CustomerStatsbeatTelemetryCounters()
+        self._language = _CUSTOMER_STATSBEAT_LANGUAGE
+        self._is_enabled = os.environ.get(_APPLICATIONINSIGHTS_STATSBEAT_ENABLED_PREVIEW, "").lower() in ("true")
+        if not self._is_enabled:
+            return
+
+        exporter_config = {
+            "connection_string": options.connection_string,
+        }
+        self._customer_statsbeat_exporter = AzureMonitorMetricExporter(**exporter_config)
+        metric_reader_options = {
+            "exporter": self._customer_statsbeat_exporter,
+            "export_interval_millis": _DEFAULT_STATS_SHORT_EXPORT_INTERVAL
+        }
+        self._customer_statsbeat_metric_reader = PeriodicExportingMetricReader(**metric_reader_options)
+        self._customer_statsbeat_meter_provider = MeterProvider(
+            metric_readers=[self._customer_statsbeat_metric_reader]
+        )
+        self._customer_statsbeat_meter = self._customer_statsbeat_meter_provider.get_meter(__name__)
+        self._customer_properties = CustomerStatsbeatProperties(
+            language=self._language,
+            version=VERSION,
+            compute_type=get_compute_type(),
+        )
+        self._success_gauge = self._customer_statsbeat_meter.create_observable_gauge(
+            name=CustomerStatsbeatMetricName.ITEM_SUCCESS_COUNT.value,
+            description="Tracks successful telemetry items sent to Azure Monitor",
+            callbacks=[self._item_success_callback]
+        )
+
+    def count_successful_items(self, count: int, telemetry_type: str) -> None:
+        if not self._is_enabled or count <= 0:
+            return
+        if telemetry_type in self._counters.total_item_success_count:
+            self._counters.total_item_success_count[telemetry_type] += count
+        else:
+            self._counters.total_item_success_count[telemetry_type] = count
+
+    def _item_success_callback(self, options: CallbackOptions) -> Iterable[Observation]: # pylint: disable=unused-argument
+        if not getattr(self, "_is_enabled", False):
+            return []
+
+        observations: List[Observation] = []
+        for telemetry_type, count in self._counters.total_item_success_count.items():
+            attributes = {
+                "language": self._customer_properties.language,
+                "version": self._customer_properties.version,
+                "compute_type": self._customer_properties.compute_type,
+                "telemetry_type": telemetry_type
+            }
+            observations.append(Observation(count, dict(attributes)))
+
+        return observations

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

@@ -0,0 +1,144 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import os
+import unittest
+from unittest import mock
+import random
+import time
+import types
+import copy
+
+# Import directly from module to avoid circular imports
+from azure.monitor.opentelemetry.exporter.statsbeat._customer_statsbeat import CustomerStatsbeatMetrics
+
+from azure.monitor.opentelemetry.exporter._constants import (
+    _APPLICATIONINSIGHTS_STATSBEAT_ENABLED_PREVIEW,
+    _REQUEST,
+    _DEPENDENCY,
+    _REQ_RETRY_NAME,
+    _CUSTOMER_STATSBEAT_LANGUAGE,
+    _DEFAULT_STATS_SHORT_EXPORT_INTERVAL,
+    _UNKNOWN,
+    _TYPE_MAP,
+)
+
+from opentelemetry import trace
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SimpleSpanProcessor
+from azure.monitor.opentelemetry.exporter.export._base import (
+    BaseExporter,
+    ExportResult,
+    )
+from azure.monitor.opentelemetry.exporter import AzureMonitorTraceExporter
+from azure.monitor.opentelemetry.exporter.statsbeat._state import _REQUESTS_MAP
+
+class TestCustomerStatsbeat(unittest.TestCase):
+    """Tests for CustomerStatsbeatMetrics."""
+    def setUp(self):
+        _REQUESTS_MAP.clear()
+        # Clear singleton instance for test isolation
+        CustomerStatsbeatMetrics._instance = None
+        
+        self.env_patcher = mock.patch.dict(os.environ, {
+            "APPLICATIONINSIGHTS_STATSBEAT_ENABLED_PREVIEW": "true"
+        })
+        self.env_patcher.start()
+        self.mock_options = mock.Mock()
+        self.mock_options.instrumentation_key = "363331ca-f431-4119-bdcd-31a75920f958"
+        self.mock_options.network_collection_interval = _DEFAULT_STATS_SHORT_EXPORT_INTERVAL
+        self.mock_options.connection_string = "InstrumentationKey=363331ca-f431-4119-bdcd-31a75920f958;IngestionEndpoint=https://eastus-8.in.applicationinsights.azure.com/"
+        self.mock_options.language = _CUSTOMER_STATSBEAT_LANGUAGE
+        self.original_trace_provider = trace._TRACER_PROVIDER
+        trace._TRACER_PROVIDER = None
+        self.mock_options.metrics = CustomerStatsbeatMetrics(self.mock_options)
+        self.mock_options.transmit_called = [False]
+
+    def tearDown(self):
+        self.env_patcher.stop()
+        # Restore trace provider
+        trace._TRACER_PROVIDER = self.original_trace_provider
+        # Clean up singleton instances to prevent cross-test contamination
+        if hasattr(self.mock_options, 'metrics') and self.mock_options.metrics:
+            metrics = self.mock_options.metrics
+            if hasattr(metrics, '_customer_statsbeat_metric_reader'):
+                try:
+                    # Shutdown to prevent additional periodic exports
+                    metrics._customer_statsbeat_metric_reader.shutdown()
+                except Exception:
+                    pass  # Ignore shutdown errors
+        CustomerStatsbeatMetrics._instance = None
+    
+    def test_successful_item_count(self):
+        successful_dependencies = 0
+
+        metrics = self.mock_options.metrics
+        metrics._counters.total_item_success_count.clear()
+
+        exporter = AzureMonitorTraceExporter(connection_string=self.mock_options.connection_string)
+
+        def patched_transmit(self_exporter, envelopes):
+            self.mock_options.transmit_called[0] = True
+
+            for envelope in envelopes:
+                if not hasattr(envelope, "data") or not envelope.data:
+                    continue
+    
+                if envelope.data.base_type == "RemoteDependencyData":
+                    base_data = envelope.data.base_data
+                    if base_data and hasattr(base_data, "success"):
+                        if base_data.success:
+                            envelope_name = "Microsoft.ApplicationInsights." + envelope.data.base_type
+                            metrics.count_successful_items(1, _TYPE_MAP.get(envelope_name, _UNKNOWN))
+
+            return ExportResult.SUCCESS
+
+        exporter._transmit = types.MethodType(patched_transmit, exporter)
+
+        resource = Resource.create({"service.name": "dependency-test", "service.instance.id": "test-instance"})
+        trace_provider = TracerProvider(resource=resource)
+        
+        processor = SimpleSpanProcessor(exporter)
+        trace_provider.add_span_processor(processor)
+
+        tracer = trace_provider.get_tracer(__name__)
+
+        # Generate random number of dependencies (between 5 and 15)
+        total_dependencies = random.randint(5, 15)
+
+        for i in range(total_dependencies):
+            success = random.choice([True, False])
+            if success:
+                successful_dependencies += 1
+
+            with tracer.start_as_current_span(
+                name=f"{'success' if success else 'failed'}-dependency-{i}",
+                kind=trace.SpanKind.CLIENT,
+                attributes={
+                    "db.system": "mysql",
+                    "db.name": "test_db",
+                    "db.operation": "query",
+                    "net.peer.name": "test-db-server",
+                    "net.peer.port": 3306,
+                    "http.status_code": 200 if success else random.choice([400, 500, 503])
+                }
+            ) as span:
+                span.set_status(trace.Status(
+                    trace.StatusCode.OK if success else trace.StatusCode.ERROR
+                ))
+                time.sleep(0.1)
+
+        trace_provider.force_flush()
+        
+        self.metrics_instance = metrics
+        
+        self.assertTrue(self.mock_options.transmit_called[0], "Exporter _transmit method was not called")
+
+        actual_count = metrics._counters.total_item_success_count.get(_DEPENDENCY, 0)
+
+        self.assertEqual(
+            actual_count,
+            successful_dependencies,
+            f"Expected {successful_dependencies} successful dependencies, got {actual_count}"
+        )