【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界面进行可视化展示