Python 日志记录器logging 百科全书 之 日志过滤

前言

在Python的logging模块中，日志过滤器（Filter）用于提供更细粒度的日志控制。通过过滤器，我们可以决定哪些日志记录应该被输出，哪些应该被忽略。这对于复杂的应用程序来说尤为重要，因为它允许我们根据日志记录的特定属性（例如日志级别、日志记录者的名称或日志记录包含的消息、上下文等）来控制日志的输出。

知识点📖📖

import logging

文章脉络：

• 点击直达：万字长文 - Python 日志记录器logging 百科全书 之 基础配置
• 点击直达：Python 日志记录器logging 百科全书 之 日志回滚
• 点击直达：【万字长文】Python 日志记录器logging 百科全书 之 日志过滤

日志过滤简介

1. 疑惑和应用场景

可能读者朋友们会有疑惑，日志记录和日志过滤都是人为的操作，既要记录，又要过滤，那为什么不在一开始就只记录过滤后的信息呢，那是否构成悖论呢？

以下是一些关于为什么需要日志过滤的考虑以及适用的应用场景：

• 动态环境： 根据信息的重要性记录不同级别（如DEBUG、INFO、ERROR）的日志，如在调试期间记录详细的信息，但在生产环境中记录更少的信息。
• 多个处理程序：如果应用程序同时将日志发送到不同的处理程序（例如，将日志同时写入文件和发送到远程服务器），每个处理程序可能需要记录不同级别的信息。这时需要使用过滤器来控制每个处理程序的输出。
• 隐私保护： 为了保护敏感信息的安全，过滤掉包含敏感信息的日志是必要的，以确保不会意外泄露。
• …

2. 日志过滤的原理

这里解释日志过滤是如何工作的，包括当日志消息被发送到处理程序前，过滤器如何干预日志记录的流程。

日志过滤的原理涉及到以下关键概念：

• 日志记录器（Logger）：在Python的logging模块中，日志记录器是用于创建和处理日志消息的对象。每个日志记录器通常与一个特定的模块或组件相关联。
• 日志处理程序（Handler）：处理程序决定了日志消息的最终去向，例如输出到控制台、写入文件等。处理程序通常与一个或多个日志记录器关联。
• 日志过滤器（Filter）：过滤器是一个可选组件，它可以附加到处理程序上，用于决定哪些日志消息应该被处理，哪些应该被忽略。过滤器在日志消息被发送到处理程序之前起作用。
• 过滤规则：过滤器根据一组规则或条件来过滤日志消息。这些规则可以包括日志级别、日志记录器名称、关键词等。如果日志消息满足过滤规则，它将被允许传递到处理程序；否则，它将被忽略。
• 日志级别：日志消息通常具有不同的级别，如DEBUG、INFO、WARNING、ERROR 和 CRITICAL。通过设置适当的过滤规则，我们可以选择记录特定级别的消息。

过滤器添加到日志记录器：

• 用于控制哪些记录器产生的消息应该被记录，记录器过滤器影响记录器级别

过滤器添加到日志处理程序：

• 用于控制哪些消息应该发送到特定处理程序，处理程序过滤器影响处理程序级别。

当日志消息被发送到处理程序时，会首先经过与之关联的过滤器。如果消息通过了过滤器的检查，它将被处理，否则将被丢弃。这样，过滤器可以在日志记录的不同阶段对消息进行筛选，以确保只有符合条件的消息被记录。

总之，日志过滤通过过滤器提供了一种强大的机制，允许开发人员有选择地记录和处理日志消息，以满足应用程序的特定需求和调试要求。

日志过滤器

logging 的日志过滤一般是基于重写 logging.Filter 类来实现自定义过滤器。

logging.Filter

步骤如下：

1. 创建自定义过滤器类，继承 logging.Filter 且重写 filter 方法
2. 创建日志记录器 和 日志处理器
3. 添加自定义过滤器类到日志处理器
4. 设置日志格式
5. 添加处理器（Handler）到记录器（Logger）
6. 记录日志~

