为Python根日志器添加Azure Blob Storage自定义Handler遇死循环问题

问题描述 #

• 基于Azure Blob Storage追加Blob实现的自定义logging.Handler，添加到根日志器时，执行logger.info("Job initiated")程序陷入卡死状态，CPU持续高占用；Blob已创建但无内容写入。
• 单个日志消息会触发emit方法反复调用，append_block()后的代码从未执行。
• 仅将该Handler添加到独立命名应用日志器时可正常工作，但导入模块的日志无法传播至该应用级日志器。

自定义Handler代码 #

class AzureBlobStorageHandler(logging.Handler):
    def __init__(self, container_client, path=None):
        super().__init__()
        
        self.path = path
        self.container_client = container_client
        self.blob_client = None
        self.blob_name = None
        self.createBlob()


    def createBlob(self):
        # Create a blob client for the log record
        now = datetime.datetime.now().strftime("%Y-%m-%dT%H%M%S")
        if self.path:
            self.blob_name  = "{}/{}.log".format(self.path, now)
        else:
            self.blob_name  = "{}.log".format(now)

        self.blob_client = self.container_client.get_blob_client(self.blob_name)
        content_settings = ContentSettings(content_type="text/plain")
        r = self.blob_client.create_append_blob(content_settings)

    
    def emit(self, record):
        # Write the log record to the blob
        log_data = self.format(record).encode("utf-8")
        self.blob_client.append_block(log_data)

主应用日志初始化代码 #

import logging
logger = logging.getLogger()

import json
import datetime
from os.path import exists as os_path_exists
from azure.storage.blob import ContainerClient, ContentSettings


conf = json.load(open("conf.json", mode="r"))
if os_path_exists("..\conf\conf_global.json"):
    conf2 = json.load(open("..\conf\conf_global.json", mode="r"))
    conf.update(conf2)

#Create an Azure Blob Container Client 
container_client = ContainerClient(
    account_url=conf["persistence"]["AZBlobStore"]["account_url"], 
    container_name=conf["persistence"]["AZBlobStore"]["container_name"],
    credential=conf["persistence"]["AZBlobStore"]["sas_credential"]
)

# Add the Azure Blob Storage handler to the logger
level = logging.INFO

handler = AzureBlobStorageHandler(container_client, path=logging_path)
handler.setLevel(level)

frmt = "%(asctime)s | %(levelname)s | in %(name)s | %(message)s\n"
time_format_str = "%Y-%m-%dT%H:%M:%S"
formatter = logging.Formatter(frmt, time_format_str)
handler.setFormatter(formatter)

logger.addHandler(handler)
logger.setLevel(level)

解决思路与方案 #

核心原因 #

根日志器会捕获全程序所有日志，包括Azure SDK内部执行append_block()时输出的日志。这些SDK日志会再次触发自定义Handler的emit方法，形成无限递归死循环，导致CPU占用飙升、程序卡死。

具体解决方案 #

1. 过滤Azure SDK日志，阻止递归触发

在初始化根日志器后，直接禁用Azure相关日志器的传播，或设置更高日志级别：

# 禁用Azure SDK日志的传播，避免被根日志器捕获
logging.getLogger("azure").propagate = False
logging.getLogger("azure.storage").propagate = False

# 或者仅记录Azure SDK的严重错误，减少日志量
logging.getLogger("azure").setLevel(logging.ERROR)

2. 在Handler的emit方法中添加递归防护

直接在emit中跳过Azure相关日志的处理：

def emit(self, record):
    # 跳过Azure SDK产生的日志，避免递归
    if record.name.startswith("azure"):
        return
    
    log_data = self.format(record).encode("utf-8")
    try:
        self.blob_client.append_block(log_data)
    except Exception as e:
        # 捕获异常时使用print而非日志，避免再次触发递归
        print(f"日志写入失败: {str(e)}")

3. 调整日志器策略，兼顾模块日志收集

如果需要捕获导入模块的日志，可使用命名日志器并强制开启传播，同时给根日志器添加Handler（已做Azure日志过滤）：

# 创建应用级命名日志器
app_logger = logging.getLogger("my_application")
app_logger.setLevel(logging.INFO)
app_logger.propagate = True  # 强制将日志传播到根日志器

# 根日志器配置（已添加过滤Azure日志的逻辑）
root_logger = logging.getLogger()
root_logger.addHandler(handler)
root_logger.setLevel(logging.INFO)

4. 异步执行日志写入，避免主线程阻塞

Azure SDK的append_block()是同步方法，网络延迟可能加剧阻塞问题，可改用线程池异步执行：

import concurrent.futures

class AzureBlobStorageHandler(logging.Handler):
    def __init__(self, container_client, path=None):
        super().__init__()
        # 初始化线程池，异步执行日志写入
        self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=2)
        self.path = path
        self.container_client = container_client
        self.blob_client = None
        self.blob_name = None
        self.createBlob()

    # 省略createBlob方法...

    def emit(self, record):
        if record.name.startswith("azure"):
            return
        log_data = self.format(record).encode("utf-8")
        # 提交异步任务，不阻塞主线程
        self.executor.submit(self._write_log, log_data)

    def _write_log(self, log_data):
        try:
            self.blob_client.append_block(log_data)
        except Exception as e:
            print(f"日志追加失败: {str(e)}")

内容的提问来源于stack exchange，提问作者BioData41