Skip to content

Commit 2c77ac5

Browse files
YunchuWanghallvictoria
YunchuWang
and
hallvictoria
authored
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 <[email protected]>
1 parent 9709e08 commit 2c77ac5

File tree

17 files changed

+1920
-60
lines changed

17 files changed

+1920
-60
lines changed

‎.github/workflows/ci_consumption_workflow.yml

+3
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,9 @@ jobs:
3535
python -m pip install --upgrade pip
3636
python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple -U azure-functions --pre
3737
python -m pip install -U -e .[dev]
38+
if [[ "${{ matrix.python-version }}" != "3.7" ]]; then
39+
python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple --pre -U -e .[test-http-v2]
40+
fi
3841
python setup.py build
3942
- name: Running 3.7 Tests
4043
if: matrix.python-version == 3.7

‎.github/workflows/ci_e2e_workflow.yml

+4
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,10 @@ jobs:
6161
python -m pip install --upgrade pip
6262
python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple -U azure-functions --pre
6363
python -m pip install -U -e .[dev]
64+
65+
if [[ "${{ matrix.python-version }}" != "3.7" ]]; then
66+
python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple --pre -U -e .[test-http-v2]
67+
fi
6468
if [[ "${{ matrix.python-version }}" != "3.7" && "${{ matrix.python-version }}" != "3.8" ]]; then
6569
python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple --pre -U -e .[test-deferred-bindings]
6670
fi

‎.github/workflows/ci_ut_workflow.yml

+5-1
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,11 @@ jobs:
5858
5959
python -m pip install --upgrade pip
6060
python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple -U azure-functions --pre
61-
python -m pip install -U -e .[dev]
61+
62+
python -m pip install -U -e .[dev]
63+
if [[ "${{ matrix.python-version }}" != "3.7" ]]; then
64+
python -m pip install --index-url https://test.pypi.org/simple/ --extra-index-url https://pypi.org/simple --pre -U -e .[test-http-v2]
65+
fi
6266
6367
# Retry a couple times to avoid certificate issue
6468
retry 5 python setup.py build

‎azure_functions_worker/bindings/meta.py

+42-4
Original file line numberDiff line numberDiff line change
@@ -4,26 +4,54 @@
44
import sys
55
import typing
66

7-
from .. import protos
87

8+
from .. import protos
99
from . import datumdef
1010
from . import generic
1111

1212
from .shared_memory_data_transfer import SharedMemoryManager
13-
from ..constants import CUSTOMER_PACKAGES_PATH
13+
from ..http_v2 import HttpV2Registry
14+
from ..constants import CUSTOMER_PACKAGES_PATH, HTTP, HTTP_TRIGGER, \
15+
BASE_EXT_SUPPORTED_PY_MINOR_VERSION
1416
from ..logging import logger
1517

18+
1619
PB_TYPE = 'rpc_data'
1720
PB_TYPE_DATA = 'data'
1821
PB_TYPE_RPC_SHARED_MEMORY = 'rpc_shared_memory'
19-
# Base extension supported Python minor version
20-
BASE_EXT_SUPPORTED_PY_MINOR_VERSION = 8
2122

2223
BINDING_REGISTRY = None
2324
DEFERRED_BINDING_REGISTRY = None
2425
deferred_bindings_cache = {}
2526

2627

