ELK日志分析平台搭建 ： （Logstash、Filebeat部署）

Logstash部署

书接上回，在部署完kafka消息队列后，此次部署logstash，它是用来搜集并且处理，过滤数据的工具。

安装jdk，logstash同样依赖jdk，首先安装jdk（也可以部署在kafka机器上，省资源加省去部署jdk步骤）

[root@kafka1 ~]# tar zxvf /usr/local/package/jdk-8u121-linux-x64.tar.gz -C /usr/local/
[root@kafka1 ~]# echo ‘
JAVA_HOME=/usr/local/jdk1.8.0_121
PATH=$JAVA_HOME/bin:$PATH
export JAVA_HOME PATH
‘ >>/etc/profile
[root@kafka1 ~]# source /etc/profile

安装logstash包

[root@elk ~]# tar zxf /usr/local/package/logstash-7.13.2.tar.gz -C /usr/local/

logstash有三个中重点，input（决定消息从哪来）；filter（决定消息如何处理）；output（决定消息去哪）

input指定了数据来源是nginx的json的日志格式，数据来源是各个机器的9092端口，这个端口是kafka消息队列的输出端口，由filebeat采集json文件传到kafka中，logstash去9092端口拿到数据并处理。

[root@elk ~]# mkdir -p /usr/local/logstash-7.13.2/etc/conf.d
[root@elk ~]# vim /usr/local/logstash-7.13.2/etc/conf.d/input.conf
input {
kafka {
type => “audit_log”
codec => “json”
topics => “nginx”
decorate_events => true
bootstrap_servers => “10.3.145.41:9092, 10.3.145.42:9092, 10.3.145.43:9092”
}
}

filter文件配置，由于采集到的是json格式文件，采用json插件去处理

[root@elk ~]# vim /usr/local/logstash-7.13.2/etc/conf.d/filter.conf
filter {
json {
source => “message”
target => “nginx”
}
}

output文件配置，host制定了es集群的地址，用于把数据塞过去
index => ‘logstash-audit_log-%{+YYYY-MM-dd}’

• 这个配置项指定了 Elasticsearch 中用于索引的名称模式。在这里，它使用了 Logstash 内置的日期格式，创建一个每天一个的索引，索引名的格式是 “logstash-audit_log-年月日”。

[root@elk ~]# vim /usr/local/logstash-7.13.2/etc/conf.d/output.conf
output {
if [type] == “audit_log” {
elasticsearch {
hosts => [“10.3.145.14″,”10.3.145.56″,”10.3.145.57”]
index => ‘logstash-audit_log-%{+YYYY-MM-dd}’
}
}
}

以我创建的配置文件目录启动logstash

[root@elk ~]# cd /usr/local/logstash-7.13.2
[root@elk ~]# nohup bin/logstash -f etc/conf.d/ –config.reload.automatic &

Fiebeat部署

fiebeat是代替logstash进行数据采集的，由于logstash运行在jvm上，资源消耗巨大，作者用go语言写了一个功能比较少但可以代替logstash在各节点上运行采集工作的替代品。filebeat部署在每个需要被采集日志的机器上。

安装filebeat

[root@kafka3 ~]# curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.13.2-x86_64.rpm
[root@kafka3 ~]# yum install -y filebeat-7.13.2-x86_64.rpm

修改配置，数据来源是本地nginx的访问日志access.log，数据去向是kafka消息队列。

[root@kafka3 ~]# vim filebeat.yml
filebeat.inputs:
type: log enabled: true paths: /var/log/nginx/access.log
output.kafka:
hosts: [“10.3.145.41:9092″,”10.3.145.42:9092″,”10.3.145.43:9092”]
topic: ‘nginx’

启动filebeat进行收集日志

./filebeat -e -c filebeat.yml

由于配置中写的文件格式为json，所以要把nginx生成的日志转为json的，进入nginx配置文件，将原本日志注释，添加新的日志

[root@kafka3 ~]# vim /etc/nginx/nginx.conf
# log_format main ‘$remote_addr – $remote_user [$time_local] “$request” ‘
# ‘$status $body_bytes_sent “$http_referer” ‘
# ‘”$http_user_agent” “$http_x_forwarded_for”‘;
log_format main '{"user_ip":"$http_x_real_ip","lan_ip":"$remote_addr","log_time":"$time_iso8601","user_req":"$request","http_code":"$status","body_bytes_sents":"$body_bytes_sent","req_time":"$request_time","user_ua":"$http_user_agent"}'; access_log /var/log/nginx/access.log main;

一切准备就绪，登录kibana，nginx日志正常显示，ELK集群搭建完成