Merge branch 'dev'

Author: Hanzhang Zeng (Roger)

.ci/linux_devops_e2e_tests.sh

@@ -5,5 +5,7 @@ export AzureWebJobsStorage=$LINUXSTORAGECONNECTIONSTRING
 export AzureWebJobsCosmosDBConnectionString=$LINUXCOSMOSDBCONNECTIONSTRING
 export AzureWebJobsEventHubConnectionString=$LINUXEVENTHUBCONNECTIONSTRING
 export AzureWebJobsServiceBusConnectionString=$LINUXSERVICEBUSCONNECTIONSTRING
+export AzureWebJobsEventGridTopicUri=$LINUXEVENTGRIDTOPICURI
+export AzureWebJobsEventGridConnectionKey=$LINUXEVENTGRIDTOPICCONNECTIONKEY
 
-pytest  --instafail --cov=./azure_functions_worker --cov-report xml --cov-branch --cov-append tests/endtoend
+pytest  --instafail --cov=./azure_functions_worker --cov-report xml --cov-branch --cov-append tests/endtoend

.gitignore

@@ -119,3 +119,6 @@ py3env/
 # PyCharm
 .idea/
 .idea_modules/
+
+# Profiling info
+prof/

azure-pipelines.yml

@@ -31,18 +31,24 @@ jobs:
         linuxCosmosDB: $(LinuxCosmosDBConnectionString36)
         linuxEventHub: $(LinuxEventHubConnectionString36)
         linuxServiceBus: $(LinuxServiceBusConnectionString36)
+        linuxEventGridTopicUri: $(LinuxEventGridTopicUriString36)
+        linuxEventGridConnectionKey: $(LinuxEventGridConnectionKeyString36)
       Python37:
         pythonVersion: '3.7'
         linuxStorage: $(LinuxStorageConnectionString37)
         linuxCosmosDB: $(LinuxCosmosDBConnectionString37)
         linuxEventHub: $(LinuxEventHubConnectionString37)
         linuxServiceBus: $(LinuxServiceBusConnectionString37)
+        linuxEventGridTopicUri: $(LinuxEventGridTopicUriString37)
+        linuxEventGridConnectionKey: $(LinuxEventGridConnectionKeyString37)
       Python38:
         pythonVersion: '3.8'
         linuxStorage: $(LinuxStorageConnectionString38)
         linuxCosmosDB: $(LinuxCosmosDBConnectionString38)
         linuxEventHub: $(LinuxEventHubConnectionString38)
         linuxServiceBus: $(LinuxServiceBusConnectionString38)
+        linuxEventGridTopicUri: $(LinuxEventGridTopicUriString38)
+        linuxEventGridConnectionKey: $(LinuxEventGridConnectionKeyString38)
   steps:
   - task: UsePythonVersion@0
     inputs:
@@ -70,6 +76,8 @@ jobs:
       LINUXCOSMOSDBCONNECTIONSTRING: $(linuxCosmosDB)
       LINUXEVENTHUBCONNECTIONSTRING: $(linuxEventHub)
       LINUXSERVICEBUSCONNECTIONSTRING: $(linuxServiceBus)
+      LINUXEVENTGRIDTOPICURI: $(linuxEventGridTopicUri)
+      LINUXEVENTGRIDTOPICCONNECTIONKEY: $(linuxEventGridConnectionKey)
     displayName: 'E2E Tests'
   - task: PublishCodeCoverageResults@1
     inputs:

azure_functions_worker/dispatcher.py

@@ -399,6 +399,7 @@ async def _handle__invocation_request(self, req):
                         exception=self._serialize_exception(ex))))
 
     async def _handle__function_environment_reload_request(self, req):
+        '''Only runs on Linux Consumption placeholder specialization.'''
         try:
             logger.info('Received FunctionEnvironmentReloadRequest, '
                         'request ID: %s', self.request_id)
@@ -410,12 +411,16 @@ async def _handle__function_environment_reload_request(self, req):
             # customer use
             import azure.functions # NoQA
 
+            # Append function project root to module finding sys.path
+            if func_env_reload_request.function_app_directory:
+                sys.path.append(func_env_reload_request.function_app_directory)
+
+            # Clear sys.path import cache, reload all module from new sys.path
             sys.path_importer_cache.clear()
 
+            # Reload environment variables
             os.environ.clear()
-
             env_vars = func_env_reload_request.environment_variables
-
             for var in env_vars:
                 os.environ[var] = env_vars[var]
 

azure_functions_worker/testutils.py

