如何在FluentD中仅发送符合指定条件的记录

针对你的需求，下面提供两种主流的FluentD条件过滤实现方式，分别对应统一输出和多目标路由场景：

一、统一输出：满足指定条件的记录发送至ElasticSearch #

使用grep过滤器先筛选出符合条件的记录，再转发到ElasticSearch。这种方式适合将所有满足（as≠200 或 x为apache/nginx）的记录统一输出到同一目标。

配置示例 #

# 原有的source配置保持不变
<source>
  bind 0.0.0.0
  <parse>
    expression /<SOME CUSTOM REGEXP>/
    @type regexp
  </parse>
  port 5514
  tag main_tag
  <transport tcp>
  </transport>
  @type syslog
</source>

# 添加过滤器：筛选符合条件的记录
<filter main_tag.local3.*>
  @type grep
  # 条件1：x字段值为apache或nginx
  <regexp>
    key x
    pattern /^(apache|nginx)$/
  </regexp>
  # 逻辑OR：满足任意一个条件即可
  <or>
    # 条件2：as字段值不为200
    <not>
      <regexp>
        key as
        pattern /^200$/
      </regexp>
    </not>
  </or>
</filter>

# 转发筛选后的记录到ElasticSearch
<match main_tag.local3.*>
  @type elasticsearch
  host your-es-hostname-or-ip
  port 9200
  index fluentd-filtered-%Y%m%d  # 按日期滚动索引
  # 可选：添加认证配置（如果ES启用了安全机制）
  # user elastic
  # password your-password
</match>

# 可选：保留原stdout输出用于调试
<match main_tag.local3.*>
  @type stdout
  @id debug_output
</match>

配置说明 #

• grep插件的<or>块表示逻辑"或"，满足其中任意一个条件的记录会被保留
• <not>块用于取反，这里表示排除as=200的记录
• 正则表达式/^(apache|nginx)$/精确匹配x字段的有效值，避免部分匹配（比如apache2不会被误匹配）

二、多目标路由：不同条件发送至不同目标 #

使用@if条件表达式在match块中直接指定路由规则，适合将不同条件的记录分发到ElasticSearch不同索引、文件或其他输出目标。

配置示例 #

# 原有的source配置保持不变
<source>
  bind 0.0.0.0
  <parse>
    expression /<SOME CUSTOM REGEXP>/
    @type regexp
  </parse>
  port 5514
  tag main_tag
  <transport tcp>
  </transport>
  @type syslog
</source>

# 路由1：as≠200的记录发送到错误日志索引
<match main_tag.local3.*>
  @type elasticsearch
  host your-es-hostname-or-ip
  port 9200
  index error-logs-%Y%m%d
  @if record["as"] != "200"
</match>

# 路由2：x=apache的记录发送到apache专属索引
<match main_tag.local3.*>
  @type elasticsearch
  host your-es-hostname-or-ip
  port 9200
  index apache-logs-%Y%m%d
  @if record["x"] == "apache"
</match>

# 路由3：x=nginx的记录保存到本地文件
<match main_tag.local3.*>
  @type file
  path /var/log/fluentd/nginx-access-logs
  append true
  time_slice_format %Y%m%d
  @if record["x"] == "nginx"
</match>

# 路由4：处理剩余未匹配的记录（可选）
<match main_tag.local3.*>
  @type stdout
  @id unmatched_records
</match>

配置说明 #

• @if支持Ruby表达式，record["字段名"]用于访问日志记录中的字段
• 注意字段类型：你的日志中as字段是字符串格式（如"200"），因此直接用字符串比较；如果是数字类型，需要转换为整数后比较（如record["as"].to_i != 200）
• 多个match块按顺序执行，匹配到第一个符合条件的块后，记录会被该块处理（默认不会继续匹配后续块，如需多匹配可添加@copy标签）

注意事项 #

1. 确保已安装对应输出插件：比如ElasticSearch插件可通过fluent-gem install fluent-plugin-elasticsearch安装（若使用fluent-package，部分插件已预安装）
2. 测试配置：使用fluentd -c your-config-file.conf -v启动FluentD，查看日志确认过滤规则是否生效
3. 性能优化：对于高流量场景，建议优先使用grep过滤器（性能优于@if表达式）

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