ELK+Kafka 企业日志收集平台(一)
背景：
最近線上上了ELK，但是只用了一臺Redis在中間作為消息隊列，以減輕前端es集群的壓力，Redis的集群解決方案暫時沒有接觸過，并且Redis作為消息隊列并不是它的強項；所以最近將Redis換成了專業的消息信息發布訂閱系統Kafka, Kafka的更多介紹大家可以看這里：傳送門? ,關于ELK的知識網上有很多的哦，?此篇博客主要是總結一下目前線上這個平臺的實施步驟，ELK是怎么跟Kafka結合起來的。好吧，動手！
ELK架構拓撲：
然而我這里的整個日志收集平臺就是這樣的拓撲：
1，使用一臺Nginx代理訪問kibana的請求;
2，兩臺es組成es集群，并且在兩臺es上面都安裝kibana;（以下對elasticsearch簡稱es）
3，中間三臺服務器就是我的kafka(zookeeper)集群啦; 上面寫的消費者/生產者這是kafka(zookeeper)中的概念;
4，最后面的就是一大堆的生產服務器啦，上面使用的是logstash，當然除了logstash也可以使用其他的工具來收集你的應用程序的日志，例如：Flume，Scribe，Rsyslog，Scripts……
角色：
軟件選用：
elasticsearch-1.7.3.tar.gz #這里需要說明一下，前幾天使用了最新的elasticsearch2.0，java-1.8.0報錯，目前未找到原因，故這里使用1.7.3版本 Logstash-2.0.0.tar.gz kibana-4.1.2-linux-x64.tar.gz 以上軟件都可以從官網下載:https://www.elastic.co/downloadsjava-1.8.0，nginx采用yum安裝部署步驟：
1.ES集群安裝配置;
2.Logstash客戶端配置(直接寫入數據到ES集群，寫入系統messages日志);
3.Kafka(zookeeper)集群配置;(Logstash寫入數據到Kafka消息系統);
4.Kibana部署;
5.Nginx負載均衡Kibana請求;
6.案例：nginx日志收集以及MySQL慢日志收集;
7.Kibana報表基本使用;
ES集群安裝配置;
es1.example.com:
1.安裝java-1.8.0以及依賴包
yum install -y epel-release yum install -y java-1.8.0 git wget lrzsz2.獲取es軟件包
wget https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-1.7.3.tar.gz tar -xf elasticsearch-1.7.3.tar.gz -C /usr/local ln -sv /usr/local/elasticsearch-1.7.3 /usr/local/elasticsearch3.修改配置文件
[root@es1 ~]# vim /usr/local/elasticsearch/config/elasticsearch.yml 32 cluster.name: es-cluster #組播的名稱地址 40 node.name: "es-node1 " #節點名稱，不能和其他節點重復 47 node.master: true #節點能否被選舉為master 51 node.data: true #節點是否存儲數據 107 index.number_of_shards: 5 #索引分片的個數 111 index.number_of_replicas: 1 #分片的副本個數 145 path.conf: /usr/local/elasticsearch/config/ #配置文件的路徑 149 path.data: /data/es/data #數據目錄路徑 159 path.work: /data/es/worker #工作目錄路徑 163 path.logs: /usr/local/elasticsearch/logs/ #日志文件路徑 167 path.plugins: /data/es/plugins #插件路徑 184 bootstrap.mlockall: true #內存不向swap交換 232 http.enabled: true #啟用http4.創建相關目錄
mkdir /data/es/{data,worker,plugins} -p5.獲取es服務管理腳本
?[root@es1 ~]# git clone https://github.com/elastic/elasticsearch-servicewrapper.git [root@es1 ~]# mv elasticsearch-servicewrapper/service /usr/local/elasticsearch/bin/ [root@es1 ~]# /usr/local/elasticsearch/bin/service/elasticsearch install Detected RHEL or Fedora: Installing the Elasticsearch daemon.. [root@es1 ~]# #這時就會在/etc/init.d/目錄下安裝上es的管理腳本啦#修改其配置: [root@es1 ~]# set.default.ES_HOME=/usr/local/elasticsearch #安裝路徑 set.default.ES_HEAP_SIZE=1024 #jvm內存大小，根據實際環境調整即可6.啟動es ，并檢查其服務是否正常
[root@es1 ~]# netstat -nlpt | grep -E "9200|"9300 tcp 0 0 0.0.0.0:9200 0.0.0.0:* LISTEN 1684/java tcp 0 0 0.0.0.0:9300 0.0.0.0:* LISTEN 1684/java訪問http://192.168.2.18:9200/ 如果出現以下提示信息說明安裝配置完成啦，
7.es1節點好啦，我們直接把目錄復制到es2
[root@es1 local]# scp -r elasticsearch-1.7.3 192.168.12.19:/usr/local/[root@es2 local]# ln -sv elasticsearch-1.7.3 elasticsearch [root@es2 local]# elasticsearch/bin/service/elasticsearch install#es2只需要修改node.name即可，其他都與es1相同配置8.安裝es的管理插件
es官方提供一個用于管理es的插件，可清晰直觀看到es集群的狀態，以及對集群的操作管理，安裝方法如下：
[root@es1 local]# /usr/local/elasticsearch/bin/plugin -i mobz/elasticsearch-head安裝好之后，訪問方式為： http://192.168.2.18:9200/_plugin/head，由于集群中現在暫時沒有數據，所以顯示為空,
? ? ? 此時，es集群的部署完成。
Logstash客戶端安裝配置;
在webserve1上面安裝Logstassh
1.downloads ?軟件包 ，這里注意，Logstash是需要依賴java環境的，所以這里還是需要yum install -y java-1.8.0.
[root@webserver1 ~]# wget https://download.elastic.co/logstash/logstash/logstash-2.0.0.tar.gz [root@webserver1 ~]# tar -xf logstash-2.0.0.tar.gz -C /usr/local [root@webserver1 ~]# cd /usr/local/ [root@webserver1 local]# ln -sv logstash-2.0.0 logstash [root@webserver1 local]# mkdir logs etc2.提供logstash管理腳本，其中里面的配置路徑可根據實際情況修改
#!/bin/bash #chkconfig: 2345 55 24 #description: logstash service manager #auto: Maoqiu Guo FILE='/usr/local/logstash/etc/*.conf' #logstash配置文件 LOGBIN='/usr/local/logstash/bin/logstash agent --verbose --config' #指定logstash配置文件的命令 LOCK='/usr/local/logstash/locks' #用鎖文件配合服務啟動與關閉 LOGLOG='--log /usr/local/logstash/logs/stdou.log' #日志START() {if [ -f $LOCK ];thenecho -e "Logstash is already \033[32mrunning\033[0m, do nothing."elseecho -e "Start logstash service.\033[32mdone\033[m"nohup ${LOGBIN} ${FILE} ${LOGLOG} &touch $LOCKfi }STOP() {if [ ! -f $LOCK ];thenecho -e "Logstash is already stop, do nothing."elseecho -e "Stop logstash serivce \033[32mdone\033[m"rm -rf $LOCKps -ef | grep logstash | grep -v "grep" | awk '{print $2}' | xargs kill -s 9 >/dev/nullfi }STATUS() {ps aux | grep logstash | grep -v "grep" >/dev/nullif [ -f $LOCK ] && [ $? -eq 0 ]; thenecho -e "Logstash is: \033[32mrunning\033[0m..."elseecho -e "Logstash is: \033[31mstopped\033[0m..."fi }TEST(){${LOGBIN} ${FILE} --configtest }case "$1" instart)START;;stop)STOP;;status)STATUS;;restart)STOP sleep 2START;;test)TEST;;*)echo "Usage: /etc/init.d/logstash (test|start|stop|status|restart)";; esac3.Logstash 向es集群寫數據
(1)編寫一個logstash配置文件
[root@webserver1 etc]# cat logstash.conf input { #數據的輸入從標準輸入stdin {} }output { #數據的輸出我們指向了es集群elasticsearch {hosts => ["192.168.2.18:9200","192.168.2.19:9200"] ＃es主機的ip及端口} } [root@webserver1 etc]#(2)檢查配置文件是否有語法錯
[root@webserver1 etc]# /usr/local/logstash/bin/logstash -f logstash.conf --configtest --verbose Configuration OK [root@webserver1 etc]#(3)既然配置ok我們手動啟動它，然后寫點東西看能否寫到es
ok.上圖已經看到logstash已經可以正常的工作啦.
４.下面演示一下如何收集系統日志
將之前的配置文件修改如下所示內容，然后啟動logstash服務就可以在web頁面中看到messages的日志寫入es，并且創建了一條索引
[root@webserver1 etc]# cat logstash.conf input { #這里的輸入使用的文件，即日志文件messsagesfile { path => "/var/log/messages" ＃這是日志文件的絕對路徑start_position => "beginning" ＃這個表示從messages的第一行讀取，即文件開始處} }output { ＃輸出到eselasticsearch {hosts => ["192.168.2.18:9200","192.168.2.19:9200"]index => "system-messages-%{+YYYY-MM}" ＃這里將按照這個索引格式來創建索引} } [root@webserver1 etc]#啟動logstash后，我們來看head這個插件的web頁面
ok，系統日志我們已經成功的收集，并且已經寫入到es集群中，那上面的演示是logstash直接將日志寫入到es集群中的，這種場合我覺得如果量不是很大的話直接像上面已將將輸出output定義到es集群即可，如果量大的話需要加上消息隊列來緩解es集群的壓力。前面已經提到了我這邊之前使用的是單臺redis作為消息隊列，但是redis不能作為list類型的集群，也就是redis單點的問題沒法解決，所以這里我選用了kafka ;下面就在三臺server上面安裝kafka集群
Kafka集群安裝配置;
在搭建kafka集群時，需要提前安裝zookeeper集群，當然kafka已經自帶zookeeper程序只需要解壓并且安裝配置就行了
kafka1上面的配置：
1.獲取軟件包.官網：http://kafka.apache.org
[root@kafka1 ~]# wget http://mirror.rise.ph/apache/kafka/0.8.2.1/kafka_2.11-0.8.2.1.tgz [root@kafka1 ~]# tar -xf kafka_2.11-0.8.2.1.tgz -C /usr/local/ [root@kafka1 ~]# cd /usr/local/ [root@kafka1 local]# ln -sv kafka_2.11-0.8.2.1 kafka2.配置zookeeper集群，修改配置文件
[root@kafka1 ~]# vim /usr/local/kafka/config/zookeeper.propertie dataDir=/data/zookeeper clienrtPort=2181 tickTime=2000 initLimit=20 syncLimit=10 server.2=192.168.2.22:2888:3888 server.3=192.168.2.23:2888:3888 server.4=192.168.2.24:2888:3888＃說明： tickTime: 這個時間是作為 Zookeeper 服務器之間或客戶端與服務器之間維持心跳的時間間隔，也就是每個 tickTime 時間就會發送一個心跳。 2888端口：表示的是這個服務器與集群中的 Leader 服務器交換信息的端口； 3888端口：表示的是萬一集群中的 Leader 服務器掛了，需要一個端口來重新進行選舉，選出一個新的 Leader，而這個端口就是用來執行選舉時服務器相互通信的端口。3.創建zookeeper所需要的目錄
[root@kafka1 ~]# mkdir /data/zookeeper4.在/data/zookeeper目錄下創建myid文件，里面的內容為數字，用于標識主機，如果這個文件沒有的話，zookeeper是沒法啟動的哦
[root@kafka1 ~]# echo 2 > /data/zookeeper/myid以上就是zookeeper集群的配置，下面等我配置好kafka之后直接復制到其他兩個節點即可
5.kafka配置
[root@kafka1 ~]# vim /usr/local/kafka/config/server.properties broker.id=2 ＃ 唯一，填數字，本文中分別為2/3/4 prot=9092 ＃ 這個broker監聽的端口 host.name=192.168.2.22 ＃ 唯一，填服務器IP log.dir=/data/kafka-logs # 該目錄可以不用提前創建，在啟動時自己會創建 zookeeper.connect=192.168.2.22:2181,192.168.2.23:2181,192.168.2.24:2181 ＃這個就是zookeeper的ip及端口 num.partitions=16 # 需要配置較大 分片影響讀寫速度 log.dirs=/data/kafka-logs # 數據目錄也要單獨配置磁盤較大的地方 log.retention.hours=168 # 時間按需求保留過期時間 避免磁盤滿6.將kafka(zookeeper)的程序目錄全部拷貝至其他兩個節點
[root@kafka1 ~]# scp -r /usr/local/kafka 192.168.2.23:/usr/local/ [root@kafka1 ~]# scp -r /usr/local/kafka 192.168.2.24:/usr/local/7.修改兩個借點的配置，注意這里除了以下兩點不同外，都是相同的配置
（1）zookeeper的配置 mkdir /data/zookeeper echo "x" > /data/zookeeper/myid （2）kafka的配置 broker.id=2 host.name=192.168.2.228.修改完畢配置之后我們就可以啟動了，這里先要啟動zookeeper集群，才能啟動kafka
我們按照順序來，kafka1 –> kafka2 –>kafka3
[root@kafka1 ~]# /usr/local/kafka/bin/zookeeper-server-start.sh /usr/local/kafka/config/zookeeper.properties & #zookeeper啟動命令 [root@kafka1 ~]# /usr/local/kafka/bin/zookeeper-server-stop.sh #zookeeper停止的命令注意，如果zookeeper有問題 nohup的日志文件會非常大，把磁盤占滿，這個zookeeper服務可以通過自己些服務腳本來管理服務的啟動與關閉。
后面兩臺執行相同操作，在啟動過程當中會出現以下報錯信息
[2015-11-13 19:18:04,225] WARN Cannot open channel to 3 at election address /192.168.2.23:3888 (org.apache.zookeeper.server.quorum.QuorumCnxManager) java.net.ConnectException: Connection refusedat java.net.PlainSocketImpl.socketConnect(Native Method)at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)at java.net.Socket.connect(Socket.java:589)at org.apache.zookeeper.server.quorum.QuorumCnxManager.connectOne(QuorumCnxManager.java:368)at org.apache.zookeeper.server.quorum.QuorumCnxManager.connectAll(QuorumCnxManager.java:402)at org.apache.zookeeper.server.quorum.FastLeaderElection.lookForLeader(FastLeaderElection.java:840)at org.apache.zookeeper.server.quorum.QuorumPeer.run(QuorumPeer.java:762) [2015-11-13 19:18:04,232] WARN Cannot open channel to 4 at election address /192.168.2.24:3888 (org.apache.zookeeper.server.quorum.QuorumCnxManager) java.net.ConnectException: Connection refusedat java.net.PlainSocketImpl.socketConnect(Native Method)at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:350)at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:206)at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:188)at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392)at java.net.Socket.connect(Socket.java:589)at org.apache.zookeeper.server.quorum.QuorumCnxManager.connectOne(QuorumCnxManager.java:368)at org.apache.zookeeper.server.quorum.QuorumCnxManager.connectAll(QuorumCnxManager.java:402)at org.apache.zookeeper.server.quorum.FastLeaderElection.lookForLeader(FastLeaderElection.java:840)at org.apache.zookeeper.server.quorum.QuorumPeer.run(QuorumPeer.java:762) [2015-11-13 19:18:04,233] INFO Notification time out: 6400 (org.apache.zookeeper.server.quorum.FastLeaderElection)由于zookeeper集群在啟動的時候，每個結點都試圖去連接集群中的其它結點，先啟動的肯定連不上后面還沒啟動的，所以上面日志前面部分的異常是可以忽略的。通過后面部分可以看到，集群在選出一個Leader后，最后穩定了。
其他節點也可能會出現類似的情況，屬于正常。
9.zookeeper服務檢查
[root@kafka1~]# netstat -nlpt | grep -E "2181|2888|3888" tcp 0 0 192.168.2.24:3888 0.0.0.0:* LISTEN 1959/java tcp 0 0 0.0.0.0:2181 0.0.0.0:* LISTEN 1959/java [root@kafka2 ~]# netstat -nlpt | grep -E "2181|2888|3888" tcp 0 0 192.168.2.23:3888 0.0.0.0:* LISTEN 1723/java tcp 0 0 0.0.0.0:2181 0.0.0.0:* LISTEN 1723/java [root@kafka3 ~]# netstat -nlpt | grep -E "2181|2888|3888" tcp 0 0 192.168.2.24:3888 0.0.0.0:* LISTEN 950/java tcp 0 0 0.0.0.0:2181 0.0.0.0:* LISTEN 950/java tcp 0 0 192.168.2.24:2888 0.0.0.0:* LISTEN 950/java #可以看出，如果哪臺是Leader,那么它就擁有2888這個端口ok. ?這時候zookeeper集群已經啟動起來了，下面啟動kafka，也是依次按照順序啟動
[root@kafka1 ~]# nohup /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties & #kafka啟動的命令 [root@kafka1 ~]# /usr/local/kafka/bin/kafka-server-stop.sh #kafka停止的命令注意，跟zookeeper服務一樣，如果kafka有問題 nohup的日志文件會非常大,把磁盤占滿，這個kafka服務同樣可以通過自己些服務腳本來管理服務的啟動與關閉。
此時三臺上面的zookeeper及kafka都已經啟動完畢，來檢測以下吧
(1)建立一個主題
[root@kafka1 ~]# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic summer #注意：factor大小不能超過broker數(2)查看有哪些主題已經創建
[root@kafka1 ~]# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.2.22:2181 #列出集群中所有的topic summer #已經創建成功(3)查看summer這個主題的詳情
[root@kafka1 ~]# /usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper 192.168.2.22:2181 --topic summer Topic:summer PartitionCount:1 ReplicationFactor:3 Configs:Topic: summer Partition: 0 Leader: 2 Replicas: 2,4,3 Isr: 2,4,3#主題名稱：summer #Partition:只有一個，從0開始 #leader ：id為2的broker #Replicas 副本存在于broker id為2,3,4的上面 #Isr:活躍狀態的broker(4)發送消息，這里使用的是生產者角色
[root@kafka1 ~]# /bin/bash /usr/local/kafka/bin/kafka-console-producer.sh --broker-list 192.168.2.22:9092 --topic summer This is a messages welcome to kafka(5)接收消息，這里使用的是消費者角色
[root@kafka2 ~]# /usr/local/kafka/bin/kafka-console-consumer.sh --zookeeper 192.168.2.24:2181 --topic summer --from-beginning This is a messages welcome to kafka如果能夠像上面一樣能夠接收到生產者發過來的消息，那說明基于kafka的zookeeper集群就成功啦。
10，下面我們將webserver1上面的logstash的輸出改到kafka上面，將數據寫入到kafka中
(1)修改webserver1上面的logstash配置，如下所示：各個參數可以到官網查詢.
root@webserver1 etc]# cat logstash.conf input { #這里的輸入還是定義的是從日志文件輸入file {type => "system-message" path => "/var/log/messages"start_position => "beginning"} }output {#stdout { codec => rubydebug } #這是標準輸出到終端，可以用于調試看有沒有輸出，注意輸出的方向可以有多個kafka { #輸出到kafkabootstrap_servers => "192.168.2.22:9092,192.168.2.23:9092,192.168.2.24:9092" #他們就是生產者topic_id => "system-messages" #這個將作為主題的名稱，將會自動創建compression_type => "snappy" #壓縮類型} } [root@webserver1 etc]#(2)配置檢測
[root@webserver1 etc]# /usr/local/logstash/bin/logstash -f logstash.conf --configtest --verbose Configuration OK [root@webserver1 etc]#(2)啟動Logstash，這里我直接在命令行執行即可
[root@webserver1 etc]# /usr/local/logstash/bin/logstash -f logstash.conf(3)驗證數據是否寫入到kafka，這里我們檢查是否生成了一個叫system-messages的主題
[root@kafka1 ~]# /usr/local/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.2.22:2181 summer system-messages #可以看到這個主題已經生成了#再看看這個主題的詳情: [root@kafka1 ~]# /usr/local/kafka/bin/kafka-topics.sh --describe --zookeeper 192.168.2.22:2181 --topic system-messages Topic:system-messages PartitionCount:16 ReplicationFactor:1 Configs:Topic: system-messages Partition: 0 Leader: 2 Replicas: 2 Isr: 2Topic: system-messages Partition: 1 Leader: 3 Replicas: 3 Isr: 3Topic: system-messages Partition: 2 Leader: 4 Replicas: 4 Isr: 4Topic: system-messages Partition: 3 Leader: 2 Replicas: 2 Isr: 2Topic: system-messages Partition: 4 Leader: 3 Replicas: 3 Isr: 3Topic: system-messages Partition: 5 Leader: 4 Replicas: 4 Isr: 4Topic: system-messages Partition: 6 Leader: 2 Replicas: 2 Isr: 2Topic: system-messages Partition: 7 Leader: 3 Replicas: 3 Isr: 3Topic: system-messages Partition: 8 Leader: 4 Replicas: 4 Isr: 4Topic: system-messages Partition: 9 Leader: 2 Replicas: 2 Isr: 2Topic: system-messages Partition: 10 Leader: 3 Replicas: 3 Isr: 3Topic: system-messages Partition: 11 Leader: 4 Replicas: 4 Isr: 4Topic: system-messages Partition: 12 Leader: 2 Replicas: 2 Isr: 2Topic: system-messages Partition: 13 Leader: 3 Replicas: 3 Isr: 3Topic: system-messages Partition: 14 Leader: 4 Replicas: 4 Isr: 4Topic: system-messages Partition: 15 Leader: 2 Replicas: 2 Isr: 2 [root@kafka1 ~]#可以看出，這個主題生成了16個分區，每個分區都有對應自己的Leader，但是我想要有10個分區，3個副本如何辦？還是跟我們上面一樣命令行來創建主題就行，當然對于logstash輸出的我們也可以提前先定義主題，然后啟動logstash 直接往定義好的主題寫數據就行啦，命令如下：
[root@kafka1 ~]# /usr/local/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.2.22:2181 --replication-factor 3 --partitions 10 --topic TOPIC_NAME好了，我們將logstash收集到的數據寫入到了kafka中了，在實驗過程中我使用while腳本測試了如果不斷的往kafka寫數據的同時停掉兩個節點，數據寫入沒有任何問題。
那如何將數據從kafka中讀取然后給我們的es集群呢？那下面我們在kafka集群上安裝Logstash，安裝步驟不再贅述；三臺上面的logstash 的配置如下，作用是將kafka集群的數據讀取然后轉交給es集群，這里為了測試我讓他新建一個索引文件，注意這里的輸入日志還是messages，主題名稱還是“system-messages”
[root@kafka1 etc]# more logstash.conf input {kafka {zk_connect => "192.168.2.22:2181,192.168.2.23:2181,192.168.2.24:2181" #消費者們topic_id => "system-messages"codec => plainreset_beginning => falseconsumer_threads => 5decorate_events => true} }output {elasticsearch {hosts => ["192.168.2.18:9200","192.168.2.19:9200"]index => "test-system-messages-%{+YYYY-MM}" #為了區分之前實驗，我這里新生成的所以名字為“test-system-messages-%{+YYYY-MM}”}}在三臺kafka上面啟動Logstash，注意我這里是在命令行啟動的；
[root@kafka1 etc]# pwd /usr/local/logstash/etc [root@kafka1 etc]# /usr/local/logstash/bin/logstash -f logstash.conf [root@kafka2 etc]# pwd /usr/local/logstash/etc [root@kafka2 etc]# /usr/local/logstash/bin/logstash -f logstash.conf [root@kafka3 etc]# pwd /usr/local/logstash/etc [root@kafka3 etc]# /usr/local/logstash/bin/logstash -f logstash.conf在webserver1上寫入測試內容，即webserver1上面利用message這個文件來測試，我先將其清空，然后啟動
[root@webserver1 etc]# >/var/log/messages [root@webserver1 etc]# echo "我將通過kafka集群達到es集群哦^0^" >> /var/log/messages #啟動logstash,讓其讀取messages中的內容下圖為我在客戶端寫入到kafka集群的同時也將其輸入到終端，這里寫入了三條內容
而下面三張圖側可以看出，三臺Logstash 很平均的從kafka集群當中讀取出來了日志內容
再來看看我們的es管理界面
ok ,看到了吧，
流程差不多就是下面 醬紫咯
由于篇幅較長，我將
4.Kibana部署;
5.Nginx負載均衡Kibana請求;
6.案例：nginx日志收集以及MySQL慢日志收集;
7.Kibana報表基本使用;
放到下一篇博客。
來源:http://blog.sctux.com/?p=445
總結
以上是生活随笔為你收集整理的ELK+Kafka 企业日志收集平台(一)的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 【ELK Stack】ELK+KafKa
- 下一篇： ELK5.3+Kafka集群配置