代码示例：

import logging


# 步骤1：创建自定义过滤器类
class MyFilter(logging.Filter):
    def filter(self, record):
        # 在这里编写过滤逻辑
        # 返回True表示允许消息通过过滤器，返回False表示不允许
        return record.levelno >= logging.WARNING  # 只允许WARNING级别及以上的消息通过


# 步骤2：创建日志记录器和日志处理器
logger = logging.getLogger("my_logger")  # 创建记录器
logger.setLevel(logging.DEBUG)  # 设置记录器的级别为DEBUG

handler = logging.StreamHandler()  # 创建处理器（这里使用了StreamHandler，将日志消息输出到控制台）
handler.setLevel(logging.DEBUG)  # 设置处理器的级别为DEBUG

# 步骤3：添加自定义过滤器到处理器
my_filter = MyFilter()
handler.addFilter(my_filter)

# 步骤4：设置日志格式
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
handler.setFormatter(formatter)

# 步骤5：添加处理器到记录器
logger.addHandler(handler)

# 步骤6：记录日志
logger.debug("This is a debug message")  # 这条消息不会被过滤器通过
logger.warning("This is a warning message")  # 这条消息会被过滤器通过

# 2023-11-14 00:28:28,714 - my_logger - WARNING - This is a warning message

代码释义：

上述代码演示了如何创建一个自定义过滤器类（步骤1），将它添加到一个处理器（步骤3），并将处理器添加到记录器（步骤5）。在这个示例中，自定义过滤器MyFilter允许通过级别为logging.WARNING及以上的日志消息。所以，在最后两行日志记录中，只有警告消息被记录。其他级别的消息被过滤掉。

实际开发中需要根据自己的需求定制自定义过滤器的逻辑，以满足不同的过滤条件。

日志过滤的选择

这里列举选择日志过滤的依据，如日志级别、关键字、进程&线程名、上下文、日志记录器名称等，以及如何决定何时使用哪种过滤器。

1. 日志级别

日志级别是特别常见的过滤依据，根据消息的重要性或严重性来过滤日志。常见的日志级别包括DEBUG、INFO、WARNING、ERROR、CRITICAL等。

示例代码：

class LevelFilter(logging.Filter):
    def __init__(self, level):
        self.level = level

    def filter(self, record):
        return record.levelno == self.level
    

2. 关键字

• 目的：过滤包含特定关键字的日志。
• 使用场景：当需要根据消息的内容特征来过滤日志时，可以使用关键字过滤器。例如，筛选包含特定错误代码或事件描述的日志消息。

示例代码：

class KeywordFilter(logging.Filter):
    def __init__(self, keyword):
        self.keyword = keyword

    def filter(self, record):
        return self.keyword in record.getMessage()

3. 进程名 & 线程名

根据线程&进程名称来过滤日志，以区分不同的应用程序实例或进程。

应用场景：

下面的两份代码基本一致。

进程名示例代码：

• 只记录线程名称为Process 1的日志信息

import logging
import multiprocessing


class ProcessNameFilter(logging.Filter):
    def __init__(self, process_name):
        super().__init__()
        self.process_name = process_name

    def filter(self, record):
        return record.processName == self.process_name


# 创建日志记录器
logger = logging.getLogger("my_logger")
logger.setLevel(logging.DEBUG)

# 创建处理器（这里使用了StreamHandler，将日志消息输出到控制台）
handler = logging.StreamHandler()

# 设置日志格式
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
handler.setFormatter(formatter)

# 添加处理器到记录器
logger.addHandler(handler)

# 添加自定义过滤器到处理器
thread_filter = ProcessNameFilter('Process 1')
handler.addFilter(thread_filter)


