logstash与filebeat收集日志

1、背景介绍

日志收集，采用的是 ELK 框架，即 elasticsearch，logstash，kibana，另外还有 filebeat 组件，其中 filebeat 用于扫描日志文件，将日志发送到 logstash 服务，logstash 服务则完成将日志切分，发送到 elasticsearch 服务。

2、filebeat的部署步骤(日志 -> logstash)

下载安装包，并解压。

创建xxx.filebeat.yml文件，注意内容的格式，否则你会找半天的错误都可能找不出原因。

启动服务

[root@hadoop1 opt] tar -xf  filebeat-5.2.2-linux-x86_64.tar.gz  ./

# 这里根据实际的业务需要编写filebeat.yml文件
[root@hadoop1 filebeat-5.2.2-linux-x86_64] vim  swordfish.filebeat.yml

# 文件的具体内容如下：
filebeat.prospectors:
- input_type: log
multiline.timeout: 1s
paths:
- /opt/swordfish/target/swordfish-all-1.0-SNAPSHOT/logs/exec-server*.log
multiline:
pattern: '^\['
negate: true
match: after
# exclude_lines: ["^\\[DEBUG\\]"]
output.logstash:
hosts: ["192.168.14.148:9999"]

启动指令-后台启动

[root@hadoop1 filebeat-5.2.2-linux-x86_64]$ nohup ./filebeat -e -c udp-filebeat.yml -d publish &

3、logstash的部署(logstash -> elasticsearch)

下载安装包，并解压。

创建swordfish-pipeline.conf文件，注意内容的格式，否则你会找半天的错误都可能找不出原因。

启动服务

# 解压
[root@hadoop1 opt]# tar -xf logstash-5.2.2.tar.gz ./
[root@hadoop1 opt]# vim swordfish-pipeline.conf

# 内容格式要注意
input {
beats {
port => "9999"
}
}

filter {
grok {
match => { "message" => ["%{LOGLEVEL:logLevel}\]\[%{NUMBER:nanoTime:integer}\](?<V1>.*)\[jobId=(?<jobId>[\w\d\-_]+)\](?<V2>.*)", "%{LOGLEVEL:logLevel}\]\[%{NUMBER:nanoTime:integer}\](?<V1>.)(?<V2>.*)"] }
add_field => { "nest_msg" => "[%{logLevel}]%{V1}%{V2}" }
remove_field => [ "message", "V1", "V2" ]
}

if ![jobId] {
drop {}
}

mutate {
convert => { "nanoTime" => "integer" }
}
}

output {
stdout {
codec => json
}
elasticsearch {
hosts => ["192.168.14.148:9200"]
codec => json
index => "swordfish-logstash-%{+YYYY.MM.dd}"
}
}

启动指令-后台启动

[root@hadoop1 logstash-5.2.2]# nohup bin/logstash -f udp-pipeline.conf &

4es的部署我们之前有介绍过，下面我们去es-head页面去看一下日志是否在es中存储。

上面图片中我们发现日志是可以收集到的。