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

幽沉谢世事，俯默窥唐虞。这篇文章主要讲述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

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

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]"

? 说明：默认的key值是default_list，keys的值是动态分配创建的，当redis接收到的日志中message字段的值包含有error字段，则创建key为error_list，当包含有DEBUG字段，则创建key为debug_list。
? 问题的解决方法是在每个应用的输出日志中新增一个能够区分这个日志的值，然后再在keys中设置，这样一来就能够把不同应用的日志输出到不同的redis的key中。
【个人公众号】? ??【微信公众号】搜索：非著名运维 --》公众号回复 “资料” 送运维自学资料大礼包哦！??上面也会分享一些运维知识，大家点点发财手关注一波，感谢大家。