如何优化Fluentd配置：避免代码重复并输出双日志至stdout

需求与现状 #

• 需求：修改Fluentd自身日志格式后输出至stdout，同时读取alert.log日志并使用相同规则格式化输出
• 尝试操作：使用<label @FLUENT_LOG>捕获自身日志并配置filter和match；尝试将<source>放入该label中复用配置但未生效
• 现状：当前工作配置中<filter>和<match>代码重复，仅Fluentd自身日志输出至stdout，alert.log日志未输出
• Fluentd版本：1.16.2

重现步骤 #

1. 启动Fluentd进程：/usr/sbin/fluentd
2. 参考以下配置测试

工作配置 #

<system>
  <log>
    format json
    time_format %Y-%m-%dT%H:%M:%S%z
  </log>
</system>
<source>
  @type tail
  path /tmp/alertLogs/alert.log
  pos_file /tmp/fluentd.pos
  read_from_head true
  tag alert_logs
  <parse>
    @type json
  </parse>
</source>
<filter alert_logs>
  @type record_modifier
  enable_ruby true
  <record>
    log ${ { message: record["message"]  } }
    type "log"
    level ${record.has_key?("level") ? record["level"]: "unavailable" }
    time ${record.has_key?("time") ? record["time"]: time.strftime('%Y-%m-%dT%H:%M:%S%z') }
  </record>
  remove_keys dummy
</filter>
<match alert_logs>
  @type stdout
</match>
<label @FLUENT_LOG>
  <filter fluent.*>
    @type record_modifier
    enable_ruby true
    <record>
      log ${ { message: record["message"]  } }
      type "log"
      level ${record.has_key?("level") ? record["level"]: "unavailable" }
      time ${record.has_key?("time") ? record["time"]: time.strftime('%Y-%m-%dT%H:%M:%S%z') }
    </record>
    remove_keys dummy
  </filter>
  <match fluent.*>
    @type stdout
  </match>
</label>

不工作配置 #

<system>
  <log>
    format json
    time_format %Y-%m-%dT%H:%M:%S%z
  </log>
</system>
<label @FLUENT_LOG>
  <source>
    @type tail
    path /tmp/alertLogs/alert.log
    pos_file /tmp/fluentd.pos
    read_from_head true
    tag alert_logs
    <parse>
    @type json
  </parse>
</source>
<filter fluent.* alert_logs>
  @type record_modifier
  enable_ruby true
  <record>
    log ${ { message: record["message"]  } }
    type "log"
    level ${record.has_key?("level") ? record["level"]: "unavailable" }
    time ${record.has_key?("time") ? record["time"]: time.strftime('%Y-%m-%dT%H:%M:%S%z') }
  </record>
  remove_keys dummy
</filter>
<match fluent.* alert_logs>
  @type stdout
  </match>
</label>

解决方案 #

问题核心是<label @FLUENT_LOG>的作用域限制：该标签仅处理Fluentd自身生成的fluent.*标签日志，普通source的日志不会自动流入这个标签。直接将source放入该标签内，生成的alert_logs日志不会被标签内的filter和match处理。

以下两种方法可解决代码重复问题：

方法1：使用<include>提取公共配置 #

1. 创建公共配置文件common_log_processing.conf，封装重复的filter和match逻辑：

<filter ${tag_pattern}>
  @type record_modifier
  enable_ruby true
  <record>
    log ${ { message: record["message"]  } }
    type "log"
    level ${record.has_key?("level") ? record["level"]: "unavailable" }
    time ${record.has_key?("time") ? record["time"]: time.strftime('%Y-%m-%dT%H:%M:%S%z') }
  </record>
  remove_keys dummy
</filter>
<match ${tag_pattern}>
  @type stdout
</match>

2. 在主配置中引用公共配置，传递不同的标签参数：

<system>
  <log>
    format json
    time_format %Y-%m-%dT%H:%M:%S%z
  </log>
</system>
<source>
  @type tail
  path /tmp/alertLogs/alert.log
  pos_file /tmp/fluentd.pos
  read_from_head true
  tag alert_logs
  <parse>
    @type json
  </parse>
</source>
<!-- 处理alert_logs -->
<include>
  path ./common_log_processing.conf
  param tag_pattern alert_logs
</include>
<!-- 处理Fluentd自身日志 -->
<label @FLUENT_LOG>
  <include>
    path ./common_log_processing.conf
    param tag_pattern fluent.*
  </include>
</label>

方法2：使用路由将日志转发到公共标签 #

设置公共处理标签，将alert_logs和Fluentd自身日志都转发到该标签处理：

<system>
  <log>
    format json
    time_format %Y-%m-%dT%H:%M:%S%z
  </log>
</system>
<source>
  @type tail
  path /tmp/alertLogs/alert.log
  pos_file /tmp/fluentd.pos
  read_from_head true
  tag alert_logs
  <parse>
    @type json
  </parse>
</source>
<!-- 将alert_logs转发到公共处理标签 -->
<match alert_logs>
  @type relabel
  @label @COMMON_PROCESSING
</match>
<!-- Fluentd自身日志转发到公共标签 -->
<label @FLUENT_LOG>
  <match fluent.*>
    @type relabel
    @label @COMMON_PROCESSING
  </match>
</label>
<!-- 公共处理逻辑，仅需编写一次 -->
<label @COMMON_PROCESSING>
  <filter **>
    @type record_modifier
    enable_ruby true
    <record>
      log ${ { message: record["message"]  } }
      type "log"
      level ${record.has_key?("level") ? record["level"]: "unavailable" }
      time ${record.has_key?("time") ? record["time"]: time.strftime('%Y-%m-%dT%H:%M:%S%z') }
    </record>
    remove_keys dummy
  </filter>
  <match **>
    @type stdout
  </match>
</label>

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