Fargate中Python日志流到CloudWatch被Airflow外层日志嵌套问题

根因说明 #

外层嵌套的日志前缀由Airflow ECSOperator的日志输出逻辑生成：ECSOperator运行时会持续拉取Fargate任务的stdout/stderr输出，每获取一行日志就调用自身的log.info()方法打印，因此所有自定义日志都会作为Airflow日志的消息体存在，被包裹上Airflow日志的固定前缀。

可行解决方案 #

方案1：自定义ECSOperator（无全局配置修改，适配所有版本）

继承原生ECSOperator，重写日志处理方法，跳过Airflow日志框架的包装，直接输出原始日志内容：

from airflow.providers.amazon.aws.operators.ecs import EcsOperator
import sys

class RawLogEcsOperator(EcsOperator):
    def _log_task_output(self, last_event_timestamp):
        # 重写原日志处理逻辑，直接打印原始日志不做包装
        events, last_timestamp = self._get_task_log_events(last_event_timestamp)
        for event in events:
            print(event["message"], file=sys.stdout)
        return last_timestamp

后续DAG中替换原有ECSOperator为上述自定义类即可，该方案不会影响其他任务的日志配置，灵活性最高。

注意：如果使用旧版本AWS Provider包，可对照对应版本ecs.py源码修改日志处理方法名，核心逻辑是跳过self.log.info()调用，直接打印事件的message字段。

方案2：修改Airflow日志全局配置（一次配置全量生效）

如果有Airflow配置修改权限，可在自定义日志配置文件（通常为logging_config.py）中为ECSOperator对应的logger单独配置无前缀格式化器：

LOGGING_CONFIG = {
    # 保留原有日志配置不变，新增/修改如下内容
    "formatters": {
        # 原有其他格式化器配置不变
        "raw_message_formatter": {
            "format": "%(message)s"
        }
    },
    "loggers": {
        # 原有其他logger配置不变
        # Airflow 2.x+ AWS Provider版本路径
        "airflow.providers.amazon.aws.operators.ecs": {
            "handlers": ["console"],
            "level": "INFO",
            "formatter": "raw_message_formatter",
            "propagate": False
        },
        # Airflow 1.x Contrib包版本路径
        # "airflow.contrib.operators.ecs_operator": {
        #     "handlers": ["console"],
        #     "level": "INFO",
        #     "formatter": "raw_message_formatter",
        #     "propagate": False
        # }
    }
}

配置完成后重启Airflow调度器与工作节点即可生效，所有ECSOperator输出的日志都会仅保留原始消息内容，无需修改现有DAG代码。

方案3：关闭内置日志拉取（快速临时方案）

如果不需要在Airflow UI中直接查看日志，可在声明ECSOperator时添加参数get_logs=False，关闭Airflow主动拉取ECS日志的逻辑，不会再生成嵌套日志，直接到任务定义中配置的ECS CloudWatch日志组查看原始日志即可。

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