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

1.Filebeat.yml配置filebeat.inputs:

- type: log

enabled: true

paths:

- /usr/local/nginx/logs/access.log

exclude_files: [.gz$,收集数据S索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，表示上面新增的到R的字段是顶级参数。

顶级字段在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.多个应用的日志都输出到redisfilebeat.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.不同的收集数据S索应用日志使用不同的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接收到的到R的日志中message字段的值包含有error字段，则创建key为error_list，通过当包含有DEBUG字段，根据则创建key为debug_list。字段

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

很赞哦!（6）