filebeat+logstash+kafka部署

 由于工作需要对业务产生的日志做分析，因此选择了比较成熟稳定的elk体系，这里记录下安装过程。因为当前生产环境已经部署kafka-0.10.0.0版本，因此相应的logstash、filebeat等都使用了最后兼容此版本kafka的相关版本。

一、环境及相关应用包版本

filebeat及logstash

下载地址

1. 
   linux系统:
   
   Linux version 2.6.32-754.17.1.el6.x86_64 (mockbuild@x86-01.bsys.centos.org) (gcc version 4.4.7 20120313 (Red Hat 4.4.7-23) (GCC) ) #1 SMP Tue Jul 2 12:42:48 UTC 2019
2. 
   filebeat版本：
   
   filebeat-6.0.1-linux-x86_64.tar.gz（选择此版本是因为此版本与0.10版本kafka兼容）
3. 
   kafka版本：
   
   v0.10.0.0
4. 
   logstash版本：
   
   v5.6.16(此版本是兼容kafka0.10的最后一个版本。注：与logstash-kafka插件有关)

二、安装

1. 
   filebeeat 安装
   
   
   
   1)
   
   解压filebeat压缩包 tar -zxvf filebeat-6.0.1-linux-x86_64.tar.gz
   
   
   2)
   
   进入filebeat 打开 filebeat.yml配置文件如下图将enabled改为true使得此配置文件有效，并且配置需要输入的日志路径 paths(注意配置文件格式，冒号后需要空格等)。
   
   
   
   
   3)
   
   注释掉默认的elasticsearch 输出配置，如下图：
   
   
   
   4)
   
   配置logstash 输出，hosts 为logstash地址可配置多个以逗号分开，loadbalance属性为true时用以开启负载，如下图：

5)

非后台服务启动filebeat： ./filebeat -e -c filebeat.yml 。-c参数用于指定配置文件。以后台服务启动：nohup ./filebeat -e -c filebeat.yml > filebeat.log &

2.

logstash安装

*


1)

解压logstash: tar zxvf logstash-5.6.5.tar.gz


2)

运行测试命令 ./logstash -e ‘input{stdin{}} output{ stdout{}}’,输入任意内容查看是否正确输出。


3)

新建文件logstash.conf 作为配置文件，配置文件示例如下图，(这里grok解析的是nginx标准日志格式：“

grok{match=>{"message" => "%{IPORHOST:remoteAddr} (%{USER:ident}|-) (%{USER:auth}|-) \[%{HTTPDATE:time}\] \"%{WORD:httpMethod} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpVersion}\" %{NUMBER:status} %{NUMBER:bytes} \"%{URI:httpReferer}\" \" (%{GREEDYDATA:useragent}|-)\" "}}

”，具体解析规则根据自己需求。)在input中配置输入的内容，这里 使用filebeat输入配置beats{port =>5044}。填写filter 消息过滤规则，配置输出源。这里是kafka，配置topic_id,code为json即消息输出为json格式,bootstrap_servers为kafka地址。

4）

非后台服务启动 bin/logstash -f config/logstash.conf 。后台服务启动：nohup bin/logstash -f config/logstash.conf &


5)

同一台服务器起多个logstah的方法，启动时增加–path.data参数，指定一个新的数据目录即可，注意logstah.conf监听的输入端口不要冲突。命令如下：

bin/logstash -f config/logstash2.conf –path.data=/root/data2