@@ -44,13 +44,14 @@
 E2E_TESTS_ROOT = TESTS_ROOT / E2E_TESTS_FOLDER
 UNIT_TESTS_FOLDER = pathlib.Path('unittests')
 UNIT_TESTS_ROOT = TESTS_ROOT / UNIT_TESTS_FOLDER
-DEFAULT_WEBHOST_DLL_PATH = PROJECT_ROOT / 'build' / 'webhost' / \
-    'Microsoft.Azure.WebJobs.Script.WebHost.dll'
+WEBHOST_DLL = "Microsoft.Azure.WebJobs.Script.WebHost.dll"
+DEFAULT_WEBHOST_DLL_PATH = PROJECT_ROOT / 'build' / 'webhost' / WEBHOST_DLL
 EXTENSIONS_PATH = PROJECT_ROOT / 'build' / 'extensions' / 'bin'
 FUNCS_PATH = TESTS_ROOT / UNIT_TESTS_FOLDER / 'http_functions'
 WORKER_PATH = PROJECT_ROOT / 'python' / 'test'
 WORKER_CONFIG = PROJECT_ROOT / '.testconfig'
 ON_WINDOWS = platform.system() == 'Windows'
+LOCALHOST = "127.0.0.1"
 
 HOST_JSON_TEMPLATE = """\
 {
@@ -317,7 +318,7 @@ def __init__(self, loop, scripts_dir):
         self._server = grpc.server(self._threadpool)
         self._servicer = _MockWebHostServicer(self)
         protos.add_FunctionRpcServicer_to_server(self._servicer, self._server)
-        self._port = self._server.add_insecure_port('127.0.0.1:0')
+        self._port = self._server.add_insecure_port(f'{LOCALHOST}:0')
 
         self._worker_id = self.make_id()
         self._request_id = self.make_id()
@@ -459,7 +460,7 @@ async def __aenter__(self):
         await self._host.start()
 
         self._worker = await dispatcher. \
-            Dispatcher.connect('127.0.0.1', self._host._port,
+            Dispatcher.connect(LOCALHOST, self._host._port,
                                self._host.worker_id,
                                self._host.request_id, connect_timeout=5.0)
 
@@ -469,6 +470,7 @@ async def __aenter__(self):
             wait([self._host._connected_fut, self._worker_task],
                  return_when=asyncio.FIRST_COMPLETED)
 
+        # noinspection PyBroadException
         try:
             if self._worker_task in done:
                 self._worker_task.result()
@@ -502,7 +504,7 @@ async def __aexit__(self, *exc):
 def start_mockhost(*, script_root=FUNCS_PATH):
     tests_dir = TESTS_ROOT
     scripts_dir = tests_dir / script_root
-    if not scripts_dir.exists() or not scripts_dir.is_dir():
+    if not (scripts_dir.exists() and scripts_dir.is_dir()):
         raise RuntimeError(
             f'invalid script_root argument: '
             f'{scripts_dir} directory does not exist')
@@ -536,7 +538,7 @@ def close(self):
 
 def _find_open_port():
     with socket.socket() as s:
-        s.bind(('127.0.0.1', 0))
+        s.bind((LOCALHOST, 0))
         s.listen(1)
         return s.getsockname()[1]
 
@@ -572,10 +574,10 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
         if not dll:
             dll = DEFAULT_WEBHOST_DLL_PATH
 
-            secrets = SECRETS_TEMPLATE
-
             os.makedirs(dll.parent / 'Secrets', exist_ok=True)
             with open(dll.parent / 'Secrets' / 'host.json', 'w') as f:
+                secrets = SECRETS_TEMPLATE
+
                 f.write(secrets)
 
         if dll and pathlib.Path(dll).exists():
@@ -605,11 +607,7 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
         ]))
 
     worker_path = os.environ.get('PYAZURE_WORKER_DIR')
-    if not worker_path:
-        worker_path = WORKER_PATH
-    else:
-        worker_path = pathlib.Path(worker_path)
-
+    worker_path = WORKER_PATH if not worker_path else pathlib.Path(worker_path)
     if not worker_path.exists():
         raise RuntimeError(f'Worker path {worker_path} does not exist')
 
@@ -640,6 +638,15 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
         if servicebus:
             extra_env['AzureWebJobsServiceBusConnectionString'] = servicebus
 
+        eventgrid_topic_uri = testconfig['azure'].get('eventgrid_topic_uri')
+        if eventgrid_topic_uri:
+            extra_env['AzureWebJobsEventGridTopicUri'] = eventgrid_topic_uri
+
+        eventgrid_topic_key = testconfig['azure'].get('eventgrid_topic_key')
+        if eventgrid_topic_key:
+            extra_env['AzureWebJobsEventGridConnectionKey'] = \
+                eventgrid_topic_key
+
     if port is not None:
         extra_env['ASPNETCORE_URLS'] = f'http://*:{port}'
 
@@ -655,11 +662,7 @@ def popen_webhost(*, stdout, stderr, script_root=FUNCS_PATH, port=None):
 
 
 def start_webhost(*, script_dir=None, stdout=None):
-    if script_dir:
-        script_root = TESTS_ROOT / script_dir
-    else:
-        script_root = FUNCS_PATH
-
+    script_root = TESTS_ROOT / script_dir if script_dir else FUNCS_PATH
     if stdout is None:
         if is_envvar_true(PYAZURE_WEBHOST_DEBUG):
             stdout = sys.stdout
@@ -670,8 +673,8 @@ def start_webhost(*, script_dir=None, stdout=None):
     proc = popen_webhost(stdout=stdout, stderr=subprocess.STDOUT,
                          script_root=script_root, port=port)
 
-    addr = f'http://127.0.0.1:{port}'
-    for n in range(10):
+    addr = f'http://{LOCALHOST}:{port}'
+    for _ in range(10):
         try:
             r = requests.get(f'{addr}/api/ping',
                              params={'code': 'testFunctionKey'})
@@ -682,11 +685,11 @@ def start_webhost(*, script_dir=None, stdout=None):
         except requests.exceptions.ConnectionError:
             pass
 
-        time.sleep(0.5)
+        time.sleep(1)
     else:
         proc.terminate()
         try:
-            proc.wait(10)
+            proc.wait(20)
         except subprocess.TimeoutExpired:
             proc.kill()
         raise RuntimeError('could not start the webworker')
@@ -697,7 +700,7 @@ def start_webhost(*, script_dir=None, stdout=None):
 def create_dummy_dispatcher():
     dummy_event_loop = asyncio.new_event_loop()
     disp = dispatcher.Dispatcher(
-        dummy_event_loop, '127.0.0.1', 0,
+        dummy_event_loop, LOCALHOST, 0,
         'test_worker_id', 'test_request_id',
         1.0, 1000)
     dummy_event_loop.close()

python/prodV2/worker.py

@@ -13,14 +13,33 @@
 # Azure environment variables
 AZURE_WEBSITE_INSTANCE_ID = "WEBSITE_INSTANCE_ID"
 AZURE_CONTAINER_NAME = "CONTAINER_NAME"
+AZURE_WEBJOBS_SCRIPT_ROOT = "AzureWebJobsScriptRoot"
 
 
 def is_azure_environment():
+    '''Check if the function app is running on the cloud'''
     return (AZURE_CONTAINER_NAME in os.environ
             or AZURE_WEBSITE_INSTANCE_ID in os.environ)
 
 
+def add_script_root_to_sys_path():
+    '''Append function project root to module finding sys.path'''
+    functions_script_root = os.getenv(AZURE_WEBJOBS_SCRIPT_ROOT)
+    if functions_script_root is not None:
+        sys.path.append(functions_script_root)
+
+
 def determine_user_pkg_paths():
+    '''This finds the user packages when function apps are running on the cloud
+
+    For Python 3.6 app, the third-party packages can live in any of the paths:
+        /home/site/wwwroot/.python_packages/lib/site-packages
+        /home/site/wwwroot/.python_packages/lib/python3.6/site-packages
+        /home/site/wwwroot/worker_venv/lib/python3.6/site-packages
+
+    For Python 3.7, we only accept:
+        /home/site/wwwroot/.python_packages/lib/site-packages
+    '''
     minor_version = sys.version_info[1]
 
     home = Path.home()
@@ -49,13 +68,19 @@ def determine_user_pkg_paths():
         user_pkg_paths = determine_user_pkg_paths()
 
         joined_pkg_paths = os.pathsep.join(user_pkg_paths)
+
+        # On cloud, we prioritize third-party user packages
+        # over worker packages in PYTHONPATH
         env['PYTHONPATH'] = f'{joined_pkg_paths}:{func_worker_dir}'
         os.execve(sys.executable,
                   [sys.executable, '-m', 'azure_functions_worker']
                   + sys.argv[1:],
                   env)
     else:
+        # On local development, we prioritize worker packages over
+        # third-party user packages (in .venv)
         sys.path.insert(1, func_worker_dir)
+        add_script_root_to_sys_path()
         from azure_functions_worker import main
 
         main.main()

python/prodV3/worker.py

@@ -13,14 +13,33 @@
 # Azure environment variables
 AZURE_WEBSITE_INSTANCE_ID = "WEBSITE_INSTANCE_ID"
 AZURE_CONTAINER_NAME = "CONTAINER_NAME"
+AZURE_WEBJOBS_SCRIPT_ROOT = "AzureWebJobsScriptRoot"
 
 
 def is_azure_environment():
+    '''Check if the function app is running on the cloud'''
     return (AZURE_CONTAINER_NAME in os.environ
             or AZURE_WEBSITE_INSTANCE_ID in os.environ)
 
 
+def add_script_root_to_sys_path():
+    '''Append function project root to module finding sys.path'''
+    functions_script_root = os.getenv(AZURE_WEBJOBS_SCRIPT_ROOT)
+    if functions_script_root is not None:
+        sys.path.append(functions_script_root)
+
+
 def determine_user_pkg_paths():
+    '''This finds the user packages when function apps are running on the cloud
+
+    For Python 3.6 app, the third-party packages can live in any of the paths:
+        /home/site/wwwroot/.python_packages/lib/site-packages
+        /home/site/wwwroot/.python_packages/lib/python3.6/site-packages
+        /home/site/wwwroot/worker_venv/lib/python3.6/site-packages
+
+    For Python 3.7 and Python 3.8, we only accept:
+        /home/site/wwwroot/.python_packages/lib/site-packages
+    '''
     minor_version = sys.version_info[1]
 
     home = Path.home()
@@ -49,13 +68,19 @@ def determine_user_pkg_paths():
         user_pkg_paths = determine_user_pkg_paths()
 
         joined_pkg_paths = os.pathsep.join(user_pkg_paths)
+
+        # On cloud, we prioritize third-party user packages
+        # over worker packages in PYTHONPATH
         env['PYTHONPATH'] = f'{joined_pkg_paths}:{func_worker_dir}'
         os.execve(sys.executable,
                   [sys.executable, '-m', 'azure_functions_worker']
                   + sys.argv[1:],
                   env)
     else:
+        # On local development, we prioritize worker packages over
+        # third-party user packages (in .venv)
         sys.path.insert(1, func_worker_dir)
+        add_script_root_to_sys_path()
         from azure_functions_worker import main
 
         main.main()

python/test/worker.py

@@ -1,4 +1,19 @@
+import sys
+import os
 from azure_functions_worker import main
 
+
+# Azure environment variables
+AZURE_WEBJOBS_SCRIPT_ROOT = "AzureWebJobsScriptRoot"
+
+
+def add_script_root_to_sys_path():
+    '''Append function project root to module finding sys.path'''
+    functions_script_root = os.getenv(AZURE_WEBJOBS_SCRIPT_ROOT)
+    if functions_script_root is not None:
+        sys.path.append(functions_script_root)
+
+
 if __name__ == '__main__':
+    add_script_root_to_sys_path()
     main.main()

setup.py

@@ -17,8 +17,10 @@
 
 
 # TODO: change this to something more stable when available.
-WEBHOST_URL = ('https://ci.appveyor.com/api/buildjobs/19gqd7drpxhkedea'
-               '/artifacts/Functions.Binaries.2.0.13036.no-runtime.zip')
+WEBHOST_URL = (
+    'https://github.com/Azure/azure-functions-host/releases/download/'
+    '/v2.0.14361/Functions.Binaries.2.0.14361.no-runtime.zip'
+)
 
 # Extensions necessary for non-core bindings.
 AZURE_EXTENSIONS = """\
@@ -195,9 +197,8 @@ def _install_webhost(self):
                 print('Downloading Azure Functions Web Host...')
                 urllib.request.urlretrieve(self.webhost_url, zipf.name)
             except Exception as e:
-                print(
-                    f"could not download Azure Functions Web Host binaries "
-                    f"from {self.webhost_url}: {e!r}", file=sys.stderr)
+                print(f"could not download Azure Functions Web Host binaries "
+                      f"from {self.webhost_url}: {e!r}", file=sys.stderr)
                 sys.exit(1)
 
             if not self.webhost_dir.exists():
@@ -257,7 +258,7 @@ def run(self):
 
 setup(
     name='azure-functions-worker',
-    version='1.1.4',
+    version='1.1.5',
     description='Python Language Worker for Azure Functions Host',
     long_description=long_description,
     long_description_content_type='text/markdown',
@@ -285,7 +286,7 @@ def run(self):
     ],
     extras_require={
         'dev': [
-            'azure-functions==1.3.0',
+            'azure-functions==1.3.1',
             'azure-eventhub~=5.1.0',
             'python-dateutil~=2.8.1',
             'flake8~=3.7.9',