EFK日誌蒐集demo

阿新 • • 發佈：2019-01-12

以前的一些東西整理下。

E：Elasticsearch
F：Flume
K：Kafka
Flume是一個分散式的日誌聚合收集工具，可以從多個且不同型別的日誌源頭收集日誌。Flume的模型如下：
Source代表資料的源頭，channel暫存資料，sink為資料的流向。如下：
這裡寫圖片描述
多個flume代理的情況下，資料可以匯聚到同一個地方，如下：

資料量多的時候，可能終端的資料處理壓力比較大，為了平衡資料生產的速度和處理速度，最好在資料生產和資料處理之間增加一個緩衝，為此使用Kafka。
Apache Kafka是一種分散式釋出-訂閱訊息系統。
Kafka的設計初衷是希望作為一個統一的資訊收集平臺,能夠實時的收集反饋資訊,並需要能夠支撐較大的資料量,且具備良好的容錯能力。
Kafka是一個典型的生產者消費者模型，生產者消費者模型的主要功能就是提供緩衝，平衡資料生產和使用的速率。
Kafka的模型如下：
這裡寫圖片描述

所以Flume負責收集資料，Kafka作為緩衝，消費者從kafka中取資料。就像是多個水管同時向池子中注水，水管從哪取水可以自己決定；同時也有多個水管從池子裡取水，水要流向哪裡也可以自己決定。
這裡寫圖片描述

Elasticsearch是一個基於Lucene的搜尋伺服器，提供了分散式的全文搜尋引擎，傳統的資料庫如果資料在PB（1024TB）級別的，搜尋會非常慢，而Elasticsearch支援PB級別的搜尋。
Elasticsearch的幾個概念：
Cluster：叢集，一個叢集下有多個節點。
Node：相同的叢集名的節點是一個叢集。
Index：Elasticsearch用來儲存資料的邏輯區域。
Document：文件，資料實體。
Document Type：文件的型別。
所以可以為了搜尋的方便，可以把kafka的資料存入Elasticsearch。
模型如下：
這裡寫圖片描述

現在建立一個簡單的日誌收集搜尋的demo：
jdk要求版本：1.7
①先簡單的捕獲異常並記錄，規範異常的記錄格式如下：記錄異常的各種資訊。因為elasticsearch的文件都採用json的格式，所以異常規範了扔到Elasticsearch裡。

{
          "localizedMessage": "分頁查詢Activity物件",
          "cause": "java.lang.NullPointerException",
          "mac": "",
          "type": "java.lang.RuntimeException",
          "extendedStackTrace 
": [
            {
              "author": "風中追風",
              "file": "ActivityManagerAction.java",
              "traceSeat": 1,
              "line": 202,
              "class": "com.cairh.xpe.coupons.backend.action.ActivityManagerAction",
              "method": "getActivityList"
            }
          ],
          "ip": "127.0.0.1",
          "message": "分頁查詢Activity物件",
          "time": "2017-10-19 09:25:07",
          "questParams": {
            "page": [
              "1"
            ],
            "rtQryPage": [
              "1"
            ],
            "activity_type": [
              "2"
            ]
          },
          "serverPort": 80,
          "thread": "http-apr-8080-exec-10",
          "remotePort": 55192,
          "URL": "http://hcsisap.xpe.com/xpe-products-coupons-backend/coupons/getActivityList.json",
          "timemillis": 1508376307761,
          "URI": "/xpe-products-coupons-backend/coupons/getActivityList.json"
}

②匯入sapling-component-logstandard專案，且在要使用異常記錄的專案的pom檔案新增依賴：
這裡寫圖片描述

<dependency>
            <groupId>com.btp</groupId>
            <artifactId>sapling-component-logstandard</artifactId>
            <version>0.0.1-SNAPSHOT</version>
</dependency>

③啟動Flume例項：
這裡寫圖片描述
在bin目錄下執行cmd：flume-ng.cmd agent -conf ../conf -conf-file ../conf/flumeToKafka.conf -name a1
flumeToKafka.conf為啟動一個flume代理的配置檔案，可以配置flume的source、channel和sink，a1為代理名稱。比如

配置source：代表從44444埠獲取資料。
這裡寫圖片描述
配置sink，傳資料到kafka的名為test的topic下。

配置channel：本地的flume資料快取。
啟動代理之後：

檢查一下上面配置的44444埠是否開啟：telnet localhost 44444,如果不是報未開啟，則說明flume代理成功啟動。
④啟動zookeeper，因為kafka依賴zookeeper，所以先啟動zookeeper：
執行zookeeper-3.4.9\bin\zkServer.cmd
這裡寫圖片描述
⑤啟動kafka：
Kafka目錄下啟動：.\bin\windows\kafka-server-start.bat .\config\server.properties

上面的flume的sink的9092埠配置在producer.properties。
啟動一個消費者：
消費者（Kafka目錄下啟動）：.\bin\windows\kafka-console-consumer.bat –zookeeper localhost:2181 –topic test
消費topic為test的資料：
這裡寫圖片描述
然後記錄日誌：

檢視kafka消費端視窗：已經有異常訊息了。

⑥準備將訊息送到elasticsearch中：
在backend的web.xml增加

` <servlet-name>action</servlet-name> <servlet-class>com.btp.logstandard.autoTask.KafkaComsumer</servlet-class> <load-on-startup>2</load-on-startup> </servlet>
啟動elasticsearch：elasticsearch-2.1.0\bin\elasticsearch.bat
elasticsearch-2.1.0\config\elasticsearch.yml可以配置叢集名稱和節點名稱。
開啟網頁：輸入：http://localhost:9200/?pretty`
這裡寫圖片描述
說明啟動成功。

啟動kibana：kibana-4.3.0-windows\bin\kibana.bat
開啟網頁輸入：http://localhost:5601/app/sense
這裡寫圖片描述
表示啟動kinaba成功。
⑦啟動web.xml改變的專案：

如果如下說明啟動成功，現在可以往elasticsearch中放資料了。
再次呼叫

Kafka消費視窗可看到：

Console下（預設存到myexceptions索引下，型別定義為exception）：
這裡寫圖片描述
說明資料存到elasticsearch中成功。
現在是要kinaba進行查詢：GET /myexceptions/exception/_search

說明資料儲存成功。
也可以從elasticsearch中將資料查出來：

注：桌面上建立一個資料夾如下：用來存放flume的資料和kafka的日誌，配置檔案可配：
這裡寫圖片描述