Linux消费计划Azure Function(Python)无法将日志保存至Blob容器

方案1：直接使用Azure Blob SDK写入日志 #

绕过本地文件系统，通过Azure Blob Storage的Python SDK直接将日志内容写入指定容器，是最直接的持久化方式。

步骤 #

1. 在requirements.txt中添加依赖：azure-storage-blob>=12.0.0
2. 编写日志写入逻辑，优先用托管身份（避免硬编码连接字符串）：

import logging
from azure.storage.blob import BlobServiceClient, AppendBlobClient
import os
from datetime import datetime

def get_append_blob_client():
    account_url = f"https://{os.environ['AZURE_STORAGE_ACCOUNT']}.blob.core.windows.net"
    # 用Function App的系统分配托管身份初始化客户端
    blob_service_client = BlobServiceClient(account_url=account_url)
    container_client = blob_service_client.get_container_client(os.environ['LOG_CONTAINER_NAME'])
    # 按日期生成日志文件名，避免单个Blob过大
    log_filename = f"{datetime.now().strftime('%Y-%m-%d')}_function.log"
    return container_client.get_append_blob_client(log_filename)

def write_log(message):
    try:
        append_blob_client = get_append_blob_client()
        # 若Blob不存在则创建
        if not append_blob_client.exists():
            append_blob_client.create_append_blob()
        # 写入带时间戳的日志内容
        log_entry = f"{datetime.now().isoformat()} - {message}\n"
        append_blob_client.append_block(log_entry.encode('utf-8'))
    except Exception as e:
        logging.error(f"日志写入Blob失败: {str(e)}")

# 在Function中调用示例
def main(req: func.HttpRequest) -> func.HttpResponse:
    write_log("函数开始处理请求")
    # 业务逻辑代码...
    write_log("函数处理请求完成")
    return func.HttpResponse("处理成功")

注意事项 #

• 给Function App的系统分配身份授予目标Blob容器的存储Blob数据参与者权限
• 按日期拆分日志文件，方便后续查询和管理

方案2：自定义Logging Handler适配Blob #

通过自定义Python Logging Handler，让现有logging框架直接将日志导向Blob Storage，无需大量修改业务代码。

import logging
from datetime import datetime
from azure.storage.blob import BlobServiceClient, AppendBlobClient
import os
from logging import Handler, LogRecord

class BlobStorageAppendHandler(Handler):
    def __init__(self, account_name, container_name):
        super().__init__()
        self.account_name = account_name
        self.container_name = container_name
        self.account_url = f"https://{account_name}.blob.core.windows.net"
        self.blob_service_client = BlobServiceClient(account_url=self.account_url)
        self.container_client = self.blob_service_client.get_container_client(container_name)
        self.current_blob_name = None
        self.append_blob_client = None

    def _get_blob_client(self):
        today = datetime.now().strftime('%Y-%m-%d')
        blob_name = f"{today}_function.log"
        if blob_name != self.current_blob_name:
            self.current_blob_name = blob_name
            self.append_blob_client = self.container_client.get_append_blob_client(blob_name)
            if not self.append_blob_client.exists():
                self.append_blob_client.create_append_blob()
        return self.append_blob_client

    def emit(self, record: LogRecord):
        try:
            log_entry = self.format(record) + "\n"
            blob_client = self._get_blob_client()
            blob_client.append_block(log_entry.encode('utf-8'))
        except Exception as e:
            self.handleError(record)

# 初始化Logging配置
def setup_logging():
    logger = logging.getLogger()
    logger.setLevel(logging.INFO)
    # 移除默认Handler避免重复输出
    for handler in logger.handlers[:]:
        logger.removeHandler(handler)
    # 添加自定义Blob Handler
    blob_handler = BlobStorageAppendHandler(
        account_name=os.environ['AZURE_STORAGE_ACCOUNT'],
        container_name=os.environ['LOG_CONTAINER_NAME']
    )
    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    blob_handler.setFormatter(formatter)
    logger.addHandler(blob_handler)

# 在Function启动时执行配置
setup_logging()

# 业务代码中直接用logging模块
def main(req: func.HttpRequest) -> func.HttpResponse:
    logging.info("函数启动")
    # 业务逻辑...
    logging.error("处理过程中出现错误")
    return func.HttpResponse("处理完成")

方案3：Azure Monitor集成（推荐） #

如果不需要独立日志文件，可直接用原生集成能力：

• Linux消费计划的Function日志默认会同步到Azure Monitor Log Analytics工作区
• 在Azure门户的「Function App -> 监控 -> 日志」中可查询所有日志
• 可通过Log Analytics的导出规则，将日志持久化到指定Blob容器，支持按时间自动拆分

优势 #

• 无需额外代码开发
• 支持结构化查询、告警和可视化
• 自带日志生命周期管理能力

常见问题说明 #

• 直接用logging.basicConfig指定Blob URL失败：Logging的FileHandler仅支持本地文件路径，无法识别Blob URL，因此报FileNotFoundError
• /temp目录无法持久化：该目录是实例级临时存储，消费计划实例动态伸缩销毁后，内容会丢失，不适合持久化日志
• Linux消费计划查看临时目录：可通过「高级工具 -> SSH」连接到实例查看，但临时存储本身不适合长期存储

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