为什么附加到根记录器的过滤器不传播到后代记录器？

Question

来自python logging documentation 的第 15.7.4 段：

请注意，每当事件发生时，都会查询附加到处理程序的过滤器 由处理程序发出，而附加到记录器的过滤器是 每当将事件记录到处理程序时都会进行咨询（使用 debug()， info() 等）这意味着由 后代记录器不会被记录器的过滤器设置过滤， 除非过滤器也已应用于那些后代记录器。

我不明白这个设计决定。将根 logger 的过滤器也应用于后代 logger 是否更有意义？

Answer 1

我同意：这是一个违反直觉的设计决策，恕我直言。

最简单的解决方案是将过滤器附加到每个可能的处理程序。例如，假设您有一个控制台处理程序、一个邮件处理程序和一个数据库处理程序，您应该将您的“根”过滤器附加到它们中的每一个。 :-/

import logging
import logging.config

class MyRootFilter(logging.Filter):
    def filter(self, record):
        # filter out log messages that include "secret"
        if "secret" in record.msg:
            return False
        else:
            return True

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'filters': {
        'my_root_filter': {
            '()': MyRootFilter,
        },
    },
    'handlers': {
        'stderr': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
            'filters': ['my_root_filter'],
        },
        'mail_admins': {
            'level': 'ERROR',
            'class': 'some.kind.of.EmailHandler',
            'filters': ['my_root_filter'],
        },
        'database': {
            'level': 'ERROR',
            'class': 'some.kind.of.DatabaseHandler',
            'filters': ['my_root_filter'],
        },
    },
    'loggers': {
        'some.sub.project': {
            'handlers': ['stderr'],
            'level': 'ERROR',
        },
    },
}

logging.config.dictConfig(LOGGING)
logging.getLogger("some.sub.project").error("hello")        # logs 'hello'
logging.getLogger("some.sub.project").error("hello secret") # filtered out!  :-)

如果有很多处理程序，您可能希望以编程方式而不是手动将根过滤器附加到每个处理程序。我建议您直接在配置字典（或文件，具体取决于您加载日志记录配置的方式）上执行此操作，而不是在加载配置之后执行此操作，因为似乎没有记录的方法获取所有处理程序的列表。我找到了 logger.handlers 和 logging._handlers，但由于它们没有记录在案，它们将来可能会中断。另外，不能保证它们是线程安全的。

之前的解决方案（在加载之前将根过滤器直接附加到配置中的每个处理程序）假设您在加载之前可以控制日志记录配置，并且不会动态添加处理程序（使用 Logger#添加处理程序（））。如果这不是真的，那么您可能需要对日志记录模块进行猴子补丁（祝您好运！）。

编辑

我拍摄了猴子修补 Logger#addHandler 的照片，只是为了好玩。它实际上工作正常并简化了配置，但我不确定我是否会推荐这样做（我讨厌猴子补丁，当出现问题时它很难调试）。使用风险自负...

import logging
import logging.config

class MyRootFilter(logging.Filter):
   [...] # same as above

LOGGING = {
    'version': 1,
    'disable_existing_loggers': False,
    'handlers': {
        'stderr': {
            'level': 'DEBUG',
            'class': 'logging.StreamHandler',
            # it's shorter: there's no explicit reference to the root filter
        },
        [...]  # other handlers go here
    },
    'loggers': {
        'some.sub.project': {
            'handlers': ['stderr'],
            'level': 'ERROR',
        },
    },
}

def monkey_patched_addHandler(self, handler):
    result = self.old_addHandler(handler)
    self.addFilter(MyRootFilter())
    return result

logging.Logger.old_addHandler = logging.Logger.addHandler
logging.Logger.addHandler = monkey_patched_addHandler

logging.config.dictConfig(LOGGING)
logging.getLogger("some.sub.project").error("hello")        # logs 'hello'
logging.getLogger("some.sub.project").error("hello secret") # filtered out!  :-)

Answer 2

与级别和处理程序不同，过滤器不会传播。但是，您可以将它附加到 将 传播的处理程序。

通过Elliot at SaltyCrane Blog查看此示例：

import logging

class MyFilter(logging.Filter):
    def filter(self, record):
        record.msg = 'MY FILTER: ' + record.msg
        return 1

myfilter = MyFilter()

myformatter = logging.Formatter("MY HANDLER: %(name)s - %(message)s")

myhandler = logging.StreamHandler()
myhandler.setFormatter(myformatter)
myhandler.addFilter(myfilter)

foo_logger = logging.getLogger('foo')
foo_logger.addHandler(myhandler)

foo_bar_logger = logging.getLogger('foo.bar')

foo_logger.error('asdfasdf')
foo_bar_logger.error('zxcvzxcv')

MY HANDLER: foo - MY FILTER: asdfasdf
MY HANDLER: foo.bar - MY FILTER: zxcvzxcv

Answer 3

这样想。记录器就像你家的排水管。记录器上的过滤器可以防止您将东西倾倒到下水道中，它不会过滤整个下水道。您在流程中的位置（上游、下游）不会改变这种行为。

处理程序是管道。管道积聚上游流量。默认管道是“跳过，传递给父级”。如果要影响上游流量，则需要在管道（处理程序）上放置一个过滤器。如果您查看logging flowchart，应该可以添加一个NullHandler（无格式或输出）过滤然后传播消息。

这是你想要的行为。

Answer 4

这里类似于 MiniQuark 是猴子修补的一种优雅方式（嵌入在上下文管理器中）：

from typing import Iterable, Iterator
from contextlib import contextmanager


@contextmanager
def apply_filter_for_new_handlers(filters: Iterable[logging.Filter]) -> Iterator[None]:
    originial_addHandler = logging.Logger.addHandler

    def monkey_patched_addHandler(self: logging.Logger, hdlr: logging.Handler) -> None: # noqa:C0103
        for log_filter in filters:
            hdlr.addFilter(log_filter)
        return originial_addHandler(self, hdlr)

    logging.Logger.addHandler = monkey_patched_addHandler
    yield

    logging.Logger.addHandler = originial_addHandler


class MyRootFilter(logging.Filter):
   [...] # same as above


with apply_filter_for_new_handlers(filters=[MyRootFilter]):
    logging.getLogger("some.sub.project").error("hello")