feat: Support OpenTelemetry Azure monitor distro (#1509)

* distro-commit

* resource detector

* tests

* lint

* Update constants.py

* fix

* Update test_opentelemetry.py

* Update test_opentelemetry.py

* fixing tests, added setting to logs

* reordered import statements

* formatting

* rename

* fix mock env variable

* Update dispatcher.py

* Update dispatcher.py

---------

Co-authored-by: hallvictoria <[email protected]>
Co-authored-by: gavin-aguiar <[email protected]>
Co-authored-by: Victoria Hall <[email protected]>

azure_functions_worker/constants.py

@@ -81,5 +81,15 @@
 # Base extension supported Python minor version
 BASE_EXT_SUPPORTED_PY_MINOR_VERSION = 8
 
+# Appsetting to turn on OpenTelemetry support/features
+# Includes turning on Azure monitor distro to send telemetry to AppInsights
 PYTHON_ENABLE_OPENTELEMETRY = "PYTHON_ENABLE_OPENTELEMETRY"
-PYTHON_ENABLE_OPENTELEMETRY_DEFAULT = True
+PYTHON_ENABLE_OPENTELEMETRY_DEFAULT = False
+
+# Appsetting to specify root logger name of logger to collect telemetry for
+# Used by Azure monitor distro
+PYTHON_AZURE_MONITOR_LOGGER_NAME = "PYTHON_AZURE_MONITOR_LOGGER_NAME"
+PYTHON_AZURE_MONITOR_LOGGER_NAME_DEFAULT = ""
+
+# Appsetting to specify AppInsights connection string
+APPLICATIONINSIGHTS_CONNECTION_STRING = "APPLICATIONINSIGHTS_CONNECTION_STRING"

azure_functions_worker/dispatcher.py

@@ -23,7 +23,10 @@
 from . import bindings, constants, functions, loader, protos
 from .bindings.shared_memory_data_transfer import SharedMemoryManager
 from .constants import (
+    APPLICATIONINSIGHTS_CONNECTION_STRING,
     METADATA_PROPERTIES_WORKER_INDEXED,
+    PYTHON_AZURE_MONITOR_LOGGER_NAME,
+    PYTHON_AZURE_MONITOR_LOGGER_NAME_DEFAULT,
     PYTHON_ENABLE_DEBUG_LOGGING,
     PYTHON_ENABLE_INIT_INDEXING,
     PYTHON_ENABLE_OPENTELEMETRY,
@@ -99,7 +102,7 @@ def __init__(self, loop: BaseEventLoop, host: str, port: int,
         self._function_metadata_exception = None
 
         # Used for checking if open telemetry is enabled
-        self._otel_libs_available = False
+        self._azure_monitor_available = False
         self._context_api = None
         self._trace_context_propagator = None
 
@@ -288,6 +291,46 @@ async def _dispatch_grpc_request(self, request):
         resp = await request_handler(request)
         self._grpc_resp_queue.put_nowait(resp)
 
+    def initialize_azure_monitor(self):
+        """Initializes OpenTelemetry and Azure monitor distro
+        """
+        self.update_opentelemetry_status()
+        try:
+            from azure.monitor.opentelemetry import configure_azure_monitor
+
+            # Set functions resource detector manually until officially
+            # include in Azure monitor distro
+            os.environ.setdefault(
+                "OTEL_EXPERIMENTAL_RESOURCE_DETECTORS",
+                "azure_functions",
+            )
+
+            configure_azure_monitor(
+                # Connection string can be explicitly specified in Appsetting
+                # If not set, defaults to env var
+                # APPLICATIONINSIGHTS_CONNECTION_STRING
+                connection_string=get_app_setting(
+                    setting=APPLICATIONINSIGHTS_CONNECTION_STRING
+                ),
+                logger_name=get_app_setting(
+                    setting=PYTHON_AZURE_MONITOR_LOGGER_NAME,
+                    default_value=PYTHON_AZURE_MONITOR_LOGGER_NAME_DEFAULT
+                ),
+            )
+            self._azure_monitor_available = True
+
+            logger.info("Successfully configured Azure monitor distro.")
+        except ImportError:
+            logger.exception(
+                "Cannot import Azure Monitor distro."
+            )
+            self._azure_monitor_available = False
+        except Exception:
+            logger.exception(
+                "Error initializing Azure monitor distro."
+            )
+            self._azure_monitor_available = False
+
     def update_opentelemetry_status(self):
         """Check for OpenTelemetry library availability and
         update the status attribute."""
@@ -299,12 +342,11 @@ def update_opentelemetry_status(self):
 
             self._context_api = context_api
             self._trace_context_propagator = TraceContextTextMapPropagator()
-            self._otel_libs_available = True
 
-            logger.info("Successfully loaded OpenTelemetry modules. "
-                        "OpenTelemetry is now enabled.")
         except ImportError:
-            self._otel_libs_available = False
+            logger.exception(
+                "Cannot import OpenTelemetry libraries."
+            )
 
     async def _handle__worker_init_request(self, request):
         logger.info('Received WorkerInitRequest, '
@@ -334,12 +376,11 @@ async def _handle__worker_init_request(self, request):
             constants.RPC_HTTP_TRIGGER_METADATA_REMOVED: _TRUE,
             constants.SHARED_MEMORY_DATA_TRANSFER: _TRUE,
         }
-
         if get_app_setting(setting=PYTHON_ENABLE_OPENTELEMETRY,
                            default_value=PYTHON_ENABLE_OPENTELEMETRY_DEFAULT):
-            self.update_opentelemetry_status()
+            self.initialize_azure_monitor()
 
-            if self._otel_libs_available:
+            if self._azure_monitor_available:
                 capabilities[constants.WORKER_OPEN_TELEMETRY_ENABLED] = _TRUE
 
         if DependencyManager.should_load_cx_dependencies():
@@ -611,7 +652,7 @@ async def _handle__invocation_request(self, request):
                     args[name] = bindings.Out()
 
             if fi.is_async:
-                if self._otel_libs_available:
+                if self._azure_monitor_available:
                     self.configure_opentelemetry(fi_context)
 
                 call_result = \
@@ -731,9 +772,9 @@ async def _handle__function_environment_reload_request(self, request):
             if get_app_setting(
                     setting=PYTHON_ENABLE_OPENTELEMETRY,
                     default_value=PYTHON_ENABLE_OPENTELEMETRY_DEFAULT):
-                self.update_opentelemetry_status()
+                self.initialize_azure_monitor()
 
-                if self._otel_libs_available:
+                if self._azure_monitor_available:
                     capabilities[constants.WORKER_OPEN_TELEMETRY_ENABLED] = (
                         _TRUE)
 
@@ -944,7 +985,7 @@ def _run_sync_func(self, invocation_id, context, func, params):
         # invocation_id from ThreadPoolExecutor's threads.
         context.thread_local_storage.invocation_id = invocation_id
         try:
-            if self._otel_libs_available:
+            if self._azure_monitor_available:
                 self.configure_opentelemetry(context)
             return ExtensionManager.get_sync_invocation_wrapper(context,
                                                                 func)(params)

azure_functions_worker/utils/app_setting_manager.py

@@ -7,6 +7,7 @@
     FUNCTIONS_WORKER_SHARED_MEMORY_DATA_TRANSFER_ENABLED,
     PYTHON_ENABLE_DEBUG_LOGGING,
     PYTHON_ENABLE_INIT_INDEXING,
+    PYTHON_ENABLE_OPENTELEMETRY,
     PYTHON_ENABLE_WORKER_EXTENSIONS,
     PYTHON_ENABLE_WORKER_EXTENSIONS_DEFAULT,
     PYTHON_ENABLE_WORKER_EXTENSIONS_DEFAULT_39,
@@ -27,7 +28,8 @@ def get_python_appsetting_state():
          PYTHON_ENABLE_WORKER_EXTENSIONS,
          FUNCTIONS_WORKER_SHARED_MEMORY_DATA_TRANSFER_ENABLED,
          PYTHON_SCRIPT_FILE_NAME,
-         PYTHON_ENABLE_INIT_INDEXING]
+         PYTHON_ENABLE_INIT_INDEXING,
+         PYTHON_ENABLE_OPENTELEMETRY]
 
     app_setting_states = "".join(
         f"{app_setting}: {current_vars[app_setting]} | "

tests/unittests/test_opentelemetry.py

@@ -1,4 +1,5 @@
 import asyncio
+import os
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -23,18 +24,45 @@ def test_update_opentelemetry_status_import_error(self):
         with patch('builtins.__import__', side_effect=ImportError):
             self.dispatcher.update_opentelemetry_status()
             # Verify that otel_libs_available is set to False due to ImportError
-            self.assertFalse(self.dispatcher._otel_libs_available)
+            self.assertFalse(self.dispatcher._azure_monitor_available)
 
     @patch('builtins.__import__')
     def test_update_opentelemetry_status_success(
             self, mock_imports):
         mock_imports.return_value = MagicMock()
         self.dispatcher.update_opentelemetry_status()
-        self.assertTrue(self.dispatcher._otel_libs_available)
+        self.assertIsNotNone(self.dispatcher._context_api)
+        self.assertIsNotNone(self.dispatcher._trace_context_propagator)
 
     @patch('builtins.__import__')
-    def test_init_request_otel_capability_enabled(
-            self, mock_imports):
+    @patch("azure_functions_worker.dispatcher.Dispatcher.update_opentelemetry_status")
+    def test_initialize_azure_monitor_success(
+        self,
+        mock_update_ot,
+        mock_imports,
+    ):
+        mock_imports.return_value = MagicMock()
+        self.dispatcher.initialize_azure_monitor()
+        mock_update_ot.assert_called_once()
+        self.assertTrue(self.dispatcher._azure_monitor_available)
+
+    @patch("azure_functions_worker.dispatcher.Dispatcher.update_opentelemetry_status")
+    def test_initialize_azure_monitor_import_error(
+        self,
+        mock_update_ot,
+    ):
+        with patch('builtins.__import__', side_effect=ImportError):
+            self.dispatcher.initialize_azure_monitor()
+            mock_update_ot.assert_called_once()
+            # Verify that otel_libs_available is set to False due to ImportError
+            self.assertFalse(self.dispatcher._azure_monitor_available)
+
+    @patch.dict(os.environ, {'PYTHON_ENABLE_OPENTELEMETRY': 'true'})
+    @patch('builtins.__import__')
+    def test_init_request_otel_capability_enabled_app_setting(
+            self,
+            mock_imports,
+    ):
         mock_imports.return_value = MagicMock()
 
         init_request = protos.StreamingMessage(
@@ -55,7 +83,11 @@ def test_init_request_otel_capability_enabled(
         self.assertIn("WorkerOpenTelemetryEnabled", capabilities)
         self.assertEqual(capabilities["WorkerOpenTelemetryEnabled"], "true")
 
-    def test_init_request_otel_capability_disabled(self):
+    @patch("azure_functions_worker.dispatcher.Dispatcher.initialize_azure_monitor")
+    def test_init_request_otel_capability_disabled_app_setting(
+        self,
+        mock_initialize_azmon,
+    ):
 
         init_request = protos.StreamingMessage(
             worker_init_request=protos.WorkerInitRequest(
@@ -70,5 +102,9 @@ def test_init_request_otel_capability_disabled(self):
         self.assertEqual(init_response.worker_init_response.result.status,
                          protos.StatusResult.Success)
 
+        # Azure monitor initialized not called
+        mock_initialize_azmon.assert_not_called()
+
+        # Verify that WorkerOpenTelemetryEnabled capability is not set
         capabilities = init_response.worker_init_response.capabilities
         self.assertNotIn("WorkerOpenTelemetryEnabled", capabilities)