使用OpenTelemetry SDK Python推送GitHub工作流日志至GCP时出现重复条目

问题场景 #

解析GitHub Workflow作业/步骤日志的TXT文件，通过代码发送至Google Cloud Logs后，每条日志消息在Logs Explorer中出现多个重复条目，仅insertId不同。

原代码 #

import logging
import google.cloud.logging
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
import dateutil.parser as dp

def get_logger(resource, name):

    client = google.cloud.logging.Client()
    client.setup_logging()

    logger = logging.getLogger(str(name))
    logger.handlers.clear()
    logger_provider = LoggerProvider(resource=resource)
    handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider)
    logger.addHandler(handler)
    return logger

resource_attributes ={"github.source": "github-exporter","workflow_run_id": GHA_RUN_ID}

resource_log = Resource(attributes=resource_attributes)
job_logger = get_logger(resource_log, "job_logger")

with open ("./logs/log1.txt") as f:
    for line in f.readlines():
        try:
            line_to_add = line[29:-1].strip()
            len_line_to_add = len(line_to_add)
            timestamp_to_add = line[0:23]
            if len_line_to_add > 0:
                # Convert ISO 8601 to timestamp
                try:
                    parsed_t = dp.isoparse(timestamp_to_add)
                except ValueError as e:
                    print("Line does not start with a date. Skip for now")
                    continue
                unix_timestamp = parsed_t.timestamp()*1000

                extra_data = {
                "log.timestamp":unix_timestamp,
                "log.time":timestamp_to_add,
                }


                if line_to_add.lower().startswith("##[debug]"):
                    job_logger._log(level=logging.DEBUG,msg=line_to_add,extra=extra_data,args="")
                else:
                    job_logger._log(level=logging.INFO,msg=line_to_add,extra=extra_data,args="")

        except Exception as e:
            print("Error exporting log line ERROR: ", e)

示例日志 #

Fri, 19 Apr 2024 12:16:29 GMT   Ubuntu
Fri, 19 Apr 2024 12:16:29 GMT   22.04.4
Fri, 19 Apr 2024 12:16:29 GMT   LTS

原因分析 #

client.setup_logging()会给root logger添加两个处理器：StreamHandler（输出到控制台）和Google Cloud专属的CloudLoggingHandler（发送到GCP日志）。而Python的logging模块中，自定义logger默认会继承root logger的所有处理器，再加上手动添加的OpenTelemetry LoggingHandler，就导致同一条日志被多个处理器重复发送到GCP。

解决方案 #

方案一：使用OpenTelemetry GCP日志导出器（推荐） #

直接用OpenTelemetry的官方GCP日志导出器，避免同时混用google-cloud-logging客户端和OTel框架，逻辑更统一。

步骤1：安装依赖

pip install opentelemetry-exporter-google-cloud-logging

步骤2：修改代码

import logging
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk._logs import LoggerProvider, LoggingHandler
from opentelemetry.exporter.cloud_logging import CloudLoggingLogExporter
import dateutil.parser as dp

def get_logger(resource, name):
    # 初始化GCP日志导出器
    exporter = CloudLoggingLogExporter()
    logger_provider = LoggerProvider(resource=resource)
    # 将导出器绑定到logger provider
    logger_provider.add_log_record_processor(exporter)
    # 创建OTel日志处理器
    handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider)
    
    logger = logging.getLogger(str(name))
    logger.handlers.clear()  # 清除可能继承的处理器
    logger.addHandler(handler)
    logger.propagate = False  # 禁止日志向上传播到root logger，彻底避免重复
    return logger

resource_attributes = {"github.source": "github-exporter", "workflow_run_id": GHA_RUN_ID}
resource_log = Resource(attributes=resource_attributes)
job_logger = get_logger(resource_log, "job_logger")

with open("./logs/log1.txt") as f:
    for line in f.readlines():
        try:
            line_to_add = line[29:-1].strip()
            len_line_to_add = len(line_to_add)
            timestamp_to_add = line[0:23]
            if len_line_to_add > 0:
                try:
                    parsed_t = dp.isoparse(timestamp_to_add)
                except ValueError:
                    print("Line does not start with a date. Skip for now")
                    continue
                unix_timestamp = parsed_t.timestamp() * 1000

                extra_data = {
                    "log.timestamp": unix_timestamp,
                    "log.time": timestamp_to_add,
                }

                # 使用标准日志方法代替私有方法_log()
                if line_to_add.lower().startswith("##[debug]"):
                    job_logger.debug(line_to_add, extra=extra_data)
                else:
                    job_logger.info(line_to_add, extra=extra_data)

        except Exception as e:
            print(f"Error exporting log line ERROR: {e}")

关键修改点

• 移除google.cloud.logging相关代码，改用OTel的CloudLoggingLogExporter直接对接GCP日志服务
• 设置logger.propagate = False，阻止日志传递到root logger，避免继承的处理器重复处理
• 替换私有方法_log()为标准的debug()/info()方法，符合Python logging最佳实践

方案二：保留client.setup_logging()的兼容方案 #

如果必须保留原有的google.cloud.logging客户端调用，只需阻止自定义logger的日志向上传播：

def get_logger(resource, name):
    client = google.cloud.logging.Client()
    client.setup_logging()

    logger = logging.getLogger(str(name))
    logger.handlers.clear()
    logger.propagate = False  # 核心：禁止日志传到root logger
    logger_provider = LoggerProvider(resource=resource)
    handler = LoggingHandler(level=logging.NOTSET, logger_provider=logger_provider)
    logger.addHandler(handler)
    return logger

但此方案存在多框架混用的潜在风险，不推荐长期使用。

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