Filebeat收集日志数据传输到Redis，通过Logstash来根据日志字段创建不同的ES索引

1.Filebeat.yml配置

filebeat.inputs:
- type: log
  enabled:true
  paths:
-/usr/local/nginx/logs/access.log
  exclude_files:['.gz$','INFO']
  multiline.pattern:'^[0-9]{4}-[0-9]{2}-[0-9]{2}'
  multiline.negate:true
  multiline.match: after
  tags:["nginx-log-messages"]
  fields:
    log_source: messages
  fields_under_root:true


output.redis:
  hosts:["192.168.0.111:6379"]
  key: nginx_log
  password: nginxredis
  db:0

参数说明：

  fields:
    log_source: messages
  fields_under_root:true

使用fields表示在filebeat收集的日志中多增加一个字段log_source,其值是messages，用来在logstash的output输出到elasticsearch中判断日志的来源，从而建立相应的索引若fields_under_root设置为true，表示上面新增的字段是顶级参数。

顶级字段在output输出到elasticsearch中的使用如下：

[root@es-master21 logstash]# vim config/logstash.conf
input {
  redis {
    data_type =>"list"
    host =>"192.168.0.111"
    db =>"0"
    port =>"6379"
    key =>"nginx_log"
    password =>"nginxredis"
}
}

output {
  # 根据redis键 messages_secure 对应的列表值中，每一行数据的其中一个参数来判断日志来源
  if [log_source]=='messages'{  # 注意判断条件的写法
    elasticsearch {
      hosts =>["192.168.0.111:9200"]
      index =>"nginx-message-%{+YYYY.MM.dd}"
      #user =>"elastic"
      #password =>"elastic123"
}
}

  #或者也可以根据tags判断
  if "nginx-log-messages"in[tags]{
    elasticsearch {
        hosts =>[""192.168.0.111:9200"]
        index => "nginx-message-%{+YYYY.MM.dd}"
    }
  }

}

2.多个应用的日志都输出到redis

filebeat.inputs:
- type: log
  enabled:true
  paths:
-/usr/local/nginx/logs/access.log
  tags:["nginx-log-access"]
  fields:
    log_source: access
  fields_under_root:true

- type: log
  enabled:true
  paths:
-/usr/local/nginx/logs/error.log
  tags:["nginx-log-error"]
  fields:
    log_source: error
  fields_under_root:true

output.redis:
  hosts:["192.168.0.111:6379"]
  key: nginx_log
  password: nginxredis
  db:0

在redis中显示的效果是都会输出到key值nginx_log对应的列表中，根据key值是没法进行区分的，只能根据key值列表中每一行数据中的log_source或者自己定义的属性来判断该行是哪一个应用的日志。

3.不同的应用日志使用不同的rediskey值

使用output.redis中的keys值，官方例子：

output.redis:
  hosts:["localhost"]
  key:"default_list"
  keys:
- key:"error_list"   # send to info_list if `message` field contains INFO
      when.contains:
        message:"error"
- key:"debug_list"  # send to debug_list if `message` field contains DEBUG
      when.contains:
        message:"DEBUG"
- key:"%{[fields.list]}"