ELK日志服务系统部署搭建

方案：ELK（elasticsearch + logstash + kibana + filebeat）使用kafka做消息中间件

1. ELK系统需要依赖java环境，新版本的ELK依赖java8，所以先要在服务器上安装jdk1.8，下载压缩包解压即可，配置环境变量：编辑/etc/profile文件，加入

保存退出之后执行source /etc/profile，使配置的环境变量生效，执行java -version命令查看java版本，出现java版本说明jdk安装配置成功

2.安装elasticsearch，下载压缩包解压，再修改elasticsearch.yml文件，设置访问elasticsearch的主机IP和端口，这里elasticsearch不能以root用户运行，需要创建新的用户去运行，这里创建新的用户elsearch，密码也是elsearch，切换elsearch用户启动elasticsearch，报错

切换到root用户，修改/etc/security/limit.conf文件加上如下内容

启动还是报错

修改/etc/sysctl.conf文件，加上这句vm.max_map_count=655360保存在重新启动，这里有可能还会报没有对文件的访问权限的错误，我们需要对创建的elsearch用户授权，授权此用户对elasticsearch文件的读写及执行权限。

3.安装kibana，下载安装包并解压，修改/config目录下kibana.yml文件，配置

访问ip + 端口号可以看见kibana的图形化界面

4.kibana汉化，github上下载汉化包Kibana_Hanization-master.zip上传至服务器，解压进入该目录下执行命令python main.py + Kibana目录

待全部翻译完毕则kibana汉化完成

5. 安装filebeat，下载压缩包并解压，修改filebeat.yml配置文件，path代表需要监听的日志文件，output里可以

配置输出的位置，filebeat可以同时监听多个文件，如：/var/log/* /*.log，目前filebeat没有对目录做递归处理，如上配置只会寻找log文件夹的子目录下的所有的以.log结尾的文件，而匹配不到在log目录下的以.log结尾的文件。

include_line表明，输出的日志中要包含的内容，可以使用通配符表示，exclude_line指的是排除掉的日志，这里可以对传输的日志进行筛选。可以输出至kibana,elasticsearch,kafka,自带的文件里没有kafka的输出配置，需要把输出到kibana和elasticsearch的代码注掉，然后加上输出到kafka的信息

topic代表把数据发送到kafka的哪个主题上，如果目前没有这个主题，则会自动创建一个主题，配置完成后运行

./filebeat -e -c filebeat.yml -d “Publish”启动filebeat，则可以把数据发送到kafka的topic

6. 使用kafka做消息中间件，做数据缓冲，安装kafka，kafka需要zookeeper做支持，先编辑/config/zookeeper.properties文件，

配置之后在bin目录下执行zookeeper-server-start.sh config/zookeeper.properties 

启动zookeeper，如果没有报错则说明配置没有问题，可以zookeeper-server-start.sh config/zookeeper.properties & 后台启动zookeeper，停止zookeeper运行命令./zookeeper-server-stop.sh然后配置kafka，编辑/config/server.properties文件

同样编辑完成之后启动afka-server-start.sh config/server.properties

没有问题就可以加上&后台启动了，停止kafka运行./kafka-server-stop.sh

7. kafka做消息中间件需要logstash来转发数据，安装logstash，logstash启动时要指定配置文件去启动，所以需要新建配置文件，在/config目录下新建test_conf文件，编辑配置文件指定启动需要的参数和配置。在配置文件中编辑

input代表数据的来源，logstash数据来源可以直接从文件中读取，可以从redis中读取，可以从kafka中读取，我们这里从kafka中获取数据，所以配置input为kafka，bootstrap_servers代表kafka所在机器的ip地址加端口，topics代表我们接收kafka哪些topic的数据，不止一个可以用逗号分隔，output指的是输出的地方，这里我们输出到elasticsearch，hosts指定elasticsearch的地址和端口，index指定发送到elasticsearch的索引，stdout {codec=>rubydebug}表示发送数据的同时会在logstash里输出数据，方便我们查看是否拉到数据,有多个服务器发送来过来数据会发送到kafka不同的topics，在发送给elasticsearch时需要发送到不同的索引，需要在配置文件里做下判断，可以在input和output里写多个输出和输入，在输时加上一个字段type,输出通过判断type来确定发到elasticsearch的索引

8. 我们先后台启动elasticsearch和kibana，访问ip + 端口号可以访问，

说明elasticsearch启动成功，访问ip + 端口号 /_cat/indices?v

可以查看当前elasticsearch里已有的索引信息，然后分别配置好kafka和zookeeper,并后台启动它们，在结合filebeat的配置文件后台启动filebeat，执行/filebeat -e -c filebeat.yml >/dev/null 2>&1 &命令，这样filebeat就会把采集到的日志根据配置发送到kafka的topic,可以执行kafka-topics.sh --list --zookeeper127.0.0.1:2181命令查看当前kafka的topic信息，若有发送过来的topic说明filebeat向kafka发送数据成功，最后启动logstash,将数据从kafka中拉取出来，根据配置的索引发送到elasticsearch中，可以查看elasticsearch的索引查看数据是否成功发送到elasticsearch，最后在kibana中创建并匹配当前已有的索引，就可以在页面查看发来的日志数据

9. 在别的服务器上配置filebeat出现一个问题，配置文件ip端口什么的配置以后，启动filebeat去发送数据的时候会报错，因为kafka,logstash，elasticsearch，kibana都在一台机器上部署，配置的都是内网ip，filebeat在别的服务器上配置时配置的内网ip是访问不到的，连接不成功数据发不过去，解决办法是，kafka的server.properties配置文件中有一个属性advertised.host.name是用来返回host.name的值，把这个属性放开，把它的值配置为外网的ip在此重新启动zookeeper和kafka，在去启动filebeat去发送数据，就不会报错，数据就可以成功的发送到kibana了。

10在部署过程中因为某些服务器上可能会有多个项目，会有多个日志文件而不在同一目录下，需要使用filebeat监听多个日志并且发送到kafka的不同topic，logstash再根据不同的topic将数据发送到elasticsearch不同的索引，要在filebeat的配置文件filebeat.yml中配置多个

 

每个输入的type可以监听一个目录下的文件，每个type定义一个fields的log_topics，output的topic可以写成 topic: '%{[fields][log_topics]}'，这样不同目录下的文件就会发送到kafka不同的topic上，在logstash将数据源发送到elasticsearch时topic进行选择，将不同的topic的数据发送到elasticsearch不同的索引上即可。