feedback

YunchuWang · YunchuWang · commit a695ba3fce92 · 2024-04-11T11:29:55.000-07:00
diff --git a/azure_functions_worker/dispatcher.py b/azure_functions_worker/dispatcher.py
@@ -31,6 +31,7 @@
                         PYTHON_SCRIPT_FILE_NAME_DEFAULT,
                         PYTHON_LANGUAGE_RUNTIME, PYTHON_ENABLE_INIT_INDEXING,
                         METADATA_PROPERTIES_WORKER_INDEXED)
+from .exceptions import HttpServerInitError
 from .extension import ExtensionManager
 from .http_v2 import http_coordinator, initialize_http_server, HttpV2Registry, \
     sync_http_request
@@ -112,8 +113,7 @@ def get_sync_tp_workers_set(self):
          3.9 scenarios (as we'll start passing only None by default), and we
          need to get that information.
 
-         Ref: concurrent.futures.thread.ThreadPoolExecutor.__init__
-         ._max_workers
+         Ref: concurrent.futures.thread.ThreadPoolExecutor.__init__._max_workers
         """
         return self._sync_call_tp._max_workers
 
@@ -310,32 +310,23 @@ async def _handle__worker_init_request(self, request):
                 self.load_function_metadata(
                     worker_init_request.function_app_directory,
                     caller_info="worker_init_request")
-            except Exception as ex:
-                self._function_metadata_exception = ex
 
-            try:
                 if HttpV2Registry.http_v2_enabled():
                     capabilities[constants.HTTP_URI] = \
                         initialize_http_server(self._host)
+
+            except HttpServerInitError:
+                raise
             except Exception as ex:
-                return protos.StreamingMessage(
-                    request_id=self.request_id,
-                    worker_init_response=protos.WorkerInitResponse(
-                        result=protos.StatusResult(
-                            status=protos.StatusResult.Failure,
-                            exception=self._serialize_exception(ex))
-                    )
-                )
+                self._function_metadata_exception = ex
 
         return protos.StreamingMessage(
             request_id=self.request_id,
             worker_init_response=protos.WorkerInitResponse(
                 capabilities=capabilities,
                 worker_metadata=self.get_worker_metadata(),
                 result=protos.StatusResult(
-                    status=protos.StatusResult.Success),
-            ),
-        )
+                    status=protos.StatusResult.Success)))
 
     async def _handle__worker_status_request(self, request):
         # Logging is not necessary in this request since the response is used
@@ -686,13 +677,15 @@ async def _handle__function_environment_reload_request(self, request):
                     self.load_function_metadata(
                         directory,
                         caller_info="environment_reload_request")
+
+                    if HttpV2Registry.http_v2_enabled():
+                        capabilities[constants.HTTP_URI] = \
+                            initialize_http_server(self._host)
+                except HttpServerInitError:
+                    raise
                 except Exception as ex:
                     self._function_metadata_exception = ex
 
-                if HttpV2Registry.http_v2_enabled():
-                    capabilities[constants.HTTP_URI] = \
-                        initialize_http_server(self._host)
-
             # Change function app directory
             if getattr(func_env_reload_request,
                        'function_app_directory', None):
diff --git a/azure_functions_worker/exceptions.py b/azure_functions_worker/exceptions.py
@@ -0,0 +1,9 @@
+# http v2 exception types
+class HttpServerInitError(Exception):
+    """Exception raised when there is an error during HTTP server
+    initialization."""
+
+
+class MissingHeaderError(ValueError):
+    """Exception raised when a required header is missing in the
+    HTTP request."""
diff --git a/azure_functions_worker/http_v2.py b/azure_functions_worker/http_v2.py
@@ -7,6 +7,8 @@
 
 from azure_functions_worker.constants import X_MS_INVOCATION_ID, \
     BASE_EXT_SUPPORTED_PY_MINOR_VERSION, PYTHON_ENABLE_INIT_INDEXING
+from azure_functions_worker.exceptions import MissingHeaderError, \
+    HttpServerInitError
 from azure_functions_worker.logging import logger
 from azure_functions_worker.utils.common import is_envvar_false
 
@@ -113,25 +115,24 @@ def set_http_request(self, invoc_id, http_request):
 
     def set_http_response(self, invoc_id, http_response):
         if invoc_id not in self._context_references:
-            raise Exception("No context reference found for invocation "
-                            f"{invoc_id}")
+            raise KeyError("No context reference found for invocation %s"
+                           % invoc_id)
         context_ref = self._context_references.get(invoc_id)
         context_ref.http_response = http_response
 
     async def get_http_request_async(self, invoc_id):
         if invoc_id not in self._context_references:
             self._context_references[invoc_id] = AsyncContextReference()
 
-        await asyncio.sleep(0)
         await self._context_references.get(
             invoc_id).http_request_available_event.wait()
         return self._pop_http_request(invoc_id)
 
     async def await_http_response_async(self, invoc_id):
         if invoc_id not in self._context_references:
-            raise Exception("No context reference found for invocation "
-                            f"{invoc_id}")
-        await asyncio.sleep(0)
+            raise KeyError("No context reference found for invocation %s"
+                           % invoc_id)
+
         await self._context_references.get(
             invoc_id).http_response_available_event.wait()
         return self._pop_http_response(invoc_id)
@@ -143,7 +144,7 @@ def _pop_http_request(self, invoc_id):
             context_ref.http_request = None
             return request
 
-        raise ValueError(f"No http request found for invocation {invoc_id}")
+        raise ValueError("No http request found for invocation %s" % invoc_id)
 
     def _pop_http_response(self, invoc_id):
         context_ref = self._context_references.get(invoc_id)
@@ -152,7 +153,7 @@ def _pop_http_response(self, invoc_id):
             context_ref.http_response = None
             return response
 
-        raise ValueError(f"No http response found for invocation {invoc_id}")
+        raise ValueError("No http response found for invocation %s" % invoc_id)
 
 
 def get_unused_tcp_port():
@@ -169,44 +170,50 @@ def get_unused_tcp_port():
 
 
 def initialize_http_server(host_addr, **kwargs):
-    ext_base = HttpV2Registry.ext_base()
-    web_extension_mod_name = ext_base.ModuleTrackerMeta.get_module()
-    extension_module = importlib.import_module(web_extension_mod_name)
-    web_app_class = extension_module.WebApp
-    web_server_class = extension_module.WebServer
-
-    unused_port = get_unused_tcp_port()
-
-    app = web_app_class()
-    request_type = ext_base.RequestTrackerMeta.get_request_type()
-
-    @app.route
-    async def catch_all(request: request_type):  # type: ignore
-        invoc_id = request.headers.get(X_MS_INVOCATION_ID)
-        if invoc_id is None:
-            raise ValueError(f"Header {X_MS_INVOCATION_ID} not found")
-        logger.info('Received HTTP request for invocation %s', invoc_id)
-        http_coordinator.set_http_request(invoc_id, request)
-        http_resp = \
-            await http_coordinator.await_http_response_async(invoc_id)
-
-        logger.info('Sending HTTP response for invocation %s', invoc_id)
-        # if http_resp is an python exception, raise it
-        if isinstance(http_resp, Exception):
-            raise http_resp
-
-        return http_resp
-
-    web_server = web_server_class(host_addr, unused_port, app)
-    web_server_run_task = web_server.serve()
-
-    loop = asyncio.get_event_loop()
-    loop.create_task(web_server_run_task)
-
-    web_server_address = f"http://{host_addr}:{unused_port}"
-    logger.info('HTTP server starting on %s', web_server_address)
-
-    return web_server_address
+    try:
+        ext_base = HttpV2Registry.ext_base()
+        web_extension_mod_name = ext_base.ModuleTrackerMeta.get_module()
+        extension_module = importlib.import_module(web_extension_mod_name)
+        web_app_class = extension_module.WebApp
+        web_server_class = extension_module.WebServer
+
+        unused_port = get_unused_tcp_port()
+
+        app = web_app_class()
+        request_type = ext_base.RequestTrackerMeta.get_request_type()
+
+        @app.route
+        async def catch_all(request: request_type):  # type: ignore
+            invoc_id = request.headers.get(X_MS_INVOCATION_ID)
+            if invoc_id is None:
+                raise MissingHeaderError("Header %s not found" %
+                                         X_MS_INVOCATION_ID)
+            logger.info('Received HTTP request for invocation %s', invoc_id)
+            http_coordinator.set_http_request(invoc_id, request)
+            http_resp = \
+                await http_coordinator.await_http_response_async(invoc_id)
+
+            logger.info('Sending HTTP response for invocation %s', invoc_id)
+            # if http_resp is an python exception, raise it
+            if isinstance(http_resp, Exception):
+                raise http_resp
+
+            return http_resp
+
+        web_server = web_server_class(host_addr, unused_port, app)
+        web_server_run_task = web_server.serve()
+
+        loop = asyncio.get_event_loop()
+        loop.create_task(web_server_run_task)
+
+        web_server_address = f"http://{host_addr}:{unused_port}"
+        logger.info('HTTP server starting on %s', web_server_address)
+
+        return web_server_address
+
+    except Exception as e:
+        raise HttpServerInitError("Error initializing HTTP server: %s" % e) \
+            from e
 
 
 async def sync_http_request(http_request, invoc_request):
