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: 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中。

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。