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:trueoutput.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.confinput {  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:trueoutput.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]}"

说明：默认的key值是default_list，keys的值是动态分配创建的，当redis接收到的日志中message字段的值包含有error字段，则创建key为error_list，当包含有DEBUG字段，则创建key为debug_list。

问题的解决方法是在每个应用的输出日志中新增一个能够区分这个日志的值，然后再在keys中设置，这样一来就能够把不同应用的日志输出到不同的redis的key中。