Merge pull request #38 from Snawoot/async_logging

async logging

postfix_mta_sts_resolver/daemon.py

@@ -84,25 +84,26 @@ async def amain(cfg, loop):  # pragma: no cover
 def main():  # pragma: no cover
     # Parse command line arguments and setup basic logging
     args = parse_args()
-    logger = utils.setup_logger('MAIN', args.verbosity, args.logfile)
-    utils.setup_logger('STS', args.verbosity, args.logfile)
-    logger.info("MTA-STS daemon starting...")
-
-    # Read config and populate with defaults
-    cfg = utils.load_config(args.config)
-
-    # Construct event loop
-    logger.info("Starting eventloop...")
-    if not args.disable_uvloop:
-        if utils.enable_uvloop():
-            logger.info("uvloop enabled.")
-        else:
-            logger.info("uvloop is not available. "
-                        "Falling back to built-in event loop.")
-    evloop = asyncio.get_event_loop()
-    logger.info("Eventloop started.")
-
-
-    evloop.run_until_complete(amain(cfg, evloop))
-    evloop.close()
-    logger.info("Server finished its work.")
+    with utils.AsyncLoggingHandler(args.logfile) as log_handler:
+        logger = utils.setup_logger('MAIN', args.verbosity, log_handler)
+        utils.setup_logger('STS', args.verbosity, log_handler)
+        logger.info("MTA-STS daemon starting...")
+
+        # Read config and populate with defaults
+        cfg = utils.load_config(args.config)
+
+        # Construct event loop
+        logger.info("Starting eventloop...")
+        if not args.disable_uvloop:
+            if utils.enable_uvloop():
+                logger.info("uvloop enabled.")
+            else:
+                logger.info("uvloop is not available. "
+                            "Falling back to built-in event loop.")
+        evloop = asyncio.get_event_loop()
+        logger.info("Eventloop started.")
+
+
+        evloop.run_until_complete(amain(cfg, evloop))
+        evloop.close()
+        logger.info("Server finished its work.")

postfix_mta_sts_resolver/utils.py

@@ -1,7 +1,9 @@
 import enum
 import logging
+import logging.handlers
 import asyncio
 import socket
+import queue
 
 import yaml
 
@@ -20,19 +22,43 @@ def __str__(self):
         return self.name
 
 
-def setup_logger(name, verbosity, logfile=None):
+class OverflowingQueue(queue.Queue):
+    def put(self, item, block=True, timeout=None):
+        try:
+            return queue.Queue.put(self, item, block, timeout)
+        except queue.Full:
+            pass
 
+    def put_nowait(self, item):
+        return self.put(item, False)
+
+
+class AsyncLoggingHandler:
+    def __init__(self, logfile=None, maxsize=1024):
+        _queue = OverflowingQueue(maxsize)
+        if logfile is None:
+            _handler = logging.StreamHandler()
+        else:
+            _handler = logging.FileHandler(logfile)
+        self._listener = logging.handlers.QueueListener(_queue, _handler)
+        self._async_handler = logging.handlers.QueueHandler(_queue)
+
+        _handler.setFormatter(logging.Formatter('%(asctime)s '
+                                                '%(levelname)-8s '
+                                                '%(name)s: %(message)s',
+                                                '%Y-%m-%d %H:%M:%S'))
+
+    def __enter__(self):
+        self._listener.start()
+        return self._async_handler
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self._listener.stop()
+
+
+def setup_logger(name, verbosity, handler):
     logger = logging.getLogger(name)
     logger.setLevel(verbosity)
-    if logfile is None:
-        handler = logging.StreamHandler()
-    else:
-        handler = logging.FileHandler(logfile)
-    handler.setLevel(verbosity)
-    handler.setFormatter(logging.Formatter('%(asctime)s '
-                                           '%(levelname)-8s '
-                                           '%(name)s: %(message)s',
-                                           '%Y-%m-%d %H:%M:%S'))
     logger.addHandler(handler)
     return logger
 

tests/test_utils.py

@@ -2,6 +2,7 @@
 import collections.abc
 import enum
 import itertools
+import time
 
 import pytest
 import postfix_mta_sts_resolver.utils as utils
@@ -98,12 +99,25 @@ def test_filter_text(vector):
 
 def test_setup_logger():
     with tempfile.NamedTemporaryFile('r') as tmpfile:
-        logger = utils.setup_logger("test", utils.LogLevel.info, tmpfile.name)
-        logger.info("Hello World!")
-        assert "Hello World!" in tmpfile.read()
+        with utils.AsyncLoggingHandler(tmpfile.name) as log_handler:
+            logger = utils.setup_logger("test", utils.LogLevel.info, log_handler)
+            logger.info("Hello World!")
+            time.sleep(1)
+            assert "Hello World!" in tmpfile.read()
+
+def test_setup_logger_overflow():
+    with tempfile.NamedTemporaryFile('r') as tmpfile:
+        with utils.AsyncLoggingHandler(tmpfile.name, 1) as log_handler:
+            logger = utils.setup_logger("test", utils.LogLevel.info, log_handler)
+            for _ in range(10):
+                logger.info("Hello World!")
+            time.sleep(1)
+            assert "Hello World!" in tmpfile.read()
 
 def test_setup_logger_stderr(capsys):
-    logger = utils.setup_logger("test", utils.LogLevel.info)
-    logger.info("Hello World!")
-    captured = capsys.readouterr()
-    assert "Hello World!" in captured.err
+    with utils.AsyncLoggingHandler() as log_handler:
+        logger = utils.setup_logger("test", utils.LogLevel.info, log_handler)
+        logger.info("Hello World!")
+        time.sleep(1)
+        captured = capsys.readouterr()
+        assert "Hello World!" in captured.err