【Logstash】

[root@localhost ~]# wget https://artifacts.elastic.co/downloads/logstash/logstash-6.3.2.tar.gz

[root@localhost ~]# tar zxvf logstash-6.3.2.tar.gz -C /usr/local/

[root@localhost ~]# mv /usr/local/logstash-6.3.2/ /usr/local/logstash

[root@localhost ~]# mkdir /usr/local/logstash/config/etc/

[root@localhost ~]# vim /usr/local/logstash/config/etc/test01.conf 

测试:小试牛刀，我们先将kafka队列消息输出到终端，主要为了测试filebeat生产web日志。推送到kafka队列，logstash端能否消费，很明显，下图所示，消费正常！

input {kafka {bootstrap_servers =>"192.168.37.134:9092,192.168.37.135:9092,192.168.37.136:9092"topics => ["webapache"]codec => "json"}}output {stdout {codec => rubydebug}}

input接收源变成kafka，通过bootstrap_server和topics两个选项指定了接收源kafka集群以及主题属性；因为我们的logstash从kafka获取的的日志数据格式为json，所以需要 在inout字段中加上codec=>"json"解析

输出端是stdout，表示标准输出（从终端输出），这里的codex是个插件，表示格式，放在stout中表示输出的格式，rubydebug是专门测试的格式，一般在用来在终端输出json格式信息

[root@localhost ~]#/usr/local/logstash/bin/logstash -f config/etc/test01.conf

[root@localhost etc]# vim /usr/local/logstash/config/etc/httpd_log.conf      #我们配置将kafka消息存入到ES集群中，

input {kafka {bootstrap_servers =>"192.168.37.134:9092,192.168.37.135:9092,192.168.37.136:9092"topics => ["webapache"]codec => "json"}}output {elasticsearch {hosts => ["192.168.37.134:9200","192.168.37.135:9200","192.168.37.136:9200"]index => "webapachelog-%{+YYYY-MM-dd}"}}

root@localhost etc]#  nohup /usr/local/logstash/bin/logstash -f httpd_log.conf &

详解：上述可以看到，

input接收源变成kafka，通过bootstrap_server和topics两个选项指定了接收源kafka集群以及主题属性；因为我们的logstash从kafka获取的的日志数据格式为json，所以需要 在inout字段中加上codec=>"json"解析

在看输出端，output输出类型是elasticseach，通过hosts指定Elasticsearch集群ip，将kafka消息队列的数据存入elasticsearch中，最后通过index定义索引名称；

 [Kibana]

[root@localhost ~]# tar zxvf kibana-6.3.2-linux-x86_64.tar.gz -C /usr/local/

[root@localhost ~]# mv /usr/local/kibana-6.3.2-linux-x86_64/ /usr/local/kibana

[root@localhost ~]# cd /usr/local/kibana/config/

[root@localhost config]# egrep -v "#|^$" kibana.yml 

server.port: 5601       #kibana默认端口server.host: "192.168.37.136"     #kibana绑定的IP地址，写本地IP即可elasticsearch.url: "http://192.168.37.134:9200"  #kibana访问Elasticsearch地址，可以任意ES集群节点IP，推荐写Client node角色IP

kibana.index: ".kibana"    #存储kibana数据信息的索引

 

这里只需要填写“webapachelog-*”即可，会自动检测ES索引对应的文件并抓取映射，前提是filebeat检测的日志消息经过kafka队列消息存入 ES中；否则会失败

选择“@timestamp”进行时间排序

 

 

Kibana界面展示：#Discover:主要用来进行日志检索，查询数据#Visualize:数据可视化，我们可以创建各种唯独的可视化图表，例如：面积图，折线图，饼图 等#Dashboard:仪盘表功能，仪盘表就是可视化图表的组合，通过将各种可视化图表组合到一起，整体上了解数据和日志的各种状态#Timeblion:时间画像，可以创建时间序列可视化图形#Dev Tools:调试工具控制台，简单的理解就是kibana提供与ES交互的平台#Management:管理界面，可以在此创建索引模式，调整kibana设置等操作

 

总结：整个流程呢，我简单的说一下（大白话说明）

1.首先是由filebeat检索监控日志，将监控的日志路径推送到kafka消息队列中生成topics消息，这里filebeat相当于生产者

2.kafka消息队列将日志消息存储起来，这里结合zookeeper集群，并允许logstash拉取

3.Logstash充当消费者，向kafka集群拉取日志数据，并转发存入Elasticsearch；

4.Kibana为webUI展示页面，与Elasticsearch进行交互，将日志数据可视化的展现出来

根据上图架构，从整体分为两个部分：即日志收集，日志检索；细分如下描述

1.首先是生产日志，filebeat收集业务服务器上的日志数据，我们只需要将其安装在需要收集日志服务器上即可，随后将日志数据实时推送到Kafka集群中，kafka集群会对日志数据进行缓冲和存储，简称缓存，哈哈，这里的filebeat相当于kafka集群中的producer(生产者)

2.filebeat推送到Kafka集群之后，logstash会主动去kafka集群中拉取数据，因为这样consumer消费者可以自主的控制消费速率和方式，从而减少消费过程中出错的几率；其实logstash拉取数据就是分为input,filter和output,中间接受不规则的数据，过滤，分析并转换成格式化数据最后输出，

3.Logstash将格式化的数据中转到Elasticsearch中进行存储索引，所有的数据都会存储到Elasticsearch集群中

4.Kibana将elasticsearch中的数据在web GUI界面进行可视化展示