Fix no currently running Dispatcher is found issue (#715)

* Fix no currently running Dispatcher is found issue

* Address unittest scenarios

* Log to stderr with console prefix when AsyncLoggingHandler fails

* Fix syntax issue

Author: Hazhzeng

azure_functions_worker/constants.py

@@ -1,5 +1,9 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+
+# Prefixes
+CONSOLE_LOG_PREFIX = "LanguageWorkerConsoleLog"
+
 # Capabilities
 RAW_HTTP_BODY_BYTES = "RawHttpBodyBytes"
 TYPED_DATA_COLLECTION = "TypedDataCollection"

azure_functions_worker/dispatcher.py

@@ -23,11 +23,12 @@
 from . import protos
 from . import constants
 
+from .constants import CONSOLE_LOG_PREFIX
 from .logging import error_logger, logger, is_system_log_category
 from .logging import enable_console_logging, disable_console_logging
 from .utils.tracing import marshall_exception_trace
 from .utils.wrappers import disable_feature_by
-from asyncio.unix_events import _UnixSelectorEventLoop
+from asyncio import BaseEventLoop
 from logging import LogRecord
 from typing import Optional
 
@@ -48,7 +49,7 @@ class Dispatcher(metaclass=DispatcherMeta):
 
     _GRPC_STOP_RESPONSE = object()
 
-    def __init__(self, loop: _UnixSelectorEventLoop, host: str, port: int,
+    def __init__(self, loop: BaseEventLoop, host: str, port: int,
                  worker_id: str, request_id: str,
                  grpc_connect_timeout: float,
                  grpc_max_msg_len: int = -1) -> None:
@@ -113,22 +114,29 @@ async def dispatch_forever(self):
             self._loop.set_task_factory(
                 lambda loop, coro: ContextEnabledTask(coro, loop=loop))
 
-            # Attach gRPC logging to the root logger
+            # Detach console logging before enabling GRPC channel logging
+            logger.info('Detaching console logging.')
+            disable_console_logging()
+
+            # Attach gRPC logging to the root logger. Since gRPC channel is
+            # established, should use it for system and user logs
             logging_handler = AsyncLoggingHandler()
             root_logger = logging.getLogger()
             root_logger.setLevel(logging.INFO)
             root_logger.addHandler(logging_handler)
-
-            # Since gRPC channel is established, should use it for logging
-            disable_console_logging()
-            logger.info('Detach console logging. Switch to gRPC logging')
+            logger.info('Switched to gRPC logging.')
+            logging_handler.flush()
 
             try:
                 await forever
             finally:
-                # Re-enable console logging when there's an exception
-                enable_console_logging()
+                logger.warn('Detaching gRPC logging due to exception.')
+                logging_handler.flush()
                 root_logger.removeHandler(logging_handler)
+
+                # Reenable console logging when there's an exception
+                enable_console_logging()
+                logger.warn('Switched to console logging due to exception.')
         finally:
             DispatcherMeta.__current_dispatcher__ = None
 
@@ -163,7 +171,7 @@ def on_logging(self, record: logging.LogRecord, formatted_msg: str) -> None:
 
         if is_system_log_category(record.name):
             log_category = protos.RpcLog.RpcLogCategory.System
-        else:
+        else:  # customers using logging will yield 'root' in record.name
             log_category = protos.RpcLog.RpcLogCategory.User
 
         log = dict(
@@ -365,6 +373,9 @@ async def _handle__invocation_request(self, req):
                     fi.return_type.binding_name, call_result,
                     pytype=fi.return_type.pytype)
 
+            # Actively flush customer print() function to console
+            sys.stdout.flush()
+
             logger.info('Successfully processed FunctionInvocationRequest, '
                         'request ID: %s, function ID: %s, invocation ID: %s',
                         self.request_id, function_id, invocation_id)
@@ -518,10 +529,23 @@ 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
+        # Since we disable console log after gRPC channel is initiated,
+        # we should redirect all the messages into dispatcher.
+
+        # When dispatcher receives an exception, it should switch back
+        # to console logging. However, it is possible that
+        # __current_dispatcher__ is set to None as there are still messages
+        # buffered in this handler, not calling the emit yet.
         msg = self.format(record)
-        Dispatcher.current.on_logging(record, msg)
+        try:
+            Dispatcher.current.on_logging(record, msg)
+        except RuntimeError as runtime_error:
+            # This will cause 'Dispatcher not found' failure.
+            # Logging such of an issue will cause infinite loop of gRPC logging
+            # To mitigate, we should suppress the 2nd level error logging here
+            # and use print function to report exception instead.
+            print(f'{CONSOLE_LOG_PREFIX} ERROR: {str(runtime_error)}',
+                  file=sys.stderr, flush=True)
 
 
 class ContextEnabledTask(asyncio.Task):

azure_functions_worker/logging.py

@@ -1,23 +1,33 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
+
+from typing import Optional
 import logging
 import logging.handlers
 import sys
 
+from .constants import CONSOLE_LOG_PREFIX
+
 
-logger = logging.getLogger('azure_functions_worker')
-error_logger = logging.getLogger('azure_functions_worker_errors')
+logger: logging.Logger = logging.getLogger('azure_functions_worker')
+error_logger: logging.Logger = (
+    logging.getLogger('azure_functions_worker_errors'))
 
-handler = None
-error_handler = None
+handler: Optional[logging.Handler] = None
+error_handler: Optional[logging.Handler] = None
 
 
 def setup(log_level, log_destination):
+    # Since handler and error_handler are moved to the global scope,
+    # before assigning to these handlers, we should define 'global' keyword
+    global handler
+    global error_handler
+
     if log_level == 'TRACE':
         log_level = 'DEBUG'
 
-    formatter = logging.Formatter(
-        'LanguageWorkerConsoleLog %(levelname)s: %(message)s')
+    formatter = logging.Formatter(f'{CONSOLE_LOG_PREFIX}'
+                                  ' %(levelname)s: %(message)s')
 
     if log_destination is None:
         # With no explicit log destination we do split logging,
@@ -51,25 +61,19 @@ def setup(log_level, log_destination):
 
 
 def disable_console_logging() -> None:
+    # We should only remove the sys.stdout stream, as error_logger is used for
+    # unexpected critical error logs handling.
     if logger and handler:
+        handler.flush()
         logger.removeHandler(handler)
 
-    if error_logger and error_handler:
-        error_logger.removeHandler(error_handler)
-
 
 def enable_console_logging() -> None:
     if logger and handler:
         logger.addHandler(handler)
 
-    if error_logger and error_handler:
-        error_logger.addHandler(error_handler)
-
 
 def is_system_log_category(ctg: str) -> bool:
-    return any(
-        [ctg.lower().startswith(c) for c in (
-            'azure_functions_worker',
-            'azure_functions_worker_errors'
-        )]
-    )
+    # Category starts with 'azure_functions_worker' or
+    # 'azure_functions_worker_errors' will be treated as system logs
+    return ctg.lower().startswith('azure_functions_worker')

tests/unittests/http_functions/hijack_current_event_loop/function.json

@@ -0,0 +1,19 @@
+{
+    "scriptFile": "main.py",
+    "bindings": [
+      {
+        "authLevel": "anonymous",
+        "type": "httpTrigger",
+        "direction": "in",
+        "name": "req",
+        "methods": [
+          "get"
+        ]
+      },
+      {
+        "type": "http",
+        "direction": "out",
+        "name": "$return"
+      }
+    ]
+  }

tests/unittests/http_functions/hijack_current_event_loop/main.py

@@ -0,0 +1,78 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+import sys
+import logging
+import asyncio
+
+import azure.functions as func
+
+
+logger = logging.getLogger('custom_logger')
+
+# Attempt to log info into system log from customer code
+disguised_logger = logging.getLogger('azure_functions_worker')
+
+
+async def parallelly_print():
+    await asyncio.sleep(0.1)
+    print('parallelly_print')
+
+
+async def parallelly_log_info():
+    await asyncio.sleep(0.2)
+    logging.info('parallelly_log_info at root logger')
+
+
+async def parallelly_log_warning():
+    await asyncio.sleep(0.3)
+    logging.warning('parallelly_log_warning at root logger')
+
+
+async def parallelly_log_error():
+    await asyncio.sleep(0.4)
+    logging.error('parallelly_log_error at root logger')
+
+
+async def parallelly_log_exception():
+    await asyncio.sleep(0.5)
+    try:
+        raise Exception('custom exception')
+    except Exception:
+        logging.exception('parallelly_log_exception at root logger',
+                          exc_info=sys.exc_info())
+
+
+async def parallelly_log_custom():
+    await asyncio.sleep(0.6)
+    logger.info('parallelly_log_custom at custom_logger')
+
+
+async def parallelly_log_system():
+    await asyncio.sleep(0.7)
+    disguised_logger.info('parallelly_log_system at disguised_logger')
+
+
+async def main(req: func.HttpRequest) -> func.HttpResponse:
+    loop = asyncio.get_event_loop()
+
+    # Create multiple tasks and schedule it into one asyncio.wait blocker
+    task_print: asyncio.Task = loop.create_task(parallelly_print())
+    task_info: asyncio.Task = loop.create_task(parallelly_log_info())
+    task_warning: asyncio.Task = loop.create_task(parallelly_log_warning())
+    task_error: asyncio.Task = loop.create_task(parallelly_log_error())
+    task_exception: asyncio.Task = loop.create_task(parallelly_log_exception())
+    task_custom: asyncio.Task = loop.create_task(parallelly_log_custom())
+    task_disguise: asyncio.Task = loop.create_task(parallelly_log_system())
+
+    # Create an awaitable future and occupy the current event loop resource
+    future = loop.create_future()
+    loop.call_soon_threadsafe(future.set_result, 'callsoon_log')
+
+    # WaitAll
+    await asyncio.wait([task_print, task_info, task_warning, task_error,
+                        task_exception, task_custom, task_disguise, future])
+
+    # Log asyncio low-level future result
+    logging.info(future.result())
+
+    return 'OK-hijack-current-event-loop'

tests/unittests/http_functions/print_logging/function.json

@@ -0,0 +1,15 @@
+{
+    "scriptFile": "main.py",
+    "bindings": [
+      {
+        "type": "httpTrigger",
+        "direction": "in",
+        "name": "req"
+      },
+      {
+        "type": "http",
+        "direction": "out",
+        "name": "$return"
+      }
+    ]
+  }

tests/unittests/http_functions/print_logging/main.py

@@ -0,0 +1,28 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+import sys
+import azure.functions
+
+
+def main(req: azure.functions.HttpRequest):
+    flush_required = False
+    is_console_log = False
+    is_stderr = False
+    message = req.params.get('message', '')
+
+    if req.params.get('flush') == 'true':
+        flush_required = True
+    if req.params.get('console') == 'true':
+        is_console_log = True
+    if req.params.get('is_stderr') == 'true':
+        is_stderr = True
+
+    # Adding LanguageWorkerConsoleLog will make function host to treat
+    # this as system log and will be propagated to kusto
+    prefix = 'LanguageWorkerConsoleLog' if is_console_log else ''
+    print(f'{prefix} {message}'.strip(),
+          file=sys.stderr if is_stderr else sys.stdout,
+          flush=flush_required)
+
+    return 'OK-print-logging'