def worker_function():
    # 获取当前进程的名称
    current_process_name = multiprocessing.current_process().name

    # 记录日志
    logger.debug(f"This is a debug message {current_process_name}")
    logger.info(f"This is an info message {current_process_name}")
    logger.warning(f"This is a warning message {current_process_name}")


if __name__ == "__main__":
    process_tasks = [multiprocessing.Process(target=worker_function, name=f'Process {i}') for i in range(5)]
    [process.start() for process in process_tasks]
    [process.join() for process in process_tasks]

线程名示例代码：

• 只记录线程名称为Thread 1的日志信息

import logging
import threading


class ThreadNameFilter(logging.Filter):
    def __init__(self, thread_name):
        super().__init__()
        self.thread_name = thread_name

    def filter(self, record):
        return record.threadName == self.thread_name


# 创建日志记录器
logger = logging.getLogger("my_logger")
logger.setLevel(logging.DEBUG)

# 创建处理器（这里使用了StreamHandler，将日志消息输出到控制台）
handler = logging.StreamHandler()

# 设置日志格式
formatter = logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
handler.setFormatter(formatter)

# 添加处理器到记录器
logger.addHandler(handler)

# 添加自定义过滤器到处理器
thread_filter = ThreadNameFilter('Thread 1')
handler.addFilter(thread_filter)


def worker_function():
    # 获取当前线程的名称
    current_thread_name = threading.current_thread().name

    # 记录日志
    logger.debug(f"This is a debug message {current_thread_name}")
    logger.info(f"This is an info message {current_thread_name}")
    logger.warning(f"This is a warning message {current_thread_name}")


if __name__ == '__main__':
    thread_tasks = [threading.Thread(target=worker_function, name=f'Thread {i}') for i in range(5)]
    [thread.start() for thread in thread_tasks]
    [thread.join() for thread in thread_tasks]

4. 上下文

在日志记录中，上下文信息是一种关键的元素，它允许我们更精细地控制哪些日志消息应该被记录和处理。上下文信息可以是与日志消息相关的任何附加信息，常见的用例包括：

下面使用一个简单的案例来展示日志过滤中上下文 的应用。

示例代码：

• 下面代码演示了如何使用自定义过滤器和额外的上下文信息来过滤日志消息。

import logging


class ContextFilter(logging.Filter):
    def __init__(self, context):
        super().__init__()
        self.context: dict = context

    def filter(self, record):
        return getattr(record, 'user_id', None) == self.context.get('user_id')


# 创建日志记录器
logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

# 创建处理器（这里使用了StreamHandler，将日志消息输出到控制台）
handler = logging.StreamHandler()

# 设置日志格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s - user_id:%(user_id)s')
handler.setFormatter(formatter)

# 添加处理器到记录器
logger.addHandler(handler)

# 添加自定义过滤器到处理器
# 创建上下文信息
context_info = {'user_id': 'user123', 'request_id': 'abcdef'}
thread_filter = ContextFilter(context_info)
handler.addFilter(thread_filter)

# 记录日志
# 这个日志将被显示，因为它匹配了上下文过滤器
logger.debug('This is a debug message', extra={'user_id': 'user123'})
# 这个日志将不会被显示
logger.debug('This log is from another user', extra={'username': 'user456'})

代码释义：

在上面的示例中，ContextFilter 过滤器根据user_id来过滤日志，只有当日志消息中的user_id与上下文信息中的匹配时，日志消息才会被记录。这种方法允许根据自定义上下文信息轻松地过滤和分析日志消息。

5. 日志记录名称

• 目的：根据生成它们的日志记录器的名称来过滤日志。
• 使用场景：为不同的模块或子系统创建不同的记录器，并为每个记录器添加不同的过滤器，以根据记录器的名称来过滤消息。

示例代码：

class LoggerNameFilter(logging.Filter):
    def __init__(self, logger_name):
        self.logger_name = logger_name

    def filter(self, record):
        return record.name == self.logger_name

6. 应用场景

简单总结一下关于日志过滤的几种选择依据的应用场景。

示例代码

