Airflow ECS Operator结合FireLens无法获取CloudWatch日志的解决办法咨询

1. 修改Fluent Bit配置，生成符合Airflow预期的日志组格式 #

直接调整Fluent Bit的CloudWatch输出配置，将日志组名称的分隔符从横杠（-）改为斜杠（/）。CloudWatch日志组名称支持斜杠作为路径分隔符，完全匹配Airflow的读取逻辑。

示例Fluent Bit输出配置片段：

[OUTPUT]
    Name cloudwatch
    Match *
    log_group_name ${awslogs_stream_prefix}/${ecs_task_id}  # 将-替换为/
    log_stream_name ${container_name}
    region <你的AWS区域>
    auto_create_group true

如果使用ECS FireLens任务定义配置，对应调整log_driver的log_group参数模板：

"logConfiguration": {
  "logDriver": "awsfirelens",
  "options": {
    "Name": "cloudwatch",
    "log_group": "${awslogs_stream_prefix}/${ecs_task_id}",
    "log_stream": "${container_name}",
    "region": "<你的AWS区域>"
  }
}

优点：从根源解决格式不匹配问题，无需修改Airflow代码，后续任务日志可被Airflow正常读取。

2. 自定义Airflow EcsOperator，适配现有日志组格式 #

若无法调整Fluent Bit配置，可修改Airflow侧的日志读取逻辑，适配横杠分隔的日志组格式。

方案A：自定义Operator继承EcsOperator（推荐） #

创建自定义Operator，重写获取日志组的方法，避免修改Airflow源码：

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

class CustomEcsOperator(EcsOperator):
    def _get_cloudwatch_log_group(self, task_id):
        # 适配横杠分隔的日志组格式
        return f"{self.awslogs_stream_prefix}-{task_id}"

之后在DAG中使用CustomEcsOperator替代原生EcsOperator即可。

方案B：修改Airflow源码（不推荐，升级会覆盖） #

找到Airflow EcsOperator构造CloudWatch日志组的代码（通常在airflow/providers/amazon/aws/operators/ecs.py），将斜杠替换为横杠：

# 原代码
log_group_name = f"{self.awslogs_stream_prefix}/{task_id}"
# 修改后
log_group_name = f"{self.awslogs_stream_prefix}-{task_id}"

优点：无需改动日志投递流程，适配现有日志格式；自定义Operator方式不会被Airflow版本升级覆盖。

3. 用Lambda+CloudWatch订阅过滤器做日志转发（折中方案） #

若以上两种方案都无法实施，可通过CloudWatch订阅过滤器触发Lambda函数，将横杠格式日志组的日志实时转发到斜杠格式的日志组中，同时保留原日志组内容。

核心步骤： #

1. 创建目标日志组（格式为{awslogs_stream_prefix}/{ecs_task_id}），或让Lambda自动创建。
2. 给原日志组（横杠格式）添加订阅过滤器，触发Lambda函数。
3. Lambda解析CloudWatch日志事件，调用PutLogEvents接口将日志写入目标日志组。

示例Lambda核心代码（Python）：

import boto3
import gzip
import base64
import json

client = boto3.client('logs')

def lambda_handler(event, context):
    # 解析压缩的日志事件
    payload = gzip.decompress(base64.b64decode(event['awslogs']['data']))
    log_events = json.loads(payload)
    
    # 构造目标日志组名称（替换第一个横杠为斜杠）
    source_log_group = log_events['logGroup']
    target_log_group = source_log_group.replace('-', '/', 1)
    
    # 转发日志到目标组
    client.put_log_events(
        logGroupName=target_log_group,
        logStreamName=log_events['logStream'],
        logEvents=log_events['logEvents'],
        sequenceToken=log_events.get('nextSequenceToken')
    )

优点：无需修改现有业务流程，快速兼容Airflow的日志读取逻辑；缺点：增加Lambda运行成本，存在一定日志延迟。

4. 切换为ECS原生awslogs日志驱动（备选） #

如果业务允许调整日志驱动方式，可放弃自定义Fluent Bit，改用ECS原生awslogs驱动，默认生成的日志组格式通常为{awslogs_stream_prefix}/{ecs_task_id}，直接匹配Airflow预期。

任务定义中的日志配置示例：

"logConfiguration": {
  "logDriver": "awslogs",
  "options": {
    "awslogs-group": "${awslogs_stream_prefix}",
    "awslogs-stream-prefix": "ecs",
    "awslogs-region": "<你的AWS区域>"
  }
}

注意：此方案会丢失Fluent Bit同时投递到OpenSearch的功能，需额外配置CloudWatch订阅过滤器将日志转发到OpenSearch。

内容的提问来源于stack exchange，提问作者erase.ego