Formalize worker logging (#231)

* Formalize worker logging

The new `--log-to` and `--log-level` command line arguments allow
configuring the logging behavior of the logger.  The default is
to log to stdout with the `INFO` level.  Internal logging has
been switched to use this logger, and a few debug-level logs
added around various RPC messages.

Fixes: #181.
Fixes: #226.

* Switch to split stderr/stdout logging by default

* Fix split logging

Author: elprans

azure/functions_worker/__init__.py

@@ -1,16 +0,0 @@
-from . import aio_compat
-from . import dispatcher
-
-
-async def start_async(host, port, worker_id, request_id, grpc_max_msg_len):
-    disp = await dispatcher.Dispatcher.connect(
-        host, port, worker_id, request_id,
-        connect_timeout=5.0, max_msg_len=grpc_max_msg_len)
-
-    await disp.dispatch_forever()
-
-
-def start(host: str, port: int, worker_id: str, request_id: str,
-          grpc_max_msg_len: int):
-    return aio_compat.run(start_async(
-        host, port, worker_id, request_id, grpc_max_msg_len))

azure/functions_worker/__main__.py

@@ -0,0 +1,4 @@
+from azure.functions_worker import main
+
+if __name__ == '__main__':
+    main.main()

azure/functions_worker/dispatcher.py

@@ -17,6 +17,8 @@
 from . import loader
 from . import protos
 
+from .logging import error_logger, logger
+
 
 class DispatcherMeta(type):
 
@@ -62,8 +64,6 @@ def __init__(self, loop, host, port, worker_id, request_id,
         self._grpc_thread = threading.Thread(
             name='grpc-thread', target=self.__poll_grpc)
 
-        self._logger = logging.getLogger('azure.functions_worker')
-
     @classmethod
     async def connect(cls, host, port, worker_id, request_id,
                       connect_timeout, max_msg_len=None):
@@ -72,6 +72,7 @@ async def connect(cls, host, port, worker_id, request_id,
                    connect_timeout, max_msg_len)
         disp._grpc_thread.start()
         await disp._grpc_connected_fut
+        logger.info('Successfully opened gRPC channel to %s:%s', host, port)
         return disp
 
     async def dispatch_forever(self):
@@ -186,14 +187,16 @@ async def _dispatch_grpc_request(self, request):
             # Don't crash on unknown messages.  Some of them can be ignored;
             # and if something goes really wrong the host can always just
             # kill the worker's process.
-            self._logger.error(
+            logger.error(
                 f'unknown StreamingMessage content type {content_type}')
             return
 
         resp = await request_handler(request)
         self._grpc_resp_queue.put_nowait(resp)
 
     async def _handle__worker_init_request(self, req):
+        logger.info('Received WorkerInitRequest, request ID %s',
+                    self.request_id)
         return protos.StreamingMessage(
             request_id=self.request_id,
             worker_init_response=protos.WorkerInitResponse(
@@ -204,6 +207,9 @@ async def _handle__function_load_request(self, req):
         func_request = req.function_load_request
         function_id = func_request.function_id
 
+        logger.info('Received FunctionLoadRequest, request ID: %s, '
+                    'function ID: %s', self.request_id, function_id)
+
         try:
             func = loader.load_function(
                 func_request.metadata.name,
@@ -214,6 +220,10 @@ async def _handle__function_load_request(self, req):
             self._functions.add_function(
                 function_id, func, func_request.metadata)
 
+            logger.info('Successfully processed FunctionLoadRequest, '
+                        'request ID: %s, function ID: %s',
+                        self.request_id, function_id)
+
             return protos.StreamingMessage(
                 request_id=self.request_id,
                 function_load_response=protos.FunctionLoadResponse(
@@ -242,6 +252,10 @@ async def _handle__invocation_request(self, req):
         assert isinstance(current_task, ContextEnabledTask)
         current_task.set_azure_invocation_id(invocation_id)
 
+        logger.info('Received FunctionInvocationRequest, request ID: %s, '
+                    'function ID: %s, invocation ID: %s',
+                    self.request_id, function_id, invocation_id)
+
         try:
             fi: functions.FunctionInfo = self._functions.get_function(
                 function_id)
@@ -303,6 +317,10 @@ async def _handle__invocation_request(self, req):
                     fi.return_type.binding_name, call_result,
                     pytype=fi.return_type.pytype)
 
+            logger.info('Successfully processed FunctionInvocationRequest, '
+                        'request ID: %s, function ID: %s, invocation ID: %s',
+                        self.request_id, function_id, invocation_id)
+
             return protos.StreamingMessage(
                 request_id=self.request_id,
                 invocation_response=protos.InvocationResponse(
@@ -371,14 +389,17 @@ def gen(resp_queue):
             if ex is grpc_req_stream:
                 # Yes, this is how grpc_req_stream iterator exits.
                 return
+            error_logger.exception('unhandled error in gRPC thread')
             raise
 
 
 class AsyncLoggingHandler(logging.Handler):
 
     def emit(self, record):
-        msg = self.format(record)
-        Dispatcher.current._on_logging(record, msg)
+        if not record.name.startswith('azure.functions_worker'):
+            # Skip worker system logs
+            msg = self.format(record)
+            Dispatcher.current._on_logging(record, msg)
 
 
 class ContextEnabledTask(asyncio.Task):

azure/functions_worker/logging.py

@@ -0,0 +1,48 @@
+import logging
+import logging.handlers
+import sys
+
+
+logger = logging.getLogger('azure.functions_worker')
+error_logger = logging.getLogger('azure.functions_worker_errors')
+
+
+def setup(log_level, log_destination):
+    if log_level == 'TRACE':
+        log_level = 'DEBUG'
+
+    formatter = logging.Formatter(
+        'LanguageWorkerConsoleLog %(levelname)s: %(message)s')
+
+    error_handler = None
+    handler = None
+
+    if log_destination is None:
+        # With no explicit log destination we do split logging,
+        # errors go into stderr, everything else -- to stdout.
+        error_handler = logging.StreamHandler(sys.stderr)
+        error_handler.setFormatter(formatter)
+        error_handler.setLevel(getattr(logging, log_level))
+
+        handler = logging.StreamHandler(sys.stdout)
+
+    elif log_destination in ('stdout', 'stderr'):
+        handler = logging.StreamHandler(getattr(sys, log_destination))
+
+    elif log_destination == 'syslog':
+        handler = logging.handlers.SysLogHandler()
+
+    else:
+        handler = logging.FileHandler(log_destination)
+
+    if error_handler is None:
+        error_handler = handler
+
+    handler.setFormatter(formatter)
+    handler.setLevel(getattr(logging, log_level))
+
+    logger.addHandler(handler)
+    logger.setLevel(getattr(logging, log_level))
+
+    error_logger.addHandler(error_handler)
+    error_logger.setLevel(getattr(logging, log_level))

azure/functions_worker/main.py

@@ -0,0 +1,51 @@
+"""Main entrypoint."""
+
+
+import argparse
+
+from . import aio_compat
+from . import dispatcher
+from . import logging
+from .logging import error_logger, logger
+
+
+def parse_args():
+    parser = argparse.ArgumentParser(
+        description='Python Azure Functions Worker')
+    parser.add_argument('--host')
+    parser.add_argument('--port', type=int)
+    parser.add_argument('--workerId', dest='worker_id')
+    parser.add_argument('--requestId', dest='request_id')
+    parser.add_argument('--log-level', type=str, default='INFO',
+                        choices=['TRACE', 'INFO', 'WARNING', 'ERROR'],)
+    parser.add_argument('--log-to', type=str, default=None,
+                        help='log destination: stdout, stderr, '
+                             'syslog, or a file path')
+    parser.add_argument('--grpcMaxMessageLength', type=int,
+                        dest='grpc_max_msg_len')
+    return parser.parse_args()
+
+
+def main():
+    args = parse_args()
+    logging.setup(log_level=args.log_level, log_destination=args.log_to)
+
+    logger.info('Starting Azure Functions Python Worker.')
+    logger.info('Worker ID: %s, Request ID: %s, Host Address: %s:%s',
+                args.worker_id, args.request_id, args.host, args.port)
+
+    try:
+        return aio_compat.run(start_async(
+            args.host, args.port, args.worker_id, args.request_id,
+            args.grpc_max_msg_len))
+    except Exception:
+        error_logger.exception('unhandled error in functions worker')
+        raise
+
+
+async def start_async(host, port, worker_id, request_id, grpc_max_msg_len):
+    disp = await dispatcher.Dispatcher.connect(
+        host, port, worker_id, request_id,
+        connect_timeout=5.0, max_msg_len=grpc_max_msg_len)
+
+    await disp.dispatch_forever()

python/worker.py

@@ -1,36 +1,4 @@
-"""Main entrypoint."""
-
-
-import argparse
-import traceback
-
-
-def parse_args():
-    parser = argparse.ArgumentParser(
-        description='Python Azure Functions Worker')
-    parser.add_argument('--host')
-    parser.add_argument('--port', type=int)
-    parser.add_argument('--workerId', dest='worker_id')
-    parser.add_argument('--requestId', dest='request_id')
-    parser.add_argument('--grpcMaxMessageLength', type=int,
-                        dest='grpc_max_msg_len')
-    return parser.parse_args()
-
-
-def main():
-    args = parse_args()
-
-    import azure.functions  # NoQA
-    import azure.functions_worker
-
-    try:
-        azure.functions_worker.start(
-            args.host, args.port, args.worker_id, args.request_id,
-            args.grpc_max_msg_len)
-    except Exception:
-        print(traceback.format_exc(), flush=True)
-        raise
-
+from azure.functions_worker import main
 
 if __name__ == '__main__':
-    main()
+    main.main()

tests/test_mock_http_functions.py

@@ -71,10 +71,3 @@ async def test_handles_unsupported_messages_gracefully(self):
             _, r = await host.load_function('return_out')
             self.assertEqual(r.response.result.status,
                              protos.StatusResult.Success)
-
-            for log in r.logs:
-                if 'unknown StreamingMessage' in log.message:
-                    break
-            else:
-                raise AssertionError('the worker did not log about an '
-                                     '"unknown StreamingMessage"')