这份代码是一个多功能的日志配置示例，适用于需要精确控制日志输出的应用程序。它适合于那些需要在不同环境（如开发和生产环境）下进行不同日志处理的场景。

代码

# -*- coding: utf-8 -*-


import logging


# 自定义过滤器 - 控制台使用
class ConsoleFilter(logging.Filter):
    def filter(self, record):
        # 过滤掉包含敏感信息的日志
        return "敏感信息" not in record.getMessage()


# 自定义过滤器 - 文件处理器使用
class FileFilter(logging.Filter):
    def filter(self, record):
        # 过滤掉上下文不匹配的日志
        if hasattr(record, 'context_id') and record.context_id != 'expected_context':
            return False
        return True


def setup_logger(name, log_file, level=logging.DEBUG):
    """配置日志记录器、处理器和过滤器"""
    _logger = logging.getLogger(name=name)
    _logger.setLevel(level=level)

    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_handler.addFilter(ConsoleFilter())

    # 文件处理器
    file_handler = logging.FileHandler(filename=log_file, encoding='utf-8', delay=True)
    file_handler.setLevel(logging.INFO)
    file_handler.addFilter(FileFilter())

    # 设置日志格式
    formatter = logging.Formatter("%(levelname)-7s - %(asctime)s - %(name)s -  %(message)s")
    console_handler.setFormatter(formatter)
    file_handler.setFormatter(formatter)

    # 添加处理器到记录器
    _logger.addHandler(console_handler)
    _logger.addHandler(file_handler)

    return _logger


if __name__ == '__main__':
    # 设置日志记录器
    logger = setup_logger("app_logger", "app.log")

    # 记录日志，包含上下文信息
    logger.debug("这是一条调试消息", extra={'context_id': 'expected_context'})  # 控制台打印, 日志不记录
    logger.info("这条消息包含敏感信息：123456", extra={'context_id': 'expected_context'})  # 控制台不打印, 日志记录
    logger.info("这是一条普通消息", extra={'context_id': 'expected_context'})  # 控制台打印, 日志记录
    logger.warning("这是一条警告消息", extra={'context_id': 'unexpected_context'})  # 控制台打印, 日志不记录
    logger.error("这是一条错误消息！", extra={'context_id': 'expected_context'})  # 控制台打印, 日志记录

代码释义：

1. 自定义过滤器：
   • ConsoleFilter：用于控制台处理器，过滤掉包含“敏感信息”的日志。
   • FileFilter：用于文件处理器，过滤掉上下文标识符（context_id）与"expected_context"不匹配的日志。
2. 日志记录器配置（setup_logger 函数）：
   • 配置一个名为 name 的日志记录器。
   • 设置两个处理器：控制台处理器（console_handler）和文件处理器（file_handler）。
   • 控制台处理器设置为 DEBUG 级别，文件处理器设置为 INFO 级别。
   • 为处理器添加对应的过滤器和格式器。
3. 日志记录操作：
   • 使用 logger 记录不同级别和内容的日志。
   • 根据过滤器的设置，不同的日志消息会被不同地处理：
     • 包含敏感信息的日志不会在控制台打印，但会记录到文件中。
     • 上下文不匹配的日志会在控制台打印，但不会记录到文件中。
     • 其他日志根据其级别既会打印到控制台也会记录到文件中。

代码运行结果如下：

如果没出错的话，日志文件中会显示如下：

总结🚀🚀

在本篇文章中，我们深入探讨了logging模块中的日志过滤功能，展示了其在大型和复杂应用程序中的关键作用。日志过滤器提供了精细的控制机制，允许开发者基于特定条件（如日志级别、关键字、进程&线程名、上下文及日志记录器名称）筛选和处理日志记录。

有效地使用日志过滤器不仅帮助开发者更好地调试和监控应用程序，而且还有助于保护敏感信息，提高应用性能和可维护性。

后话

本次分享到此结束，

see you~😎😎