feat: Http V2 Support (#1455)

* Http Proxy Support

Http Proxy Support

integration

refactor

update ext name

add to env reload

add dep

revert

* final changes and tests

* fix ppl

* revert

* fix

* fix

* fix

* fix

* flake8

* flake8

* flake8

* only run for 3.8+

* FIX

* skip tests for 3.7-

* fix

* fix

* fix

* fix

* fix styles

* fix

* s

* fix

* fix

* revert

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* tests

* fix

* test

* tests

* fix

* fix

* fix

* move init server

* fix

* test

* update ppls

* fix test

* fix tests

* fix

* fix

* fix

* Revert "fix"

This reverts commit bc9ca7f.

* skip

* fix tests

* style

* test

* test

* ff

* test

* style

* skip

* test

* fix

* fix

* fix

* replay

* f

* fix

* codecov fix

* Mount base extension to fix consumption test failures

* style

* revert

* revert

* revert ppls

* test

* Revert "test"

This reverts commit 4142767.

* address feedback

* revert ut ppl

* skip checking has http func

* fix test

* style

* revert cache ppl

* revert ppl

* pip fix

* try fix 3.7 ppl

* try

* oh meta

* fix

* fix

* fix

* dont pin fastapi to fix cmake err

* fix

* fix

* update azfunc base

* update

* feedback

* feedback

* feedback

* add docs

* merge fix

* feedback

* feedback

* feedback

* feedback

---------

Co-authored-by: hallvictoria <victoriahall@microsoft.com>

Author: YunchuWang

.github/workflows/ci_consumption_workflow.yml

@@ -35,6 +35,9 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple -U azure-functions --pre
           python -m pip install -U -e .[dev]
+          if [[ "${{ matrix.python-version }}" != "3.7" ]]; then
+             python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple --pre -U -e .[test-http-v2]
+          fi
           python setup.py build
       - name: Running 3.7 Tests
         if: matrix.python-version == 3.7

.github/workflows/ci_e2e_workflow.yml

@@ -61,6 +61,10 @@ jobs:
           python -m pip install --upgrade pip
           python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple -U azure-functions --pre
           python -m pip install -U -e .[dev]
+ 
+          if [[ "${{ matrix.python-version }}" != "3.7" ]]; then
+             python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple --pre -U -e .[test-http-v2]
+          fi
           if [[ "${{ matrix.python-version }}" != "3.7" && "${{ matrix.python-version }}" != "3.8" ]]; then
               python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple --pre -U -e .[test-deferred-bindings]
           fi

.github/workflows/ci_ut_workflow.yml

@@ -58,7 +58,11 @@ jobs:
 
         python -m pip install --upgrade pip
         python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple -U azure-functions --pre
-        python -m pip install -U -e .[dev]
+
+        python -m pip install -U -e .[dev] 
+        if [[ "${{ matrix.python-version }}" != "3.7" ]]; then
+           python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple --pre -U -e .[test-http-v2]
+        fi
 
         # Retry a couple times to avoid certificate issue
         retry 5 python setup.py build

azure_functions_worker/bindings/meta.py

@@ -4,26 +4,54 @@
 import sys
 import typing
 
-from .. import protos
 
+from .. import protos
 from . import datumdef
 from . import generic
 
 from .shared_memory_data_transfer import SharedMemoryManager
-from ..constants import CUSTOMER_PACKAGES_PATH
+from ..http_v2 import HttpV2Registry
+from ..constants import CUSTOMER_PACKAGES_PATH, HTTP, HTTP_TRIGGER, \
+    BASE_EXT_SUPPORTED_PY_MINOR_VERSION
 from ..logging import logger
 
+
 PB_TYPE = 'rpc_data'
 PB_TYPE_DATA = 'data'
 PB_TYPE_RPC_SHARED_MEMORY = 'rpc_shared_memory'
-# Base extension supported Python minor version
-BASE_EXT_SUPPORTED_PY_MINOR_VERSION = 8
 
 BINDING_REGISTRY = None
 DEFERRED_BINDING_REGISTRY = None
 deferred_bindings_cache = {}
 
 
+def _check_http_input_type_annotation(bind_name: str, pytype: type,
+                                      is_deferred_binding: bool) -> bool:
+    if HttpV2Registry.http_v2_enabled():
+        return HttpV2Registry.ext_base().RequestTrackerMeta \
+            .check_type(pytype)
+
+    binding = get_binding(bind_name, is_deferred_binding)
+    return binding.check_input_type_annotation(pytype)
+
+
+def _check_http_output_type_annotation(bind_name: str, pytype: type) -> bool:
+    if HttpV2Registry.http_v2_enabled():
+        return HttpV2Registry.ext_base().ResponseTrackerMeta.check_type(pytype)
+
+    binding = get_binding(bind_name)
+    return binding.check_output_type_annotation(pytype)
+
+
+INPUT_TYPE_CHECK_OVERRIDE_MAP = {
+    HTTP_TRIGGER: _check_http_input_type_annotation
+}
+
+OUTPUT_TYPE_CHECK_OVERRIDE_MAP = {
+    HTTP: _check_http_output_type_annotation
+}
+
+
 def load_binding_registry() -> None:
     """
     Tries to load azure-functions from the customer's BYO. If it's
@@ -89,11 +117,21 @@ def is_trigger_binding(bind_name: str) -> bool:
 def check_input_type_annotation(bind_name: str,
                                 pytype: type,
                                 is_deferred_binding: bool) -> bool:
+    global INPUT_TYPE_CHECK_OVERRIDE_MAP
+    if bind_name in INPUT_TYPE_CHECK_OVERRIDE_MAP:
+        return INPUT_TYPE_CHECK_OVERRIDE_MAP[bind_name](bind_name, pytype,
+                                                        is_deferred_binding)
+
     binding = get_binding(bind_name, is_deferred_binding)
+
     return binding.check_input_type_annotation(pytype)
 
 
 def check_output_type_annotation(bind_name: str, pytype: type) -> bool:
+    global OUTPUT_TYPE_CHECK_OVERRIDE_MAP
+    if bind_name in OUTPUT_TYPE_CHECK_OVERRIDE_MAP:
+        return OUTPUT_TYPE_CHECK_OVERRIDE_MAP[bind_name](bind_name, pytype)
+
     binding = get_binding(bind_name)
     return binding.check_output_type_annotation(pytype)
 

azure_functions_worker/constants.py

@@ -10,6 +10,7 @@
 WORKER_STATUS = "WorkerStatus"
 SHARED_MEMORY_DATA_TRANSFER = "SharedMemoryDataTransfer"
 FUNCTION_DATA_CACHE = "FunctionDataCache"
+HTTP_URI = "HttpUri"
 
 # Platform Environment Variables
 AZURE_WEBJOBS_SCRIPT_ROOT = "AzureWebJobsScriptRoot"
@@ -54,9 +55,22 @@
 RETRY_POLICY = "retry_policy"
 
 # Paths
-CUSTOMER_PACKAGES_PATH = "/home/site/wwwroot/.python_packages/lib/site-packages"
+CUSTOMER_PACKAGES_PATH = "/home/site/wwwroot/.python_packages/lib/site" \
+                         "-packages"
 
 # Flag to index functions in handle init request
 PYTHON_ENABLE_INIT_INDEXING = "PYTHON_ENABLE_INIT_INDEXING"
 
 METADATA_PROPERTIES_WORKER_INDEXED = "worker_indexed"
+
+# Header names
+X_MS_INVOCATION_ID = "x-ms-invocation-id"
+
+# Trigger Names
+HTTP_TRIGGER = "httpTrigger"
+
+# Output Names
+HTTP = "http"
+
+# Base extension supported Python minor version
+BASE_EXT_SUPPORTED_PY_MINOR_VERSION = 8

azure_functions_worker/dispatcher.py

@@ -19,7 +19,6 @@
 from datetime import datetime
 
 import grpc
-
 from . import bindings, constants, functions, loader, protos
 from .bindings.shared_memory_data_transfer import SharedMemoryManager
 from .constants import (PYTHON_ROLLBACK_CWD_PATH,
@@ -33,6 +32,8 @@
                         PYTHON_LANGUAGE_RUNTIME, PYTHON_ENABLE_INIT_INDEXING,
                         METADATA_PROPERTIES_WORKER_INDEXED)
 from .extension import ExtensionManager
+from .http_v2 import http_coordinator, initialize_http_server, HttpV2Registry, \
+    sync_http_request, HttpServerInitError
 from .logging import disable_console_logging, enable_console_logging
 from .logging import (logger, error_logger, is_system_log_category,
                       CONSOLE_LOG_PREFIX, format_exception)
@@ -158,6 +159,7 @@ async def dispatch_forever(self):  # sourcery skip: swap-if-expression
 
             log_level = logging.INFO if not is_envvar_true(
                 PYTHON_ENABLE_DEBUG_LOGGING) else logging.DEBUG
+
             root_logger.setLevel(log_level)
             root_logger.addHandler(logging_handler)
             logger.info('Switched to gRPC logging.')
@@ -189,7 +191,8 @@ def stop(self) -> None:
 
         self._stop_sync_call_tp()
 
-    def on_logging(self, record: logging.LogRecord, formatted_msg: str) -> None:
+    def on_logging(self, record: logging.LogRecord,
+                   formatted_msg: str) -> None:
         if record.levelno >= logging.CRITICAL:
             log_level = protos.RpcLog.Critical
         elif record.levelno >= logging.ERROR:
@@ -306,6 +309,13 @@ async def _handle__worker_init_request(self, request):
                 self.load_function_metadata(
                     worker_init_request.function_app_directory,
                     caller_info="worker_init_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
 
@@ -508,6 +518,7 @@ async def _handle__invocation_request(self, request):
             logger.info(', '.join(function_invocation_logs))
 
             args = {}
+
             for pb in invoc_request.input_data:
                 pb_type_info = fi.input_types[pb.name]
                 if bindings.is_trigger_binding(pb_type_info.binding_name):
@@ -523,7 +534,19 @@ async def _handle__invocation_request(self, request):
                     shmem_mgr=self._shmem_mgr,
                     is_deferred_binding=pb_type_info.deferred_bindings_enabled)
 
-            fi_context = self._get_context(invoc_request, fi.name, fi.directory)
+            http_v2_enabled = self._functions.get_function(function_id) \
+                                  .is_http_func and \
+                HttpV2Registry.http_v2_enabled()
+
+            if http_v2_enabled:
+                http_request = await http_coordinator.get_http_request_async(
+                    invocation_id)
+
+                await sync_http_request(http_request, invoc_request)
+                args[fi.trigger_metadata.get('param_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
@@ -536,17 +559,21 @@ async def _handle__invocation_request(self, request):
                     args[name] = bindings.Out()
 
             if fi.is_async:
-                call_result = await self._run_async_func(
-                    fi_context, fi.func, args
-                )
+                call_result = \
+                    await self._run_async_func(fi_context, fi.func, args)
             else:
                 call_result = await self._loop.run_in_executor(
                     self._sync_call_tp,
                     self._run_sync_func,
                     invocation_id, fi_context, fi.func, args)
+
             if call_result is not None and not fi.has_return:
-                raise RuntimeError(f'function {fi.name!r} without a $return '
-                                   'binding returned a non-None value')
+                raise RuntimeError(
+                    f'function {fi.name!r} without a $return binding'
+                    'returned a non-None value')
+
+            if http_v2_enabled:
+                http_coordinator.set_http_response(invocation_id, call_result)
 
             output_data = []
             cache_enabled = self._function_data_cache_enabled
@@ -566,10 +593,12 @@ async def _handle__invocation_request(self, request):
                     output_data.append(param_binding)
 
             return_value = None
-            if fi.return_type is not None:
+            if fi.return_type is not None and not http_v2_enabled:
                 return_value = bindings.to_outgoing_proto(
-                    fi.return_type.binding_name, call_result,
-                    pytype=fi.return_type.pytype)
+                    fi.return_type.binding_name,
+                    call_result,
+                    pytype=fi.return_type.pytype,
+                )
 
             # Actively flush customer print() function to console
             sys.stdout.flush()
@@ -584,6 +613,9 @@ async def _handle__invocation_request(self, request):
                     output_data=output_data))
 
         except Exception as ex:
+            if http_v2_enabled:
+                http_coordinator.set_http_response(invocation_id, ex)
+
             return protos.StreamingMessage(
                 request_id=self.request_id,
                 invocation_response=protos.InvocationResponse(
@@ -640,11 +672,18 @@ async def _handle__function_environment_reload_request(self, request):
             # reload_customer_libraries call clears the registry
             bindings.load_binding_registry()
 
+            capabilities = {}
             if is_envvar_true(PYTHON_ENABLE_INIT_INDEXING):
                 try:
                     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
 
@@ -655,7 +694,7 @@ async def _handle__function_environment_reload_request(self, request):
                     func_env_reload_request.function_app_directory)
 
             success_response = protos.FunctionEnvironmentReloadResponse(
-                capabilities={},
+                capabilities=capabilities,
                 worker_metadata=self.get_worker_metadata(),
                 result=protos.StatusResult(
                     status=protos.StatusResult.Success))
@@ -676,8 +715,10 @@ async def _handle__function_environment_reload_request(self, request):
 
     def index_functions(self, function_path: str):
         indexed_functions = loader.index_function_app(function_path)
-        logger.info('Indexed function app and found %s functions',
-                    len(indexed_functions))
+        logger.info(
+            "Indexed function app and found %s functions",
+            len(indexed_functions)
+        )
 
         if indexed_functions:
             fx_metadata_results, fx_bindings_logs = (
@@ -747,7 +788,8 @@ async def _handle__close_shared_memory_resources_request(self, request):
     @staticmethod
     def _get_context(invoc_request: protos.InvocationRequest, name: str,
                      directory: str) -> bindings.Context:
-        """ For more information refer: https://aka.ms/azfunc-invocation-context
+        """ For more information refer:
+        https://aka.ms/azfunc-invocation-context
         """
         trace_context = bindings.TraceContext(
             invoc_request.trace_context.trace_parent,
@@ -889,7 +931,6 @@ def gen(resp_queue):
 
 
 class AsyncLoggingHandler(logging.Handler):
-
     def emit(self, record: LogRecord) -> None:
         # Since we disable console log after gRPC channel is initiated,
         # we should redirect all the messages into dispatcher.