From 33124bbe79565fdfa0399826a2db4ab1ee509046 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Fri, 26 Jun 2026 13:38:12 -0500 Subject: [PATCH 1/8] Move otel context propagation --- .../handle_event.py | 14 +- .../azure_functions_runtime/handle_event.py | 16 +- workers/azure_functions_worker/dispatcher.py | 13 +- workers/tests/unittests/test_opentelemetry.py | 192 ++++++++++++++++++ 4 files changed, 216 insertions(+), 19 deletions(-) diff --git a/runtimes/v1/azure_functions_runtime_v1/handle_event.py b/runtimes/v1/azure_functions_runtime_v1/handle_event.py index da3b39e2f..db0858a2e 100644 --- a/runtimes/v1/azure_functions_runtime_v1/handle_event.py +++ b/runtimes/v1/azure_functions_runtime_v1/handle_event.py @@ -156,6 +156,14 @@ async def invocation_request(request): fi: FunctionInfo = _functions.get_function( function_id) assert fi is not None + + # Initialize context and configure OpenTelemetry early + # to ensure all logs have proper trace context (Operation Id) + fi_context = get_context(invoc_request, fi.name, + fi.directory) + if otel_manager.get_azure_monitor_available(): + configure_opentelemetry(fi_context) + logger.info("Function name: %s, Function Type: %s", fi.name, ("async" if fi.is_async else "sync")) @@ -174,9 +182,6 @@ async def invocation_request(request): pb, trigger_metadata=trigger_metadata) - fi_context = get_context(invoc_request, fi.name, - fi.directory) - # Use local thread storage to store the invocation ID # for a customer's threads fi_context.thread_local_storage.invocation_id = invocation_id @@ -188,9 +193,6 @@ async def invocation_request(request): args[name] = Out() if fi.is_async: - if otel_manager.get_azure_monitor_available(): - configure_opentelemetry(fi_context) - # Not supporting Extensions call_result = await execute_async(fi.func, args) else: diff --git a/runtimes/v2/azure_functions_runtime/handle_event.py b/runtimes/v2/azure_functions_runtime/handle_event.py index 0ed133f93..677e1b130 100644 --- a/runtimes/v2/azure_functions_runtime/handle_event.py +++ b/runtimes/v2/azure_functions_runtime/handle_event.py @@ -182,6 +182,15 @@ async def invocation_request(request): fi: FunctionInfo = _functions.get_function( function_id) assert fi is not None + + # Initialize context and configure OpenTelemetry early + # to ensure all logs have proper trace context (Operation Id) + fi_context = get_context(invoc_request, fi.name, + fi.directory) + if (otel_manager.get_azure_monitor_available() + or otel_manager.get_otel_libs_available()): + configure_opentelemetry(fi_context) + logger.info("Function name: %s, Function Type: %s", fi.name, ("async" if fi.is_async else "sync")) @@ -217,9 +226,6 @@ async def invocation_request(request): await sync_http_request(http_request, func_http_request) args[trigger_arg_name] = http_request - fi_context = get_context(invoc_request, fi.name, - fi.directory) - # Use local thread storage to store the invocation ID # for a customer's threads fi_context.thread_local_storage.invocation_id = invocation_id @@ -234,10 +240,6 @@ async def invocation_request(request): args[name] = Out() if fi.is_async: - if (otel_manager.get_azure_monitor_available() - or otel_manager.get_otel_libs_available()): - configure_opentelemetry(fi_context) - # Extensions are not supported call_result = await execute_async(fi.func, args) else: diff --git a/workers/azure_functions_worker/dispatcher.py b/workers/azure_functions_worker/dispatcher.py index fa8be1117..37609e347 100644 --- a/workers/azure_functions_worker/dispatcher.py +++ b/workers/azure_functions_worker/dispatcher.py @@ -611,6 +611,13 @@ async def _handle__invocation_request(self, request): function_id) assert fi is not None + # Initialize context and configure OpenTelemetry early + # to ensure all logs have proper trace context (Operation Id) + fi_context = self._get_context(invoc_request, fi.name, + fi.directory) + if self._azure_monitor_available or self._otel_libs_available: + self.configure_opentelemetry(fi_context) + function_invocation_logs: List[str] = [ 'Received FunctionInvocationRequest', f'request ID: {self.request_id}', @@ -659,9 +666,6 @@ async def _handle__invocation_request(self, request): await sync_http_request(http_request, func_http_request) args[trigger_arg_name] = http_request - fi_context = self._get_context(invoc_request, fi.name, - fi.directory) - # Use local thread storage to store the invocation ID # for a customer's threads fi_context.thread_local_storage.invocation_id = invocation_id @@ -676,9 +680,6 @@ async def _handle__invocation_request(self, request): args[name] = bindings.Out() if fi.is_async: - if self._azure_monitor_available or self._otel_libs_available: - self.configure_opentelemetry(fi_context) - call_result = \ await self._run_async_func(fi_context, fi.func, args) else: diff --git a/workers/tests/unittests/test_opentelemetry.py b/workers/tests/unittests/test_opentelemetry.py index bdbedc874..925d04eff 100644 --- a/workers/tests/unittests/test_opentelemetry.py +++ b/workers/tests/unittests/test_opentelemetry.py @@ -217,3 +217,195 @@ def test_init_request_enable_azure_monitor_disabled_app_setting( # Verify that WorkerOpenTelemetryEnabled capability is not set capabilities = init_response.worker_init_response.capabilities self.assertNotIn("WorkerOpenTelemetryEnabled", capabilities) + + +class TestOpenTelemetryContextPropagation(unittest.TestCase): + """Tests to verify OpenTelemetry context is propagated before logging. + + Issue #1626: OpenTelemetry context must be configured before the first + log is emitted in _handle__invocation_request to ensure all logs have + proper Operation Id in Application Insights. + """ + + def setUp(self): + self.loop = asyncio.new_event_loop() + asyncio.set_event_loop(self.loop) + self.dispatcher = testutils.create_dummy_dispatcher() + self.call_order = [] + + def tearDown(self): + self.loop.close() + + def _track_configure_otel(self, *args, **kwargs): + """Track when configure_opentelemetry is called.""" + self.call_order.append('configure_opentelemetry') + + def _track_logger_info(self, *args, **kwargs): + """Track when logger.info is called.""" + self.call_order.append('logger_info') + + @patch.dict(os.environ, {'PYTHON_ENABLE_OPENTELEMETRY': 'true'}) + @patch('azure_functions_worker.dispatcher.logger') + def test_otel_configured_before_first_log_in_invocation_request( + self, + mock_logger, + ): + """Verify configure_opentelemetry is called before the first log. + + This test ensures that when OpenTelemetry is enabled, the context + is propagated before any logging occurs, so all logs have proper + trace context (Operation Id). + """ + # Set up tracking for call order + mock_logger.info.side_effect = self._track_logger_info + + # Enable OpenTelemetry on the dispatcher + self.dispatcher._otel_libs_available = True + self.dispatcher._azure_monitor_available = False + + # Mock configure_opentelemetry to track when it's called + self.dispatcher.configure_opentelemetry = MagicMock( + side_effect=self._track_configure_otel + ) + + # Mock _get_context to return a valid context + mock_context = MagicMock() + mock_context.trace_context = MagicMock() + mock_context.trace_context.trace_parent = "00-trace-parent" + mock_context.trace_context.trace_state = "state" + mock_context.thread_local_storage = MagicMock() + self.dispatcher._get_context = MagicMock(return_value=mock_context) + + # Mock the functions registry + mock_fi = MagicMock() + mock_fi.name = "test_function" + mock_fi.is_async = True + mock_fi.directory = "/test/dir" + mock_fi.input_types = {} + mock_fi.output_types = {} + mock_fi.requires_context = False + mock_fi.has_return = False + mock_fi.settlement_client_arg = None + mock_fi.func = MagicMock(return_value=None) + self.dispatcher._functions = MagicMock() + self.dispatcher._functions.get_function = MagicMock(return_value=mock_fi) + + # Create a mock invocation request + invoc_request = protos.StreamingMessage( + invocation_request=protos.InvocationRequest( + invocation_id="test-inv-123", + function_id="test-func-id", + trace_context=protos.RpcTraceContext( + trace_parent="00-trace-parent", + trace_state="state" + ) + ) + ) + + # Mock _run_async_func to return None + self.dispatcher._run_async_func = MagicMock( + return_value=asyncio.coroutine(lambda: None)() + ) + + # Run the invocation request (will fail but we only care about call order) + try: + self.loop.run_until_complete( + self.dispatcher._handle__invocation_request(invoc_request)) + except Exception: + # We expect this may fail due to incomplete mocking, + # but we only care about verifying call order + pass + + # Verify configure_opentelemetry was called + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + + # Find the position of configure_opentelemetry and first logger.info + if 'configure_opentelemetry' in self.call_order and \ + 'logger_info' in self.call_order: + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) + + @patch.dict(os.environ, {'PYTHON_APPLICATIONINSIGHTS_ENABLE_TELEMETRY': 'true'}) + @patch('azure_functions_worker.dispatcher.logger') + def test_azure_monitor_configured_before_first_log( + self, + mock_logger, + ): + """Verify configure_opentelemetry is called before first log with Azure Monitor.""" + # Set up tracking for call order + mock_logger.info.side_effect = self._track_logger_info + + # Enable Azure Monitor on the dispatcher + self.dispatcher._otel_libs_available = False + self.dispatcher._azure_monitor_available = True + + # Mock configure_opentelemetry to track when it's called + self.dispatcher.configure_opentelemetry = MagicMock( + side_effect=self._track_configure_otel + ) + + # Mock _get_context to return a valid context + mock_context = MagicMock() + mock_context.trace_context = MagicMock() + mock_context.trace_context.trace_parent = "00-trace-parent" + mock_context.trace_context.trace_state = "state" + mock_context.thread_local_storage = MagicMock() + self.dispatcher._get_context = MagicMock(return_value=mock_context) + + # Mock the functions registry + mock_fi = MagicMock() + mock_fi.name = "test_function" + mock_fi.is_async = True + mock_fi.directory = "/test/dir" + mock_fi.input_types = {} + mock_fi.output_types = {} + mock_fi.requires_context = False + mock_fi.has_return = False + mock_fi.settlement_client_arg = None + mock_fi.func = MagicMock(return_value=None) + self.dispatcher._functions = MagicMock() + self.dispatcher._functions.get_function = MagicMock(return_value=mock_fi) + + # Create a mock invocation request + invoc_request = protos.StreamingMessage( + invocation_request=protos.InvocationRequest( + invocation_id="test-inv-456", + function_id="test-func-id", + trace_context=protos.RpcTraceContext( + trace_parent="00-trace-parent", + trace_state="state" + ) + ) + ) + + # Mock _run_async_func to return None + self.dispatcher._run_async_func = MagicMock( + return_value=asyncio.coroutine(lambda: None)() + ) + + # Run the invocation request + try: + self.loop.run_until_complete( + self.dispatcher._handle__invocation_request(invoc_request)) + except Exception: + pass + + # Verify configure_opentelemetry was called before first log + if 'configure_opentelemetry' in self.call_order and \ + 'logger_info' in self.call_order: + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry should be called before first log. " + f"Call order: {self.call_order}" + ) From 91aa85c55a15df8abb1faf36bdc10ca73b084d79 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Thu, 23 Jul 2026 14:16:45 -0500 Subject: [PATCH 2/8] minor updates --- .../handle_event.py | 7 +- .../v1/tests/unittests/test_opentelemetry.py | 141 ++++++++++++++++- .../azure_functions_runtime/handle_event.py | 4 +- .../v2/tests/unittests/test_opentelemetry.py | 143 +++++++++++++++++- workers/azure_functions_worker/dispatcher.py | 4 +- .../azure_functions_worker/protos/__init__.py | 1 + workers/proxy_worker/protos/__init__.py | 1 + workers/tests/unittests/test_opentelemetry.py | 78 ++++++---- 8 files changed, 336 insertions(+), 43 deletions(-) diff --git a/runtimes/v1/azure_functions_runtime_v1/handle_event.py b/runtimes/v1/azure_functions_runtime_v1/handle_event.py index db0858a2e..66a245cdc 100644 --- a/runtimes/v1/azure_functions_runtime_v1/handle_event.py +++ b/runtimes/v1/azure_functions_runtime_v1/handle_event.py @@ -157,11 +157,12 @@ async def invocation_request(request): function_id) assert fi is not None - # Initialize context and configure OpenTelemetry early - # to ensure all logs have proper trace context (Operation Id) + # Initialize context and configure OpenTelemetry before emitting + # invocation-scoped logs so they include trace context (Operation Id) fi_context = get_context(invoc_request, fi.name, fi.directory) - if otel_manager.get_azure_monitor_available(): + if (otel_manager.get_azure_monitor_available() + or otel_manager.get_otel_libs_available()): configure_opentelemetry(fi_context) logger.info("Function name: %s, Function Type: %s", diff --git a/runtimes/v1/tests/unittests/test_opentelemetry.py b/runtimes/v1/tests/unittests/test_opentelemetry.py index f992ac85d..2cb25d1b5 100644 --- a/runtimes/v1/tests/unittests/test_opentelemetry.py +++ b/runtimes/v1/tests/unittests/test_opentelemetry.py @@ -4,13 +4,15 @@ import tests.protos as protos -from azure_functions_runtime_v1.handle_event import otel_manager, worker_init_request +from azure_functions_runtime_v1.handle_event import (otel_manager, worker_init_request, + invocation_request) from azure_functions_runtime_v1.otel import (initialize_azure_monitor, update_opentelemetry_status) from azure_functions_runtime_v1.logging import logger from tests.utils.constants import UNIT_TESTS_FOLDER from tests.utils.mock_classes import FunctionRequest, Request, WorkerRequest -from unittest.mock import MagicMock, patch +from tests.utils import testutils +from unittest.mock import AsyncMock, MagicMock, patch FUNCTION_APP_DIRECTORY = UNIT_TESTS_FOLDER / 'basic_functions' @@ -207,3 +209,138 @@ async def test_init_request_enable_azure_monitor_disabled_app_setting( # Verify that WorkerOpenTelemetryEnabled capability is not set capabilities = init_response.capabilities self.assertNotIn("WorkerOpenTelemetryEnabled", capabilities) + + +class TestOpenTelemetryContextPropagation(testutils.AsyncTestCase): + """Tests to verify OpenTelemetry context is propagated before logging in v1. + + Issue #1626: OpenTelemetry context must be configured before the first + log is emitted in invocation_request to ensure all logs have proper + Operation Id in Application Insights. + """ + + def setUp(self): + self.call_order = [] + + def _track_configure_otel(self, *args, **kwargs): + self.call_order.append('configure_opentelemetry') + + def _track_logger_info(self, *args, **kwargs): + self.call_order.append('logger_info') + + def _make_mock_request(self, invocation_id="test-inv-id", + function_id="test-func-id"): + """Create a minimal mock invocation request.""" + mock_invoc = MagicMock() + mock_invoc.invocation_id = invocation_id + mock_invoc.function_id = function_id + mock_invoc.input_data = [] + + mock_request = MagicMock() + mock_request.request.invocation_request = mock_invoc + return mock_request + + def _make_mock_fi(self): + """Create a minimal mock FunctionInfo.""" + mock_fi = MagicMock() + mock_fi.name = "test_function" + mock_fi.is_async = True + mock_fi.directory = "/test/dir" + mock_fi.input_types = {} + mock_fi.output_types = {} + mock_fi.requires_context = False + mock_fi.has_return = False + mock_fi.return_type = None + return mock_fi + + @patch("azure_functions_runtime_v1.handle_event" + ".otel_manager.get_azure_monitor_available", return_value=True) + @patch("azure_functions_runtime_v1.handle_event" + ".otel_manager.get_otel_libs_available", return_value=False) + @patch("azure_functions_runtime_v1.handle_event.execute_async", + new_callable=AsyncMock) + @patch("azure_functions_runtime_v1.handle_event.get_context") + @patch("azure_functions_runtime_v1.handle_event._functions") + @patch("azure_functions_runtime_v1.handle_event.configure_opentelemetry") + @patch("azure_functions_runtime_v1.handle_event.logger") + async def test_otel_configured_before_first_log_azure_monitor( + self, + mock_logger, + mock_configure_otel, + mock_functions, + mock_get_context, + mock_execute_async, + mock_get_otel_libs, + mock_get_azure_monitor, + ): + """Verify configure_opentelemetry is called before first log (Azure Monitor).""" + mock_logger.info.side_effect = self._track_logger_info + mock_configure_otel.side_effect = self._track_configure_otel + mock_functions.get_function.return_value = self._make_mock_fi() + mock_get_context.return_value = MagicMock() + mock_execute_async.return_value = None + + try: + await invocation_request(self._make_mock_request()) + except Exception: + pass + + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) + + @patch("azure_functions_runtime_v1.handle_event" + ".otel_manager.get_azure_monitor_available", return_value=False) + @patch("azure_functions_runtime_v1.handle_event" + ".otel_manager.get_otel_libs_available", return_value=True) + @patch("azure_functions_runtime_v1.handle_event.execute_async", + new_callable=AsyncMock) + @patch("azure_functions_runtime_v1.handle_event.get_context") + @patch("azure_functions_runtime_v1.handle_event._functions") + @patch("azure_functions_runtime_v1.handle_event.configure_opentelemetry") + @patch("azure_functions_runtime_v1.handle_event.logger") + async def test_otel_configured_before_first_log_otel_libs( + self, + mock_logger, + mock_configure_otel, + mock_functions, + mock_get_context, + mock_execute_async, + mock_get_otel_libs, + mock_get_azure_monitor, + ): + """Verify configure_opentelemetry is called before first log (otel libs).""" + mock_logger.info.side_effect = self._track_logger_info + mock_configure_otel.side_effect = self._track_configure_otel + mock_functions.get_function.return_value = self._make_mock_fi() + mock_get_context.return_value = MagicMock() + mock_execute_async.return_value = None + + try: + await invocation_request(self._make_mock_request()) + except Exception: + pass + + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) diff --git a/runtimes/v2/azure_functions_runtime/handle_event.py b/runtimes/v2/azure_functions_runtime/handle_event.py index 677e1b130..8dbc19880 100644 --- a/runtimes/v2/azure_functions_runtime/handle_event.py +++ b/runtimes/v2/azure_functions_runtime/handle_event.py @@ -183,8 +183,8 @@ async def invocation_request(request): function_id) assert fi is not None - # Initialize context and configure OpenTelemetry early - # to ensure all logs have proper trace context (Operation Id) + # Initialize context and configure OpenTelemetry before emitting + # invocation-scoped logs so they include trace context (Operation Id) fi_context = get_context(invoc_request, fi.name, fi.directory) if (otel_manager.get_azure_monitor_available() diff --git a/runtimes/v2/tests/unittests/test_opentelemetry.py b/runtimes/v2/tests/unittests/test_opentelemetry.py index 696dfcb55..3043498da 100644 --- a/runtimes/v2/tests/unittests/test_opentelemetry.py +++ b/runtimes/v2/tests/unittests/test_opentelemetry.py @@ -4,14 +4,16 @@ import tests.protos as protos -from azure_functions_runtime.handle_event import otel_manager, worker_init_request +from azure_functions_runtime.handle_event import (otel_manager, worker_init_request, + invocation_request) from azure_functions_runtime.otel import (initialize_azure_monitor, update_opentelemetry_status, OTelManager) from azure_functions_runtime.logging import logger from tests.utils.constants import UNIT_TESTS_FOLDER from tests.utils.mock_classes import FunctionRequest, Request, WorkerRequest -from unittest.mock import MagicMock, patch +from tests.utils import testutils +from unittest.mock import AsyncMock, MagicMock, patch FUNCTION_APP_DIRECTORY = UNIT_TESTS_FOLDER / 'basic_functions' @@ -242,3 +244,140 @@ def test_set_and_get_trace_context_propagator(self): dummy_propagator = object() self.manager.set_trace_context_propagator(dummy_propagator) self.assertIs(self.manager.get_trace_context_propagator(), dummy_propagator) + + +class TestOpenTelemetryContextPropagation(testutils.AsyncTestCase): + """Tests to verify OpenTelemetry context is propagated before logging in v2. + + Issue #1626: OpenTelemetry context must be configured before the first + log is emitted in invocation_request to ensure all logs have proper + Operation Id in Application Insights. + """ + + def setUp(self): + self.call_order = [] + + def _track_configure_otel(self, *args, **kwargs): + self.call_order.append('configure_opentelemetry') + + def _track_logger_info(self, *args, **kwargs): + self.call_order.append('logger_info') + + def _make_mock_request(self, invocation_id="test-inv-id", + function_id="test-func-id"): + """Create a minimal mock invocation request.""" + mock_invoc = MagicMock() + mock_invoc.invocation_id = invocation_id + mock_invoc.function_id = function_id + mock_invoc.input_data = [] + + mock_request = MagicMock() + mock_request.request.invocation_request = mock_invoc + return mock_request + + def _make_mock_fi(self): + """Create a minimal mock FunctionInfo.""" + mock_fi = MagicMock() + mock_fi.name = "test_function" + mock_fi.is_async = True + mock_fi.directory = "/test/dir" + mock_fi.input_types = {} + mock_fi.output_types = {} + mock_fi.requires_context = False + mock_fi.has_return = False + mock_fi.return_type = None + mock_fi.settlement_client_arg = None + mock_fi.is_http_func = False + return mock_fi + + @patch("azure_functions_runtime.handle_event" + ".otel_manager.get_azure_monitor_available", return_value=True) + @patch("azure_functions_runtime.handle_event" + ".otel_manager.get_otel_libs_available", return_value=False) + @patch("azure_functions_runtime.handle_event.execute_async", + new_callable=AsyncMock) + @patch("azure_functions_runtime.handle_event.get_context") + @patch("azure_functions_runtime.handle_event._functions") + @patch("azure_functions_runtime.handle_event.configure_opentelemetry") + @patch("azure_functions_runtime.handle_event.logger") + async def test_otel_configured_before_first_log_azure_monitor( + self, + mock_logger, + mock_configure_otel, + mock_functions, + mock_get_context, + mock_execute_async, + mock_get_otel_libs, + mock_get_azure_monitor, + ): + """Verify configure_opentelemetry is called before first log (Azure Monitor).""" + mock_logger.info.side_effect = self._track_logger_info + mock_configure_otel.side_effect = self._track_configure_otel + mock_functions.get_function.return_value = self._make_mock_fi() + mock_get_context.return_value = MagicMock() + mock_execute_async.return_value = None + + try: + await invocation_request(self._make_mock_request()) + except Exception: + pass + + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) + + @patch("azure_functions_runtime.handle_event" + ".otel_manager.get_azure_monitor_available", return_value=False) + @patch("azure_functions_runtime.handle_event" + ".otel_manager.get_otel_libs_available", return_value=True) + @patch("azure_functions_runtime.handle_event.execute_async", + new_callable=AsyncMock) + @patch("azure_functions_runtime.handle_event.get_context") + @patch("azure_functions_runtime.handle_event._functions") + @patch("azure_functions_runtime.handle_event.configure_opentelemetry") + @patch("azure_functions_runtime.handle_event.logger") + async def test_otel_configured_before_first_log_otel_libs( + self, + mock_logger, + mock_configure_otel, + mock_functions, + mock_get_context, + mock_execute_async, + mock_get_otel_libs, + mock_get_azure_monitor, + ): + """Verify configure_opentelemetry is called before first log (otel libs).""" + mock_logger.info.side_effect = self._track_logger_info + mock_configure_otel.side_effect = self._track_configure_otel + mock_functions.get_function.return_value = self._make_mock_fi() + mock_get_context.return_value = MagicMock() + mock_execute_async.return_value = None + + try: + await invocation_request(self._make_mock_request()) + except Exception: + pass + + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) diff --git a/workers/azure_functions_worker/dispatcher.py b/workers/azure_functions_worker/dispatcher.py index 37609e347..2ae77ccba 100644 --- a/workers/azure_functions_worker/dispatcher.py +++ b/workers/azure_functions_worker/dispatcher.py @@ -611,8 +611,8 @@ async def _handle__invocation_request(self, request): function_id) assert fi is not None - # Initialize context and configure OpenTelemetry early - # to ensure all logs have proper trace context (Operation Id) + # Initialize context and configure OpenTelemetry before emitting + # invocation-scoped logs so they include trace context (Operation Id) fi_context = self._get_context(invoc_request, fi.name, fi.directory) if self._azure_monitor_available or self._otel_libs_available: diff --git a/workers/azure_functions_worker/protos/__init__.py b/workers/azure_functions_worker/protos/__init__.py index e9c4f2397..8123fb2cc 100644 --- a/workers/azure_functions_worker/protos/__init__.py +++ b/workers/azure_functions_worker/protos/__init__.py @@ -28,6 +28,7 @@ RpcLog, RpcSharedMemory, RpcDataType, + RpcTraceContext, CloseSharedMemoryResourcesRequest, CloseSharedMemoryResourcesResponse, FunctionsMetadataRequest, diff --git a/workers/proxy_worker/protos/__init__.py b/workers/proxy_worker/protos/__init__.py index e9c4f2397..8123fb2cc 100644 --- a/workers/proxy_worker/protos/__init__.py +++ b/workers/proxy_worker/protos/__init__.py @@ -28,6 +28,7 @@ RpcLog, RpcSharedMemory, RpcDataType, + RpcTraceContext, CloseSharedMemoryResourcesRequest, CloseSharedMemoryResourcesResponse, FunctionsMetadataRequest, diff --git a/workers/tests/unittests/test_opentelemetry.py b/workers/tests/unittests/test_opentelemetry.py index 925d04eff..31e2e04c5 100644 --- a/workers/tests/unittests/test_opentelemetry.py +++ b/workers/tests/unittests/test_opentelemetry.py @@ -8,6 +8,7 @@ from tests.utils import testutils from azure_functions_worker import protos +from azure_functions_worker.dispatcher import ContextEnabledTask class TestOpenTelemetry(unittest.TestCase): @@ -228,9 +229,20 @@ class TestOpenTelemetryContextPropagation(unittest.TestCase): """ def setUp(self): + # Import directly from azure_functions_worker so this test always tests + # the correct dispatcher regardless of the Python version (testutils + # redirects to proxy_worker on Python >= 3.13). + from azure_functions_worker.dispatcher import Dispatcher, ContextEnabledTask self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) - self.dispatcher = testutils.create_dummy_dispatcher() + # Use ContextEnabledTask factory so dispatcher's assertion passes + self.loop.set_task_factory( + lambda loop, coro, context=None: ContextEnabledTask( + coro, loop=loop, context=context)) + self.dispatcher = Dispatcher( + self.loop, '127.0.0.1', 0, + 'test_worker_id', 'test_request_id', + 1.0, 1000) self.call_order = [] def tearDown(self): @@ -303,9 +315,9 @@ def test_otel_configured_before_first_log_in_invocation_request( ) # Mock _run_async_func to return None - self.dispatcher._run_async_func = MagicMock( - return_value=asyncio.coroutine(lambda: None)() - ) + async def _noop(): + return None + self.dispatcher._run_async_func = MagicMock(return_value=_noop()) # Run the invocation request (will fail but we only care about call order) try: @@ -316,22 +328,21 @@ def test_otel_configured_before_first_log_in_invocation_request( # but we only care about verifying call order pass - # Verify configure_opentelemetry was called + # Assert both events were observed self.assertIn('configure_opentelemetry', self.call_order, "configure_opentelemetry should have been called") - - # Find the position of configure_opentelemetry and first logger.info - if 'configure_opentelemetry' in self.call_order and \ - 'logger_info' in self.call_order: - otel_index = self.call_order.index('configure_opentelemetry') - first_log_index = self.call_order.index('logger_info') - - self.assertLess( - otel_index, first_log_index, - f"configure_opentelemetry (index {otel_index}) should be called " - f"before the first logger.info (index {first_log_index}). " - f"Call order: {self.call_order}" - ) + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + + # Assert configure_opentelemetry was called before the first logger.info + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) @patch.dict(os.environ, {'PYTHON_APPLICATIONINSIGHTS_ENABLE_TELEMETRY': 'true'}) @patch('azure_functions_worker.dispatcher.logger') @@ -387,9 +398,9 @@ def test_azure_monitor_configured_before_first_log( ) # Mock _run_async_func to return None - self.dispatcher._run_async_func = MagicMock( - return_value=asyncio.coroutine(lambda: None)() - ) + async def _noop(): + return None + self.dispatcher._run_async_func = MagicMock(return_value=_noop()) # Run the invocation request try: @@ -398,14 +409,17 @@ def test_azure_monitor_configured_before_first_log( except Exception: pass - # Verify configure_opentelemetry was called before first log - if 'configure_opentelemetry' in self.call_order and \ - 'logger_info' in self.call_order: - otel_index = self.call_order.index('configure_opentelemetry') - first_log_index = self.call_order.index('logger_info') - - self.assertLess( - otel_index, first_log_index, - f"configure_opentelemetry should be called before first log. " - f"Call order: {self.call_order}" - ) + # Assert both events were observed + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + + # Assert configure_opentelemetry was called before the first logger.info + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry should be called before first log. " + f"Call order: {self.call_order}" + ) From 70169130aa5229d15c2fd6eed8856427a715fddf Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Mon, 10 Aug 2026 14:55:58 -0500 Subject: [PATCH 3/8] add tests --- .../v1/tests/unittests/test_opentelemetry.py | 57 ++++++++++++- .../v2/tests/unittests/test_opentelemetry.py | 57 ++++++++++++- workers/tests/unittests/test_opentelemetry.py | 79 ++++++++++++++++++- 3 files changed, 186 insertions(+), 7 deletions(-) diff --git a/runtimes/v1/tests/unittests/test_opentelemetry.py b/runtimes/v1/tests/unittests/test_opentelemetry.py index 2cb25d1b5..ac70901f1 100644 --- a/runtimes/v1/tests/unittests/test_opentelemetry.py +++ b/runtimes/v1/tests/unittests/test_opentelemetry.py @@ -4,8 +4,11 @@ import tests.protos as protos -from azure_functions_runtime_v1.handle_event import (otel_manager, worker_init_request, - invocation_request) +from azure_functions_runtime_v1.handle_event import ( + invocation_request, + otel_manager, + worker_init_request, +) from azure_functions_runtime_v1.otel import (initialize_azure_monitor, update_opentelemetry_status) from azure_functions_runtime_v1.logging import logger @@ -344,3 +347,53 @@ async def test_otel_configured_before_first_log_otel_libs( f"before the first logger.info (index {first_log_index}). " f"Call order: {self.call_order}" ) + + @patch("azure_functions_runtime_v1.handle_event" + ".otel_manager.get_azure_monitor_available", return_value=True) + @patch("azure_functions_runtime_v1.handle_event" + ".otel_manager.get_otel_libs_available", return_value=False) + @patch("azure_functions_runtime_v1.handle_event.run_sync_func", + return_value=None) + @patch("azure_functions_runtime_v1.handle_event.get_context") + @patch("azure_functions_runtime_v1.handle_event._functions") + @patch("azure_functions_runtime_v1.handle_event.configure_opentelemetry") + @patch("azure_functions_runtime_v1.handle_event.logger") + async def test_otel_configured_before_first_log_sync_function( + self, + mock_logger, + mock_configure_otel, + mock_functions, + mock_get_context, + mock_run_sync_func, + mock_get_otel_libs, + mock_get_azure_monitor, + ): + """Verify configure_opentelemetry is called before first log (sync function).""" + mock_logger.info.side_effect = self._track_logger_info + mock_configure_otel.side_effect = self._track_configure_otel + mock_fi = self._make_mock_fi() + mock_fi.is_async = False + mock_functions.get_function.return_value = mock_fi + invocation_context = MagicMock() + mock_get_context.return_value = invocation_context + request = self._make_mock_request() + + try: + await invocation_request(request) + except Exception: + pass + + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + mock_configure_otel.assert_called_once_with(invocation_context) + + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) diff --git a/runtimes/v2/tests/unittests/test_opentelemetry.py b/runtimes/v2/tests/unittests/test_opentelemetry.py index 3043498da..b16b1890f 100644 --- a/runtimes/v2/tests/unittests/test_opentelemetry.py +++ b/runtimes/v2/tests/unittests/test_opentelemetry.py @@ -4,8 +4,11 @@ import tests.protos as protos -from azure_functions_runtime.handle_event import (otel_manager, worker_init_request, - invocation_request) +from azure_functions_runtime.handle_event import ( + invocation_request, + otel_manager, + worker_init_request, +) from azure_functions_runtime.otel import (initialize_azure_monitor, update_opentelemetry_status, OTelManager) @@ -381,3 +384,53 @@ async def test_otel_configured_before_first_log_otel_libs( f"before the first logger.info (index {first_log_index}). " f"Call order: {self.call_order}" ) + + @patch("azure_functions_runtime.handle_event" + ".otel_manager.get_azure_monitor_available", return_value=True) + @patch("azure_functions_runtime.handle_event" + ".otel_manager.get_otel_libs_available", return_value=False) + @patch("azure_functions_runtime.handle_event.run_sync_func", + return_value=None) + @patch("azure_functions_runtime.handle_event.get_context") + @patch("azure_functions_runtime.handle_event._functions") + @patch("azure_functions_runtime.handle_event.configure_opentelemetry") + @patch("azure_functions_runtime.handle_event.logger") + async def test_otel_configured_before_first_log_sync_function( + self, + mock_logger, + mock_configure_otel, + mock_functions, + mock_get_context, + mock_run_sync_func, + mock_get_otel_libs, + mock_get_azure_monitor, + ): + """Verify configure_opentelemetry is called before first log (sync function).""" + mock_logger.info.side_effect = self._track_logger_info + mock_configure_otel.side_effect = self._track_configure_otel + mock_fi = self._make_mock_fi() + mock_fi.is_async = False + mock_functions.get_function.return_value = mock_fi + invocation_context = MagicMock() + mock_get_context.return_value = invocation_context + request = self._make_mock_request() + + try: + await invocation_request(request) + except Exception: + pass + + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + mock_configure_otel.assert_called_once_with(invocation_context) + + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) diff --git a/workers/tests/unittests/test_opentelemetry.py b/workers/tests/unittests/test_opentelemetry.py index 31e2e04c5..00443e042 100644 --- a/workers/tests/unittests/test_opentelemetry.py +++ b/workers/tests/unittests/test_opentelemetry.py @@ -8,7 +8,6 @@ from tests.utils import testutils from azure_functions_worker import protos -from azure_functions_worker.dispatcher import ContextEnabledTask class TestOpenTelemetry(unittest.TestCase): @@ -232,7 +231,10 @@ def setUp(self): # Import directly from azure_functions_worker so this test always tests # the correct dispatcher regardless of the Python version (testutils # redirects to proxy_worker on Python >= 3.13). - from azure_functions_worker.dispatcher import Dispatcher, ContextEnabledTask + from azure_functions_worker.dispatcher import ( + ContextEnabledTask, + Dispatcher, + ) self.loop = asyncio.new_event_loop() asyncio.set_event_loop(self.loop) # Use ContextEnabledTask factory so dispatcher's assertion passes @@ -350,7 +352,7 @@ def test_azure_monitor_configured_before_first_log( self, mock_logger, ): - """Verify configure_opentelemetry is called before first log with Azure Monitor.""" + """Verify OTel is configured before the first Azure Monitor log.""" # Set up tracking for call order mock_logger.info.side_effect = self._track_logger_info @@ -423,3 +425,74 @@ async def _noop(): f"configure_opentelemetry should be called before first log. " f"Call order: {self.call_order}" ) + + @patch.dict(os.environ, {'PYTHON_ENABLE_OPENTELEMETRY': 'true'}) + @patch('azure_functions_worker.dispatcher.logger') + def test_otel_configured_before_first_log_sync_function( + self, + mock_logger, + ): + """Verify OTel is configured before the first sync function log.""" + mock_logger.info.side_effect = self._track_logger_info + + self.dispatcher._otel_libs_available = True + self.dispatcher._azure_monitor_available = False + + self.dispatcher.configure_opentelemetry = MagicMock( + side_effect=self._track_configure_otel + ) + + mock_context = MagicMock() + mock_context.trace_context = MagicMock() + mock_context.trace_context.trace_parent = "00-trace-parent" + mock_context.trace_context.trace_state = "state" + mock_context.thread_local_storage = MagicMock() + self.dispatcher._get_context = MagicMock(return_value=mock_context) + + mock_fi = MagicMock() + mock_fi.name = "test_function" + mock_fi.is_async = False + mock_fi.directory = "/test/dir" + mock_fi.input_types = {} + mock_fi.output_types = {} + mock_fi.requires_context = False + mock_fi.has_return = False + mock_fi.settlement_client_arg = None + mock_fi.func = MagicMock(return_value=None) + self.dispatcher._functions = MagicMock() + self.dispatcher._functions.get_function = MagicMock(return_value=mock_fi) + + invoc_request = protos.StreamingMessage( + invocation_request=protos.InvocationRequest( + invocation_id="test-inv-sync", + function_id="test-func-id", + trace_context=protos.RpcTraceContext( + trace_parent="00-trace-parent", + trace_state="state" + ) + ) + ) + + self.dispatcher._run_sync_func = MagicMock(return_value=None) + + try: + self.loop.run_until_complete( + self.dispatcher._handle__invocation_request(invoc_request)) + except Exception: + pass + + self.assertIn('configure_opentelemetry', self.call_order, + "configure_opentelemetry should have been called") + self.assertIn('logger_info', self.call_order, + "logger.info should have been called") + self.dispatcher.configure_opentelemetry.assert_called_once_with( + mock_context) + + otel_index = self.call_order.index('configure_opentelemetry') + first_log_index = self.call_order.index('logger_info') + self.assertLess( + otel_index, first_log_index, + f"configure_opentelemetry (index {otel_index}) should be called " + f"before the first logger.info (index {first_log_index}). " + f"Call order: {self.call_order}" + ) From f8e66a77292bde77e15aa7bf00f1b387dc3338cb Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Mon, 10 Aug 2026 15:58:46 -0500 Subject: [PATCH 4/8] remove double call for sync functions --- runtimes/v1/azure_functions_runtime_v1/utils/executor.py | 2 -- runtimes/v2/azure_functions_runtime/utils/executor.py | 3 --- workers/azure_functions_worker/dispatcher.py | 2 -- 3 files changed, 7 deletions(-) diff --git a/runtimes/v1/azure_functions_runtime_v1/utils/executor.py b/runtimes/v1/azure_functions_runtime_v1/utils/executor.py index 12dd04e40..96842b587 100644 --- a/runtimes/v1/azure_functions_runtime_v1/utils/executor.py +++ b/runtimes/v1/azure_functions_runtime_v1/utils/executor.py @@ -31,8 +31,6 @@ def run_sync_func(invocation_id, context, func, params): context.thread_local_storage.invocation_id = invocation_id token = invocation_id_cv.set(invocation_id) try: - if otel_manager.get_azure_monitor_available(): - configure_opentelemetry(context) result = functools.partial(execute_sync, func) return result(params) finally: diff --git a/runtimes/v2/azure_functions_runtime/utils/executor.py b/runtimes/v2/azure_functions_runtime/utils/executor.py index 718647ae6..81d86a465 100644 --- a/runtimes/v2/azure_functions_runtime/utils/executor.py +++ b/runtimes/v2/azure_functions_runtime/utils/executor.py @@ -30,9 +30,6 @@ def run_sync_func(invocation_id, context, func, params): context.thread_local_storage.invocation_id = invocation_id token = invocation_id_cv.set(invocation_id) try: - if (otel_manager.get_azure_monitor_available() - or otel_manager.get_otel_libs_available()): - configure_opentelemetry(context) result = functools.partial(execute_sync, func) return result(params) finally: diff --git a/workers/azure_functions_worker/dispatcher.py b/workers/azure_functions_worker/dispatcher.py index 2ae77ccba..cd138a0e5 100644 --- a/workers/azure_functions_worker/dispatcher.py +++ b/workers/azure_functions_worker/dispatcher.py @@ -1015,8 +1015,6 @@ 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._azure_monitor_available or self._otel_libs_available: - self.configure_opentelemetry(context) return ExtensionManager.get_sync_invocation_wrapper(context, func)(params) finally: From 5e68f0aa9e7eff48bdc3740fe0f7ae2e5ff19ba5 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Tue, 11 Aug 2026 10:51:31 -0500 Subject: [PATCH 5/8] lint --- runtimes/v1/azure_functions_runtime_v1/utils/executor.py | 2 -- runtimes/v2/azure_functions_runtime/utils/executor.py | 2 -- 2 files changed, 4 deletions(-) diff --git a/runtimes/v1/azure_functions_runtime_v1/utils/executor.py b/runtimes/v1/azure_functions_runtime_v1/utils/executor.py index 96842b587..15675327f 100644 --- a/runtimes/v1/azure_functions_runtime_v1/utils/executor.py +++ b/runtimes/v1/azure_functions_runtime_v1/utils/executor.py @@ -7,8 +7,6 @@ from typing import Any -from ..otel import otel_manager, configure_opentelemetry - def get_current_loop(): return asyncio.events.get_event_loop() diff --git a/runtimes/v2/azure_functions_runtime/utils/executor.py b/runtimes/v2/azure_functions_runtime/utils/executor.py index 81d86a465..35da603da 100644 --- a/runtimes/v2/azure_functions_runtime/utils/executor.py +++ b/runtimes/v2/azure_functions_runtime/utils/executor.py @@ -6,8 +6,6 @@ from typing import Any -from ..otel import otel_manager, configure_opentelemetry - def get_current_loop(): return asyncio.events.get_event_loop() From 98e8cbebd0adcf96d65edd68e2f23d3ab51b8402 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Fri, 14 Aug 2026 09:56:06 -0500 Subject: [PATCH 6/8] remove unnecessary imports --- workers/azure_functions_worker/protos/__init__.py | 1 - workers/proxy_worker/protos/__init__.py | 1 - 2 files changed, 2 deletions(-) diff --git a/workers/azure_functions_worker/protos/__init__.py b/workers/azure_functions_worker/protos/__init__.py index 8123fb2cc..e9c4f2397 100644 --- a/workers/azure_functions_worker/protos/__init__.py +++ b/workers/azure_functions_worker/protos/__init__.py @@ -28,7 +28,6 @@ RpcLog, RpcSharedMemory, RpcDataType, - RpcTraceContext, CloseSharedMemoryResourcesRequest, CloseSharedMemoryResourcesResponse, FunctionsMetadataRequest, diff --git a/workers/proxy_worker/protos/__init__.py b/workers/proxy_worker/protos/__init__.py index 8123fb2cc..e9c4f2397 100644 --- a/workers/proxy_worker/protos/__init__.py +++ b/workers/proxy_worker/protos/__init__.py @@ -28,7 +28,6 @@ RpcLog, RpcSharedMemory, RpcDataType, - RpcTraceContext, CloseSharedMemoryResourcesRequest, CloseSharedMemoryResourcesResponse, FunctionsMetadataRequest, From b08148d7587290c979bb5dec088388c361f23b89 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Fri, 14 Aug 2026 10:54:04 -0500 Subject: [PATCH 7/8] remove extra call --- runtimes/v1/azure_functions_runtime_v1/utils/executor.py | 1 - runtimes/v2/azure_functions_runtime/utils/executor.py | 1 - workers/azure_functions_worker/dispatcher.py | 1 - workers/tests/unittests/test_opentelemetry.py | 7 ++++--- 4 files changed, 4 insertions(+), 6 deletions(-) diff --git a/runtimes/v1/azure_functions_runtime_v1/utils/executor.py b/runtimes/v1/azure_functions_runtime_v1/utils/executor.py index 15675327f..741c257a2 100644 --- a/runtimes/v1/azure_functions_runtime_v1/utils/executor.py +++ b/runtimes/v1/azure_functions_runtime_v1/utils/executor.py @@ -26,7 +26,6 @@ def execute_sync(function, args) -> Any: def run_sync_func(invocation_id, context, func, params): # This helper exists because we need to access the current # invocation_id from ThreadPoolExecutor's threads. - context.thread_local_storage.invocation_id = invocation_id token = invocation_id_cv.set(invocation_id) try: result = functools.partial(execute_sync, func) diff --git a/runtimes/v2/azure_functions_runtime/utils/executor.py b/runtimes/v2/azure_functions_runtime/utils/executor.py index 35da603da..bcf15ca16 100644 --- a/runtimes/v2/azure_functions_runtime/utils/executor.py +++ b/runtimes/v2/azure_functions_runtime/utils/executor.py @@ -25,7 +25,6 @@ def execute_sync(function, args) -> Any: def run_sync_func(invocation_id, context, func, params): # This helper exists because we need to access the current # invocation_id from ThreadPoolExecutor's threads. - context.thread_local_storage.invocation_id = invocation_id token = invocation_id_cv.set(invocation_id) try: result = functools.partial(execute_sync, func) diff --git a/workers/azure_functions_worker/dispatcher.py b/workers/azure_functions_worker/dispatcher.py index cd138a0e5..3c9cd8503 100644 --- a/workers/azure_functions_worker/dispatcher.py +++ b/workers/azure_functions_worker/dispatcher.py @@ -1013,7 +1013,6 @@ def _create_sync_call_tp( def _run_sync_func(self, invocation_id, context, func, params): # This helper exists because we need to access the current # invocation_id from ThreadPoolExecutor's threads. - context.thread_local_storage.invocation_id = invocation_id try: return ExtensionManager.get_sync_invocation_wrapper(context, func)(params) diff --git a/workers/tests/unittests/test_opentelemetry.py b/workers/tests/unittests/test_opentelemetry.py index 00443e042..35ab1746d 100644 --- a/workers/tests/unittests/test_opentelemetry.py +++ b/workers/tests/unittests/test_opentelemetry.py @@ -8,6 +8,7 @@ from tests.utils import testutils from azure_functions_worker import protos +from azure_functions_worker.protos.FunctionRpc_pb2 import RpcTraceContext class TestOpenTelemetry(unittest.TestCase): @@ -309,7 +310,7 @@ def test_otel_configured_before_first_log_in_invocation_request( invocation_request=protos.InvocationRequest( invocation_id="test-inv-123", function_id="test-func-id", - trace_context=protos.RpcTraceContext( + trace_context=RpcTraceContext( trace_parent="00-trace-parent", trace_state="state" ) @@ -392,7 +393,7 @@ def test_azure_monitor_configured_before_first_log( invocation_request=protos.InvocationRequest( invocation_id="test-inv-456", function_id="test-func-id", - trace_context=protos.RpcTraceContext( + trace_context=RpcTraceContext( trace_parent="00-trace-parent", trace_state="state" ) @@ -466,7 +467,7 @@ def test_otel_configured_before_first_log_sync_function( invocation_request=protos.InvocationRequest( invocation_id="test-inv-sync", function_id="test-func-id", - trace_context=protos.RpcTraceContext( + trace_context=RpcTraceContext( trace_parent="00-trace-parent", trace_state="state" ) From 9bba7641f479d41da85e0f389e3a3e594cdd96c6 Mon Sep 17 00:00:00 2001 From: Victoria Hall Date: Fri, 14 Aug 2026 11:33:46 -0500 Subject: [PATCH 8/8] restore --- .../handle_event.py | 1 - .../utils/executor.py | 1 + runtimes/v1/tests/unittests/test_executor.py | 57 +++++++++++++++++++ .../azure_functions_runtime/utils/executor.py | 1 + runtimes/v2/tests/unittests/test_executor.py | 57 +++++++++++++++++++ workers/azure_functions_worker/dispatcher.py | 2 +- workers/tests/unittests/test_dispatcher.py | 53 ++++++++++++++++- 7 files changed, 169 insertions(+), 3 deletions(-) create mode 100644 runtimes/v1/tests/unittests/test_executor.py create mode 100644 runtimes/v2/tests/unittests/test_executor.py diff --git a/runtimes/v1/azure_functions_runtime_v1/handle_event.py b/runtimes/v1/azure_functions_runtime_v1/handle_event.py index 66a245cdc..f0d9e0d61 100644 --- a/runtimes/v1/azure_functions_runtime_v1/handle_event.py +++ b/runtimes/v1/azure_functions_runtime_v1/handle_event.py @@ -235,7 +235,6 @@ async def invocation_request(request): # Actively flush customer print() function to console sys.stdout.flush() - logger.debug("Successfully completed WorkerInvocationRequest.") return protos.InvocationResponse( invocation_id=invocation_id, diff --git a/runtimes/v1/azure_functions_runtime_v1/utils/executor.py b/runtimes/v1/azure_functions_runtime_v1/utils/executor.py index 741c257a2..15675327f 100644 --- a/runtimes/v1/azure_functions_runtime_v1/utils/executor.py +++ b/runtimes/v1/azure_functions_runtime_v1/utils/executor.py @@ -26,6 +26,7 @@ def execute_sync(function, args) -> Any: def run_sync_func(invocation_id, context, func, params): # This helper exists because we need to access the current # invocation_id from ThreadPoolExecutor's threads. + context.thread_local_storage.invocation_id = invocation_id token = invocation_id_cv.set(invocation_id) try: result = functools.partial(execute_sync, func) diff --git a/runtimes/v1/tests/unittests/test_executor.py b/runtimes/v1/tests/unittests/test_executor.py new file mode 100644 index 000000000..1acbc2bea --- /dev/null +++ b/runtimes/v1/tests/unittests/test_executor.py @@ -0,0 +1,57 @@ +import concurrent.futures +import threading +import unittest +from unittest.mock import MagicMock + +from azure_functions_runtime_v1.utils.executor import ( + invocation_id_cv, + run_sync_func, +) + + +class TestSyncInvocationContext(unittest.TestCase): + + def setUp(self): + self.context = MagicMock() + self.context.thread_local_storage = threading.local() + self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) + + def tearDown(self): + self.executor.shutdown() + + def _get_current_invocation_ids(self): + return (self.context.thread_local_storage.invocation_id, + invocation_id_cv.get()) + + def test_invocation_id_cleared_after_success(self): + observed_invocation_ids = [] + + def invoke(): + observed_invocation_ids.append(self._get_current_invocation_ids()) + + self.executor.submit( + run_sync_func, + 'test-invocation', self.context, invoke, {}).result() + + remaining_invocation_ids = self.executor.submit( + self._get_current_invocation_ids).result() + self.assertEqual( + observed_invocation_ids, + [('test-invocation', 'test-invocation')]) + self.assertEqual(remaining_invocation_ids, (None, None)) + + def test_invocation_id_cleared_after_exception(self): + def invoke(): + self.assertEqual( + self._get_current_invocation_ids(), + ('test-invocation', 'test-invocation')) + raise RuntimeError('test error') + + with self.assertRaisesRegex(RuntimeError, 'test error'): + self.executor.submit( + run_sync_func, + 'test-invocation', self.context, invoke, {}).result() + + remaining_invocation_ids = self.executor.submit( + self._get_current_invocation_ids).result() + self.assertEqual(remaining_invocation_ids, (None, None)) diff --git a/runtimes/v2/azure_functions_runtime/utils/executor.py b/runtimes/v2/azure_functions_runtime/utils/executor.py index bcf15ca16..35da603da 100644 --- a/runtimes/v2/azure_functions_runtime/utils/executor.py +++ b/runtimes/v2/azure_functions_runtime/utils/executor.py @@ -25,6 +25,7 @@ def execute_sync(function, args) -> Any: def run_sync_func(invocation_id, context, func, params): # This helper exists because we need to access the current # invocation_id from ThreadPoolExecutor's threads. + context.thread_local_storage.invocation_id = invocation_id token = invocation_id_cv.set(invocation_id) try: result = functools.partial(execute_sync, func) diff --git a/runtimes/v2/tests/unittests/test_executor.py b/runtimes/v2/tests/unittests/test_executor.py new file mode 100644 index 000000000..840d42784 --- /dev/null +++ b/runtimes/v2/tests/unittests/test_executor.py @@ -0,0 +1,57 @@ +import concurrent.futures +import threading +import unittest +from unittest.mock import MagicMock + +from azure_functions_runtime.utils.executor import ( + invocation_id_cv, + run_sync_func, +) + + +class TestSyncInvocationContext(unittest.TestCase): + + def setUp(self): + self.context = MagicMock() + self.context.thread_local_storage = threading.local() + self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) + + def tearDown(self): + self.executor.shutdown() + + def _get_current_invocation_ids(self): + return (self.context.thread_local_storage.invocation_id, + invocation_id_cv.get()) + + def test_invocation_id_cleared_after_success(self): + observed_invocation_ids = [] + + def invoke(): + observed_invocation_ids.append(self._get_current_invocation_ids()) + + self.executor.submit( + run_sync_func, + 'test-invocation', self.context, invoke, {}).result() + + remaining_invocation_ids = self.executor.submit( + self._get_current_invocation_ids).result() + self.assertEqual( + observed_invocation_ids, + [('test-invocation', 'test-invocation')]) + self.assertEqual(remaining_invocation_ids, (None, None)) + + def test_invocation_id_cleared_after_exception(self): + def invoke(): + self.assertEqual( + self._get_current_invocation_ids(), + ('test-invocation', 'test-invocation')) + raise RuntimeError('test error') + + with self.assertRaisesRegex(RuntimeError, 'test error'): + self.executor.submit( + run_sync_func, + 'test-invocation', self.context, invoke, {}).result() + + remaining_invocation_ids = self.executor.submit( + self._get_current_invocation_ids).result() + self.assertEqual(remaining_invocation_ids, (None, None)) diff --git a/workers/azure_functions_worker/dispatcher.py b/workers/azure_functions_worker/dispatcher.py index 3c9cd8503..b472bab50 100644 --- a/workers/azure_functions_worker/dispatcher.py +++ b/workers/azure_functions_worker/dispatcher.py @@ -723,7 +723,6 @@ async def _handle__invocation_request(self, request): # Actively flush customer print() function to console sys.stdout.flush() - return protos.StreamingMessage( request_id=self.request_id, invocation_response=protos.InvocationResponse( @@ -1013,6 +1012,7 @@ def _create_sync_call_tp( def _run_sync_func(self, invocation_id, context, func, params): # This helper exists because we need to access the current # invocation_id from ThreadPoolExecutor's threads. + context.thread_local_storage.invocation_id = invocation_id try: return ExtensionManager.get_sync_invocation_wrapper(context, func)(params) diff --git a/workers/tests/unittests/test_dispatcher.py b/workers/tests/unittests/test_dispatcher.py index 8b140a00e..d501f3204 100644 --- a/workers/tests/unittests/test_dispatcher.py +++ b/workers/tests/unittests/test_dispatcher.py @@ -2,12 +2,14 @@ # Licensed under the MIT License. import asyncio import collections as col +import concurrent.futures import contextvars import os import sys +import threading import unittest from typing import Optional, Tuple -from unittest.mock import patch +from unittest.mock import MagicMock, patch from tests.utils import testutils from tests.utils.testutils import UNIT_TESTS_ROOT @@ -631,6 +633,55 @@ async def test_dispatcher_functions_metadata_request_with_retry(self): del sys.modules['function_app'] +class TestSyncInvocationContext(unittest.TestCase): + + def setUp(self): + self.dispatcher = Dispatcher.__new__(Dispatcher) + self.context = MagicMock() + self.context.thread_local_storage = threading.local() + self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=1) + + def tearDown(self): + self.executor.shutdown() + + @patch('azure_functions_worker.dispatcher.ExtensionManager' + '.get_sync_invocation_wrapper') + def test_invocation_id_cleared_after_success(self, mock_get_wrapper): + observed_invocation_ids = [] + mock_get_wrapper.return_value = lambda params: ( + observed_invocation_ids.append( + self.context.thread_local_storage.invocation_id)) + + self.executor.submit( + self.dispatcher._run_sync_func, + 'test-invocation', self.context, MagicMock(), {}).result() + + remaining_invocation_id = self.executor.submit( + lambda: self.context.thread_local_storage.invocation_id).result() + self.assertEqual(observed_invocation_ids, ['test-invocation']) + self.assertIsNone(remaining_invocation_id) + + @patch('azure_functions_worker.dispatcher.ExtensionManager' + '.get_sync_invocation_wrapper') + def test_invocation_id_cleared_after_exception(self, mock_get_wrapper): + def invoke(params): + self.assertEqual( + self.context.thread_local_storage.invocation_id, + 'test-invocation') + raise RuntimeError('test error') + + mock_get_wrapper.return_value = invoke + + with self.assertRaisesRegex(RuntimeError, 'test error'): + self.executor.submit( + self.dispatcher._run_sync_func, + 'test-invocation', self.context, MagicMock(), {}).result() + + remaining_invocation_id = self.executor.submit( + lambda: self.context.thread_local_storage.invocation_id).result() + self.assertIsNone(remaining_invocation_id) + + class TestDispatcherSteinLegacyFallback(testutils.AsyncTestCase): def setUp(self):