(logging): Add multiprocess support for logging.

src/meta/ipc/server.py

@@ -4,10 +4,14 @@ import logging
 import string
 import random
 
-from ..logger import log_context, log_app, logging_context
+from ..logger import log_context, log_app, logging_context, setup_main_logger
+from ..config import conf
 
 logger = logging.getLogger(__name__)
 
+for name in conf.config.options('LOGGING_LEVELS', no_defaults=True):
+    logging.getLogger(name).setLevel(conf.logging_levels[name])
+
 
 uuid_alphabet = string.ascii_lowercase + string.digits
 
@@ -166,6 +170,7 @@ class AppServer:
 
 
 async def start_server():
+    setup_main_logger()
     address = {'host': '127.0.0.1', 'port': '5000'}
     server = AppServer()
     await server.start(address)

src/meta/logger.py

@@ -3,7 +3,8 @@ import logging
 import asyncio
 from typing import List
 from logging.handlers import QueueListener, QueueHandler
-from queue import SimpleQueue
+import queue
+import multiprocessing
 from contextlib import contextmanager
 from io import StringIO
 from functools import wraps
@@ -199,6 +200,7 @@ class WebHookHandler(logging.StreamHandler):
         return self.loop
 
     def emit(self, record):
+        self.format(record)
         self.get_loop().call_soon_threadsafe(self._post, record)
 
     def _post(self, record):
@@ -292,35 +294,42 @@ if webhook := conf.logging['critical_log']:
     handler.setLevel(logging.CRITICAL)
     handlers.append(handler)
 
-if handlers:
-    # First create a separate loop to run the handlers on
-    import threading
-
-    def run_loop(loop):
-        asyncio.set_event_loop(loop)
-        try:
-            loop.run_forever()
-        finally:
-            loop.run_until_complete(loop.shutdown_asyncgens())
-            loop.close()
-
-    loop = asyncio.new_event_loop()
-    loop_thread = threading.Thread(target=lambda: run_loop(loop))
-    loop_thread.daemon = True
-    loop_thread.start()
-
-    for handler in handlers:
-        handler.loop = loop
-
-    queue: SimpleQueue[logging.LogRecord] = SimpleQueue()
 
+def make_queue_handler(queue):
     qhandler = QueueHandler(queue)
     qhandler.setLevel(logging.INFO)
     qhandler.addFilter(ContextInjection())
-    # qhandler.addFilter(ThreadFilter('MainThread'))
-    logger.addHandler(qhandler)
+    return qhandler
 
-    listener = QueueListener(
-        queue, *handlers, respect_handler_level=True
-    )
-    listener.start()
+
+def setup_main_logger(multiprocess=False):
+    q = multiprocessing.Queue() if multiprocess else queue.SimpleQueue()
+    if handlers:
+        # First create a separate loop to run the handlers on
+        import threading
+
+        def run_loop(loop):
+            asyncio.set_event_loop(loop)
+            try:
+                loop.run_forever()
+            finally:
+                loop.run_until_complete(loop.shutdown_asyncgens())
+                loop.close()
+
+        loop = asyncio.new_event_loop()
+        loop_thread = threading.Thread(target=lambda: run_loop(loop))
+        loop_thread.daemon = True
+        loop_thread.start()
+
+        for handler in handlers:
+            handler.loop = loop
+
+        qhandler = make_queue_handler(q)
+        # qhandler.addFilter(ThreadFilter('MainThread'))
+        logger.addHandler(qhandler)
+
+        listener = QueueListener(
+            q, *handlers, respect_handler_level=True
+        )
+        listener.start()
+    return q