使用exec_filter处理Fluentd日志并输出到stdout遇阻求助

我有一个运行Fluentd的Docker容器，负责收集其他多个容器的日志。想通过脚本处理传入的日志，查阅文档后选用了exec_filter插件，但配置完成后执行docker logs fluentd-container --follow看不到任何输出。

最初的fluent.conf配置：

<source>
  @type forward
  port 24226
  bind 0.0.0.0
  tag logs
</source>

<match *>
  @type exec_filter
  command ruby /rtest.rb
  tag logs
  <format>
    @type json
  </format>
  <parse>
    @type msgpack
  </parse>
  <buffer>
    flush_interval 10s
  </buffer>
</match>

配套的rtest.rb脚本：

require 'json'
require 'msgpack'

begin
  while line = STDIN.gets # 持续从标准输入读取事件
    line.chomp!

    # 输入格式取决于exec_filter的in_format设置
    json = JSON.parse(line)

    # 处理逻辑：添加新字段
    json['new_field'] = "Hey from exec_filter script!"

    # 输出到标准输出，格式取决于exec_filter的out_format设置
    STDOUT.print MessagePack.pack(json)

    # 刷新缓冲区避免事件堆积
    STDOUT.flush
  end
rescue Interrupt # 忽略关闭时的中断异常
end

后来尝试添加stdout输出的配置，但问题依旧：

<source>
  @type forward
  port 24226
  bind 0.0.0.0
</source>

<match *>
  @type stdout
  format json
  tag logs
</match>

<match *>
  @type exec_filter
  command ruby /rtest.rb
  tag logs
  <format>
    @type json
  </format>
  <parse>
    @type msgpack
  </parse>
  <buffer>
    flush_interval 10s
  </buffer>
</match>

• exec_filter定位理解错误：exec_filter是Fluentd的过滤型插件，它的作用是接收日志事件，通过外部脚本处理后将事件重新发回Fluentd内部管道，而非直接输出到容器stdout。最初的配置里，处理后的日志没有被任何输出插件接收，自然看不到日志。
• 多match规则执行顺序问题：Fluentd的<match>按配置文件顺序匹配，第二个配置里第一个<match *>会捕获所有日志并输出到stdout，导致后续的exec_filter根本拿不到日志事件。
• tag传递与匹配逻辑混乱：配置中的tag设置未形成完整流转链路，处理后的日志没有被正确匹配到输出插件。

要实现「处理日志后输出到容器stdout」，需让exec_filter处理后的日志带上专属tag，再用<match>匹配该tag并输出到stdout，同时调整脚本和插件格式配置以方便调试。

调整后的fluent.conf #

<source>
  @type forward
  port 24226
  bind 0.0.0.0
  tag raw.logs  <!-- 给原始日志标记明确的专属tag -->
</source>

<!-- 用exec_filter处理原始日志，处理后输出带新tag的事件 -->
<match raw.logs>
  @type exec_filter
  command ruby /rtest.rb
  tag processed.logs  <!-- 处理后的日志标记新tag -->
  <format>
    @type json  <!-- 输入给脚本的格式为json -->
  </format>
  <parse>
    @type json  <!-- 脚本输出格式改为json，避免二进制格式调试麻烦 -->
  </parse>
  <buffer>
    flush_interval 1s  <!-- 缩短刷新间隔，更快看到处理结果 -->
  </buffer>
</match>

<!-- 匹配处理后的tag，输出到容器stdout -->
<match processed.logs>
  @type stdout
  format json_lines  <!-- 用json_lines格式输出，每行一个json，可读性更强 -->
</match>

调整后的rtest.rb #

将输出格式改为json，避免msgpack二进制格式的调试障碍：

require 'json'

begin
  while line = STDIN.gets
    line.chomp!
    json = JSON.parse(line)
    
    # 自定义处理逻辑：添加新字段和处理时间
    json['new_field'] = "Hey from exec_filter script!"
    json['processed_at'] = Time.now.to_s
    
    # 输出json格式到stdout
    puts json.to_json
    STDOUT.flush
  end
rescue Interrupt
end

1. 确保rtest.rb被正确复制到容器的/目录，并且拥有执行权限：可在Dockerfile中添加RUN chmod +x /rtest.rb命令
2. 重启Fluentd容器
3. 让其他容器通过forward模式将日志发送到Fluentd的24226端口
4. 执行docker logs fluentd-container --follow，即可看到带new_field和processed_at字段的处理后日志

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