Merge pull request #220 from aiokitchen/simplify-logging

Simplify logging. Allow to passing handlers to entrypoint

Author: mosquito

aiomisc/aggregate.py

@@ -6,14 +6,7 @@
 from dataclasses import dataclass
 from inspect import Parameter
 from typing import (
-    Any,
-    Callable,
-    Coroutine,
-    Generic,
-    Iterable,
-    List,
-    Optional,
-    Protocol,
+    Any, Callable, Coroutine, Generic, Iterable, List, Optional, Protocol,
     TypeVar,
 )
 
@@ -245,7 +238,7 @@ def __init__(
 
 
 def aggregate(
-    leeway_ms: float, max_count: Optional[int] = None
+    leeway_ms: float, max_count: Optional[int] = None,
 ) -> Callable[[AggregateFunc[V, R]], Callable[[V], Coroutine[Any, Any, R]]]:
     """
     Parametric decorator that aggregates multiple
@@ -276,7 +269,7 @@ def aggregate(
     :return:
     """
     def decorator(
-        func: AggregateFunc[V, R]
+        func: AggregateFunc[V, R],
     ) -> Callable[[V], Coroutine[Any, Any, R]]:
         aggregator = Aggregator(
             func, max_count=max_count, leeway_ms=leeway_ms,
@@ -289,7 +282,7 @@ def aggregate_async(
     leeway_ms: float, max_count: Optional[int] = None,
 ) -> Callable[
     [AggregateAsyncFunc[V, R]],
-    Callable[[V], Coroutine[Any, Any, R]]
+    Callable[[V], Coroutine[Any, Any, R]],
 ]:
     """
     Same as ``aggregate``, but with ``func`` arguments of type ``Arg``
@@ -302,7 +295,7 @@ def aggregate_async(
     :return:
     """
     def decorator(
-        func: AggregateAsyncFunc[V, R]
+        func: AggregateAsyncFunc[V, R],
     ) -> Callable[[V], Coroutine[Any, Any, R]]:
         aggregator = AggregatorAsync(
             func, max_count=max_count, leeway_ms=leeway_ms,

aiomisc/backoff.py

@@ -89,7 +89,7 @@ def asyncbackoff(
     exceptions += asyncio.TimeoutError,
 
     def decorator(
-        func: Callable[P, Coroutine[Any, Any, T]]
+        func: Callable[P, Coroutine[Any, Any, T]],
     ) -> Callable[P, Coroutine[Any, Any, T]]:
         if attempt_timeout is not None:
             func = timeout(attempt_timeout)(func)

aiomisc/entrypoint.py

@@ -6,8 +6,8 @@
 import threading
 from concurrent.futures import Executor
 from typing import (
-    Any, Callable, Coroutine, FrozenSet, MutableSet, Optional, Set, Tuple,
-    TypeVar, Union,
+    Any, Callable, Coroutine, FrozenSet, Iterable, MutableSet, Optional, Set,
+    Tuple, TypeVar, Union,
 )
 from weakref import WeakSet
 
@@ -115,6 +115,7 @@ async def _start(self) -> None:
                 date_format=self.log_date_format,
                 buffered=self.log_buffering,
                 loop=self.loop,
+                handlers=self.log_handlers,
                 buffer_size=self.log_buffer_size,
                 flush_interval=self.log_flush_interval,
             )
@@ -146,6 +147,7 @@ def __init__(
         log_date_format: Optional[str] = DEFAULT_LOG_DATE_FORMAT,
         log_flush_interval: float = DEFAULT_AIOMISC_LOG_FLUSH,
         log_config: bool = DEFAULT_AIOMISC_LOG_CONFIG,
+        log_handlers: Iterable[logging.Handler] = (),
         policy: asyncio.AbstractEventLoopPolicy = event_loop_policy,
         debug: bool = DEFAULT_AIOMISC_DEBUG,
         catch_signals: Optional[Tuple[int, ...]] = None,
@@ -196,6 +198,7 @@ def __init__(
         self.log_date_format = log_date_format
         self.log_flush_interval = log_flush_interval
         self.log_format = log_format
+        self.log_handlers = tuple(log_handlers)
         self.log_level = log_level
         self.policy = policy
 

aiomisc/log.py

@@ -1,45 +1,82 @@
 import asyncio
-import atexit
-import logging
 import logging.handlers
-import time
+import threading
 import traceback
+import warnings
 from contextlib import suppress
-from functools import partial
+from queue import Empty, Queue
 from socket import socket
 from typing import (
     Any, Callable, Dict, Iterable, List, Optional, Tuple, Type, Union,
 )
-from weakref import finalize
 
 import aiomisc_log
 from aiomisc_log.enum import LogFormat, LogLevel
 
-from .thread_pool import run_in_new_thread
-
-
-def _thread_flusher(
-    handler: logging.handlers.MemoryHandler,
-    flush_interval: Union[float, int],
-    loop: asyncio.AbstractEventLoop,
-) -> None:
-    def has_no_target() -> bool:
-        return True
-
-    def has_target() -> bool:
-        return bool(handler.target)
-
-    is_target = has_no_target
-
-    if isinstance(handler, logging.handlers.MemoryHandler):
-        is_target = has_target
-
-    while not loop.is_closed() and is_target():
-        with suppress(Exception):
-            if handler.buffer:
-                handler.flush()
-
-        time.sleep(flush_interval)
+from .counters import Statistic
+
+
+class ThreadedHandlerStatistic(Statistic):
+    threads: int
+    records: int
+    errors: int
+    flushes: int
+
+
+class ThreadedHandler(logging.Handler):
+    def __init__(
+        self, target: logging.Handler, flush_interval: float = 0.1,
+        buffered: bool = True, queue_size: int = 0,
+    ):
+        super().__init__()
+        self._buffered = buffered
+        self._target = target
+        self._flush_interval = flush_interval
+        self._flush_event = threading.Event()
+        self._queue: Queue[Optional[logging.LogRecord]] = Queue(queue_size)
+        self._close_event = threading.Event()
+        self._thread = threading.Thread(target=self._in_thread, daemon=True)
+        self._statistic = ThreadedHandlerStatistic()
+
+    def start(self) -> None:
+        self._statistic.threads += 1
+        self._thread.start()
+
+    def close(self) -> None:
+        self._queue.put(None)
+        del self._queue
+        self.flush()
+        self._close_event.set()
+        super().close()
+
+    def flush(self) -> None:
+        self._statistic.flushes += 1
+        self._flush_event.set()
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if self._buffered:
+            self._queue.put_nowait(record)
+        else:
+            self._queue.put(record)
+        self._statistic.records += 1
+
+    def _in_thread(self) -> None:
+        queue = self._queue
+        while not self._close_event.is_set():
+            self._flush_event.wait(self._flush_interval)
+            try:
+                self.acquire()
+                while True:
+                    record = queue.get(timeout=self._flush_interval)
+                    if record is None:
+                        return
+                    with suppress(Exception):
+                        self._target.handle(record)
+            except Empty:
+                pass
+            finally:
+                self.release()
+        self._statistic.threads -= 1
 
 
 def suppressor(
@@ -54,27 +91,18 @@ def wrapper() -> None:
 
 def wrap_logging_handler(
     handler: logging.Handler,
-    loop: Optional[asyncio.AbstractEventLoop] = None,
     buffer_size: int = 1024,
     flush_interval: Union[float, int] = 0.1,
+    loop: Optional[asyncio.AbstractEventLoop] = None,
 ) -> logging.Handler:
-    buffered_handler = logging.handlers.MemoryHandler(
-        buffer_size,
+    warnings.warn("wrap_logging_handler is deprecated", DeprecationWarning)
+    handler = ThreadedHandler(
         target=handler,
-        flushLevel=logging.CRITICAL,
-    )
-
-    run_in_new_thread(
-        _thread_flusher, args=(
-            buffered_handler, flush_interval, loop,
-        ), no_return=True, statistic_name="logger",
+        queue_size=buffer_size,
+        flush_interval=flush_interval,
     )
-
-    at_exit_flusher = suppressor(handler.flush)
-    atexit.register(at_exit_flusher)
-    finalize(buffered_handler, partial(atexit.unregister, at_exit_flusher))
-
-    return buffered_handler
+    handler.start()
+    return handler
 
 
 class UnhandledLoopHook(aiomisc_log.UnhandledHook):
@@ -109,7 +137,7 @@ def __call__(
         protocol: Optional[asyncio.Protocol] = context.pop("protocol", None)
         transport: Optional[asyncio.Transport] = context.pop("transport", None)
         sock: Optional[socket] = context.pop("socket", None)
-        source_traceback: List[traceback.FrameSummary] = (
+        source_tb: List[traceback.FrameSummary] = (
             context.pop("source_traceback", None) or []
         )
 
@@ -129,51 +157,52 @@ def __call__(
 
         self._fill_transport_extra(transport, extra)
         self.logger.exception(message, exc_info=exception, extra=extra)
-        if source_traceback:
-            self.logger.error(
-                "".join(traceback.format_list(source_traceback)),
-            )
+        if source_tb:
+            self.logger.error("".join(traceback.format_list(source_tb)))
 
 
 def basic_config(
     level: Union[int, str] = LogLevel.default(),
     log_format: Union[str, LogFormat] = LogFormat.default(),
-    buffered: bool = True, buffer_size: int = 1024,
+    buffered: bool = True,
+    buffer_size: int = 0,
     flush_interval: Union[int, float] = 0.2,
     loop: Optional[asyncio.AbstractEventLoop] = None,
     handlers: Iterable[logging.Handler] = (),
     **kwargs: Any,
 ) -> None:
-    loop = loop or asyncio.get_event_loop()
     unhandled_hook = UnhandledLoopHook(logger_name="asyncio.unhandled")
 
-    def wrap_handler(handler: logging.Handler) -> logging.Handler:
-        nonlocal buffer_size, buffered, loop, unhandled_hook
+    if loop is None:
+        loop = asyncio.get_event_loop()
 
-        unhandled_hook.add_handler(handler)
+    forever_task = asyncio.gather(
+        loop.create_future(), return_exceptions=True,
+    )
+    loop.set_exception_handler(unhandled_hook)
 
-        if buffered:
-            return wrap_logging_handler(
-                handler=handler,
-                buffer_size=buffer_size,
-                flush_interval=flush_interval,
-                loop=loop,
-            )
-        return handler
+    log_handlers = []
+
+    for user_handler in handlers:
+        handler = ThreadedHandler(
+            buffered=buffered,
+            flush_interval=flush_interval,
+            queue_size=buffer_size,
+            target=user_handler,
+        )
+        unhandled_hook.add_handler(handler)
+        forever_task.add_done_callback(lambda _: handler.close())
+        log_handlers.append(handler)
+        handler.start()
 
     aiomisc_log.basic_config(
-        level=level,
-        log_format=log_format,
-        handler_wrapper=wrap_handler,
-        handlers=handlers,
-        **kwargs,
+        level=level, log_format=log_format, handlers=log_handlers, **kwargs,
     )
 
-    loop.set_exception_handler(unhandled_hook)
-
 
 __all__ = (
     "LogFormat",
     "LogLevel",
     "basic_config",
+    "ThreadedHandler",
 )

aiomisc/service/uvicorn.py

@@ -117,7 +117,7 @@ async def start(self) -> Any:
             self.sock = config.bind_socket()
         self.server = Server(config)
         self.serve_task = asyncio.create_task(
-            self.server.serve(sockets=[self.sock])
+            self.server.serve(sockets=[self.sock]),
         )
 
     async def stop(self, exception: Optional[Exception] = None) -> None:

aiomisc/timeout.py

@@ -16,7 +16,7 @@
 
 
 def timeout(
-    value: Number
+    value: Number,
 ) -> Callable[
     [Callable[P, Coroutine[Any, Any, T]]],
     Callable[P, Coroutine[Any, Any, T]],

aiomisc_log/__init__.py

@@ -125,7 +125,6 @@ def basic_config(
     handlers: Iterable[logging.Handler] = (),
     **kwargs: Any,
 ) -> None:
-
     if isinstance(level, str):
         level = LogLevel[level]