ELK+filebeat日志分析系统部署文档

159次阅读

共计 2503 个字符，预计需要花费 7 分钟才能阅读完成。

环境说明
架构说明及架构图

ELK+filebeat 日志分析系统部署文档

filebeat 部署在客户端用于收集日志并把收集到的日志发送到 logstash.
logstash 把收集到的日志处理之后交给 elasticsearch.
kibana 从 elasticsearch 中提取数据并进行展示.
之所以使用 filebeat 进行日志收集是因为 filebeat 不会像 logstash 使用大量的资源, 影响业务服务器.

环境需求
需要 Java 环境和 redis

yum install java
yum install redis 使用版本
java 1.8.0_111

redis 2.8.16

filebeat 5.1.2

logstash 5.1.2

elasticsearch 5.1.1

kibana 5.1.1

安装配置
filebeat 安装及配置
安装 filebeat

rpm –import http://packages.elasticsearch.org/GPG-KEY-elasticsearch
cat > /etc/yum.repos.d/elk.repo <<EOF
[elasticsearch-5.x]
name=Elasticsearch repository for 5.x packages
baseurl=https://artifacts.elastic.co/packages/5.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF
yum clean all
yum install filebeat -y

配置 filebeat

vim /etc/filebeat/filebeat.yml
paths:
– /var/log/nginx/access.log
tags: [“nginx”]
output.logstash:
hosts: [“1.8.101.53:5044”]

其中 path 选项为 filebeat 发送给 logstash 的路径, 多个日志可以使用 *.log 通配. file 不会自动递归日志目录下的子目录, 如果需要递归子目录可以使用类似 /var/log/*/*.log 的结构. tags 选项会向 log 中添加一个标签, 此标签可以提供给 logstash 用于区分不同客户端不同业务的 log. output 指定发送 log 到哪台服务器的哪个服务, 默认输出到 elasticsearch. 本例使用 logstash, 所以需要注释掉发送到 elasticsearch 的配置, 并启用发送到 logstash 的配置.

启动 filebeat

/usr/share/filebeat/bin/filebeat -c /etc/filebeat/filebeat.yml -e &logstash

安装及配置
安装 logstash

rpm –import http://packages.elasticsearch.org/GPG-KEY-elasticsearch
cat > /etc/yum.repos.d/elk.repo <<EOF
[elasticsearch-5.x]
name=Elasticsearch repository for 5.x packages
baseurl=https://artifacts.elastic.co/packages/5.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF
yum clean all
yum install logstash
ln -s /usr/share/logstash/bin/logstash /usr/bin/logstash 配置 logstash

vim /etc/logstash/conf.d/nginx.conf
input {
beats {
port => 5044
}
}
filter {
if “nginx” in [tags] {
grok {
match => [“message”,”%{COMBINEDAPACHELOG}+%{GREEDYDATA:extra_fields}”]
overwrite => [“message”]
}
mutate {
convert => [“response”,”integer”]
convert => [“bytes”,”integer”]
convert => [“responsetime”,”float”]
}
geoip {
source=>”clientip”
target => “geoip”
}
date {
match => [“timestamp”,”dd/MMM/YYYY:HH:mm:ss Z”]
remove_field => [“timestamp”]
}
useragent {
source=>”agent”
}
}
}
output {
if “nginx” == [tags][0] {
elasticsearch {
hosts => [“1.8.101.53:9200”]
index => “access-%{+YYYY.MM.dd}”
}
}
}