Fluentd单实例接收两类数据如何路由到Elasticsearch不同索引

问题解答 #

首先你之前的配置存在一处不规范的地方：<parse>节点必须是HTTP输入<source>的子节点，放在外层会不生效，先修正这个问题。

关于HTTP路径与tag的对应关系

Fluentd HTTP输入的规则就是：请求URL的路径去掉开头的/后，自动作为该条事件的tag。你之前直接请求根路径http://fluentd:24225，对应的tag是空字符串，原有<match *.**>是匹配所有tag的通配规则，所以可以正常处理。你看到的示例中的app.log就是请求路径，对应的tag就是app.log，和match规则里的匹配值直接对应。

实现方案1：按URL路径路由（最简便，匹配你当前的测试思路）

这种方式不需要修改JSON结构，只要调整请求路径即可：

1. 调整后的完整HTTP输入配置如下：

<source>
  @type http
  port 24225
  bind 0.0.0.0
  <parse>
    @type json
    time_key date
  </parse>
</source>

2. 两类数据分别请求不同的路径：

• 第一类数据发往 http://fluentd:24225/app1
• 第二类数据发往 http://fluentd:24225/app2

3. 替换原来的通配match规则，分tag配置不同的Elasticsearch索引前缀：

# 匹配路径为app1的请求，写入第一个索引
<match app1>
  @type copy
  <store>
    @type elasticsearch
    host elasticsearch
    port 9200
    logstash_format true
    logstash_prefix fluentd_app1 # 自定义第一个索引的前缀
    logstash_dateformat %Y%m%d
    include_tag_key true
    type_name access_log
    tag_key @log_name
    flush_interval 1s
  </store>
  <store>
    @type stdout
  </store>
</match>

# 匹配路径为app2的请求，写入第二个索引
<match app2>
  @type copy
  <store>
    @type elasticsearch
    host elasticsearch
    port 9200
    logstash_format true
    logstash_prefix fluentd_app2 # 自定义第二个索引的前缀
    logstash_dateformat %Y%m%d
    include_tag_key true
    type_name access_log
    tag_key @log_name
    flush_interval 1s
  </store>
  <store>
    @type stdout
  </store>
</match>

实现方案2：按JSON字段路由

如果不想修改请求路径，要根据JSON内部的字段值区分索引，可以用rewrite_tag_filter插件重打tag再路由：

1. 假设你的JSON里有标识分类的字段log_type，值为type1和type2，先在HTTP source和match规则之间加如下filter配置：

<filter *.**>
  @type rewrite_tag_filter
  <rule>
    key log_type
    pattern /^type1$/
    tag log.type1
  </rule>
  <rule>
    key log_type
    pattern /^type2$/
    tag log.type2
  </rule>
</filter>

2. 后面的match规则对应匹配log.type1和log.type2，分别配置不同的logstash_prefix即可，写法和方案1一致。

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