28+
def _check_http_input_type_annotation(bind_name: str, pytype: type,
29+
is_deferred_binding: bool) -> bool:
30+
if HttpV2Registry.http_v2_enabled():
31+
return HttpV2Registry.ext_base().RequestTrackerMeta \
32+
.check_type(pytype)
33+
34+
binding = get_binding(bind_name, is_deferred_binding)
35+
return binding.check_input_type_annotation(pytype)
36+
37+
38+
def _check_http_output_type_annotation(bind_name: str, pytype: type) -> bool:
39+
if HttpV2Registry.http_v2_enabled():
40+
return HttpV2Registry.ext_base().ResponseTrackerMeta.check_type(pytype)
41+
42+
binding = get_binding(bind_name)
43+
return binding.check_output_type_annotation(pytype)
44+
45+
46+
INPUT_TYPE_CHECK_OVERRIDE_MAP = {
47+
HTTP_TRIGGER: _check_http_input_type_annotation
48+
}
49+
50+
OUTPUT_TYPE_CHECK_OVERRIDE_MAP = {
51+
HTTP: _check_http_output_type_annotation
52+
}
53+
54+
2755
def load_binding_registry() -> None:
2856
"""
2957
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:
89117
def check_input_type_annotation(bind_name: str,
90118
pytype: type,
91119
is_deferred_binding: bool) -> bool:
120+
global INPUT_TYPE_CHECK_OVERRIDE_MAP
121+
if bind_name in INPUT_TYPE_CHECK_OVERRIDE_MAP:
122+
return INPUT_TYPE_CHECK_OVERRIDE_MAP[bind_name](bind_name, pytype,
123+
is_deferred_binding)
124+
92125
binding = get_binding(bind_name, is_deferred_binding)
126+
93127
return binding.check_input_type_annotation(pytype)
94128

95129

96130
def check_output_type_annotation(bind_name: str, pytype: type) -> bool:
131+
global OUTPUT_TYPE_CHECK_OVERRIDE_MAP
132+
if bind_name in OUTPUT_TYPE_CHECK_OVERRIDE_MAP:
133+
return OUTPUT_TYPE_CHECK_OVERRIDE_MAP[bind_name](bind_name, pytype)
134+
97135
binding = get_binding(bind_name)
98136
return binding.check_output_type_annotation(pytype)
99137

‎azure_functions_worker/constants.py

+15-1
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@
1010
WORKER_STATUS = "WorkerStatus"
1111
SHARED_MEMORY_DATA_TRANSFER = "SharedMemoryDataTransfer"
1212
FUNCTION_DATA_CACHE = "FunctionDataCache"
13+
HTTP_URI = "HttpUri"
1314

1415
# Platform Environment Variables
1516
AZURE_WEBJOBS_SCRIPT_ROOT = "AzureWebJobsScriptRoot"
@@ -54,9 +55,22 @@
5455
RETRY_POLICY = "retry_policy"
5556

5657
# Paths
57-
CUSTOMER_PACKAGES_PATH = "/home/site/wwwroot/.python_packages/lib/site-packages"
58+
CUSTOMER_PACKAGES_PATH = "/home/site/wwwroot/.python_packages/lib/site" \
59+
"-packages"
5860

5961
# Flag to index functions in handle init request
6062
PYTHON_ENABLE_INIT_INDEXING = "PYTHON_ENABLE_INIT_INDEXING"
6163

6264
METADATA_PROPERTIES_WORKER_INDEXED = "worker_indexed"
65+
66+
# Header names
67+
X_MS_INVOCATION_ID = "x-ms-invocation-id"
68+
69+
# Trigger Names
70+
HTTP_TRIGGER = "httpTrigger"
71+
72+
# Output Names
73+
HTTP = "http"
74+
75+
# Base extension supported Python minor version
76+
BASE_EXT_SUPPORTED_PY_MINOR_VERSION = 8

‎azure_functions_worker/dispatcher.py

+57-16
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
from datetime import datetime
2020

2121
import grpc
22-
2322
from . import bindings, constants, functions, loader, protos
2423
from .bindings.shared_memory_data_transfer import SharedMemoryManager
2524
from .constants import (PYTHON_ROLLBACK_CWD_PATH,
@@ -33,6 +32,8 @@
3332
PYTHON_LANGUAGE_RUNTIME, PYTHON_ENABLE_INIT_INDEXING,
3433
METADATA_PROPERTIES_WORKER_INDEXED)
3534
from .extension import ExtensionManager
35+
from .http_v2 import http_coordinator, initialize_http_server, HttpV2Registry, \
36+
sync_http_request, HttpServerInitError
3637
from .logging import disable_console_logging, enable_console_logging
3738
from .logging import (logger, error_logger, is_system_log_category,
3839
CONSOLE_LOG_PREFIX, format_exception)
@@ -158,6 +159,7 @@ async def dispatch_forever(self): # sourcery skip: swap-if-expression
158159

159160
log_level = logging.INFO if not is_envvar_true(
160161
PYTHON_ENABLE_DEBUG_LOGGING) else logging.DEBUG
162+
161163
root_logger.setLevel(log_level)
162164
root_logger.addHandler(logging_handler)
163165
logger.info('Switched to gRPC logging.')
@@ -189,7 +191,8 @@ def stop(self) -> None:
189191

190192
self._stop_sync_call_tp()
191193

192-
def on_logging(self, record: logging.LogRecord, formatted_msg: str) -> None:
194+
def on_logging(self, record: logging.LogRecord,
195+
formatted_msg: str) -> None:
193196
if record.levelno >= logging.CRITICAL:
194197
log_level = protos.RpcLog.Critical
195198
elif record.levelno >= logging.ERROR:
@@ -306,6 +309,13 @@ async def _handle__worker_init_request(self, request):
306309
self.load_function_metadata(
307310
worker_init_request.function_app_directory,
308311
caller_info="worker_init_request")
312+
313+
if HttpV2Registry.http_v2_enabled():
314+
capabilities[constants.HTTP_URI] = \
315+
initialize_http_server(self._host)
316+
317+
except HttpServerInitError:
318+
raise
309319
except Exception as ex:
310320
self._function_metadata_exception = ex
311321

@@ -508,6 +518,7 @@ async def _handle__invocation_request(self, request):
508518
logger.info(', '.join(function_invocation_logs))
509519

510520
args = {}
521+
511522
for pb in invoc_request.input_data:
512523
pb_type_info = fi.input_types[pb.name]
513524
if bindings.is_trigger_binding(pb_type_info.binding_name):
@@ -523,7 +534,19 @@ async def _handle__invocation_request(self, request):
523534
shmem_mgr=self._shmem_mgr,
524535
is_deferred_binding=pb_type_info.deferred_bindings_enabled)
525536

526-
fi_context = self._get_context(invoc_request, fi.name, fi.directory)
537+
http_v2_enabled = self._functions.get_function(function_id) \
538+
.is_http_func and \
539+
HttpV2Registry.http_v2_enabled()
540+
541+
if http_v2_enabled:
542+
http_request = await http_coordinator.get_http_request_async(
543+
invocation_id)
544+
545+
await sync_http_request(http_request, invoc_request)
546+
args[fi.trigger_metadata.get('param_name')] = http_request
547+
548+
fi_context = self._get_context(invoc_request, fi.name,
549+
fi.directory)
527550

528551
# Use local thread storage to store the invocation ID
529552
# for a customer's threads
@@ -536,17 +559,21 @@ async def _handle__invocation_request(self, request):
536559
args[name] = bindings.Out()
537560

538561
if fi.is_async:
539-
call_result = await self._run_async_func(
540-
fi_context, fi.func, args
541-
)
562+
call_result = \
563+
await self._run_async_func(fi_context, fi.func, args)
542564
else:
543565
call_result = await self._loop.run_in_executor(
544566
self._sync_call_tp,
545567
self._run_sync_func,
546568
invocation_id, fi_context, fi.func, args)
569+
547570
if call_result is not None and not fi.has_return:
548-
raise RuntimeError(f'function {fi.name!r} without a $return '
549-
'binding returned a non-None value')
571+
raise RuntimeError(
572+
f'function {fi.name!r} without a $return binding'
573+
'returned a non-None value')
574+
575+
if http_v2_enabled:
576+
http_coordinator.set_http_response(invocation_id, call_result)
550577

551578
output_data = []
552579
cache_enabled = self._function_data_cache_enabled
@@ -566,10 +593,12 @@ async def _handle__invocation_request(self, request):
566593
output_data.append(param_binding)
567594

568595
return_value = None
569-
if fi.return_type is not None:
596+
if fi.return_type is not None and not http_v2_enabled:
570597
return_value = bindings.to_outgoing_proto(
571-
fi.return_type.binding_name, call_result,
572-
pytype=fi.return_type.pytype)
598+
fi.return_type.binding_name,
599+
call_result,
600+
pytype=fi.return_type.pytype,
601+
)
573602

574603
# Actively flush customer print() function to console
575604
sys.stdout.flush()
@@ -584,6 +613,9 @@ async def _handle__invocation_request(self, request):
584613
output_data=output_data))
585614

586615
except Exception as ex:
616+
if http_v2_enabled:
617+
http_coordinator.set_http_response(invocation_id, ex)
618+
587619
return protos.StreamingMessage(
588620
request_id=self.request_id,
589621
invocation_response=protos.InvocationResponse(
@@ -640,11 +672,18 @@ async def _handle__function_environment_reload_request(self, request):
640672
# reload_customer_libraries call clears the registry
641673
bindings.load_binding_registry()
642674

675+
capabilities = {}
643676
if is_envvar_true(PYTHON_ENABLE_INIT_INDEXING):
644677
try:
645678
self.load_function_metadata(
646679
directory,
647680
caller_info="environment_reload_request")
681+
682+
if HttpV2Registry.http_v2_enabled():
683+
capabilities[constants.HTTP_URI] = \
684+
initialize_http_server(self._host)
685+
except HttpServerInitError:
686+
raise
648687
except Exception as ex:
649688
self._function_metadata_exception = ex
650689

@@ -655,7 +694,7 @@ async def _handle__function_environment_reload_request(self, request):
655694
func_env_reload_request.function_app_directory)
656695

657696
success_response = protos.FunctionEnvironmentReloadResponse(
658-
capabilities={},
697+
capabilities=capabilities,
659698
worker_metadata=self.get_worker_metadata(),
660699
result=protos.StatusResult(
661700
status=protos.StatusResult.Success))
@@ -676,8 +715,10 @@ async def _handle__function_environment_reload_request(self, request):
676715

677716
def index_functions(self, function_path: str):
678717
indexed_functions = loader.index_function_app(function_path)
679-
logger.info('Indexed function app and found %s functions',
680-
len(indexed_functions))
718+
logger.info(
719+
"Indexed function app and found %s functions",
720+
len(indexed_functions)
721+
)
681722

682723
if indexed_functions:
683724
fx_metadata_results, fx_bindings_logs = (
@@ -747,7 +788,8 @@ async def _handle__close_shared_memory_resources_request(self, request):
747788
@staticmethod
748789
def _get_context(invoc_request: protos.InvocationRequest, name: str,
749790
directory: str) -> bindings.Context:
750-
""" For more information refer: https://aka.ms/azfunc-invocation-context
791+
""" For more information refer:
792+
https://aka.ms/azfunc-invocation-context
751793
"""
752794
trace_context = bindings.TraceContext(
753795
invoc_request.trace_context.trace_parent,
@@ -889,7 +931,6 @@ def gen(resp_queue):
889931

890932

891933
class AsyncLoggingHandler(logging.Handler):
892-
893934
def emit(self, record: LogRecord) -> None:
894935
# Since we disable console log after gRPC channel is initiated,
895936
# we should redirect all the messages into dispatcher.

0 commit comments

Comments
 (0)