基於Kafka+ELK搭建海量日誌平臺

早在傳統的單體應用時代，查看日誌大都經過SSH客戶端登服務器去看，使用較多的命令就是 less 或者 tail。若是服務部署了好幾臺，就要分別登陸到這幾臺機器上看，等到了分佈式和微服務架構流行時代，一個從APP或H5發起的請求除了須要登錄服務器去排查日誌，每每還會通過MQ和RPC調用遠程到了別的主機繼續處理，開發人員定位問題可能還須要根據TraceID或者業務惟一主鍵去跟蹤服務的鏈路日誌，基於傳統SSH方式登錄主機查看日誌的方式就像圖中排查線路的工人同樣困難，線上服務器幾十上百之多，出了問題難以快速響應，所以須要高效、實時的日誌存儲和檢索平臺，ELK就提供這樣一套解決方案。

1、ELK架構的分類

ELK是一套應用組件，由Elasticsearch、Logstash和Kibana三部分組件組成，簡稱ELK；它是一套開源免費、功能強大的日誌分析管理系統。ELK能夠將咱們的系統日誌、網站日誌、應用系統日誌等各類日誌進行收集、過濾、清洗，而後進行集中存放並可用於實時檢索、分析。

---

• Elasticsearch ：分佈式搜索引擎。具備高可伸縮、高可靠、易管理等特色。能夠用於全文檢索、結構化檢索和分析，並能將這三者結合起來。Elasticsearch 基於 Lucene 開發，如今使用最廣的開源搜索引擎之一，Wikipedia 、StackOverflow、Github 等都基於它來構建本身的搜索引擎。
• Logstash ：數據收集處理引擎。支持動態的從各類數據源蒐集數據，並對數據進行過濾、分析、豐富、統一格式等操做，而後存儲以供後續使用。
• Kibana ：可視化化平臺。它可以搜索、展現存儲在 Elasticsearch 中索引數據。使用它能夠很方便的用圖表、表格、地圖展現和分析數據。

1.最簡單的ELK架構

此架構主要是將Logstash部署在各個節點上搜集相關日誌、數據，並通過分析、過濾後發送給遠端服務器上的Elasticsearch進行存儲。Elasticsearch再將數據以分片的形式壓縮存儲，並提供多種API供用戶查詢、操做。用戶能夠經過Kibana Web直觀的對日誌進行查詢，並根據需求生成數據報表。

此架構的優勢是搭建簡單，易於上手。缺點是Logstash消耗系統資源比較大，運行時佔用CPU和內存資源較高。另外，因爲沒有消息隊列緩存，可能存在數據丟失的風險，適合於數據量小的環境使用。

2.引入Kafka的典型ELK架構

爲保證日誌傳輸數據的可靠性和穩定性，引入Kafka做爲消息緩衝隊列，位於各個節點上的Logstash Agent（一級Logstash，主要用來傳輸數據）先將數據傳遞給消息隊列，接着，Logstash server（二級Logstash，主要用來拉取消息隊列數據，過濾並分析數據）將格式化的數據傳遞給Elasticsearch進行存儲。最後，由Kibana將日誌和數據呈現給用戶。因爲引入了Kafka緩衝機制，即便遠端Logstash server因故障中止運行，數據也不會丟失，可靠性獲得了大大的提高。

該架構優勢在於引入了消息隊列機制，提高日誌數據的可靠性，但依然存在Logstash佔用系統資源過多的問題，在海量數據應用場景下，可能會出現性能瓶頸。

3.FileBeats+Kafka+ELK集羣架構

該架構從上面架構基礎上改進而來的，主要是將前端收集數據的Logstash Agent換成了filebeat，消息隊列使用了kafka集羣，而後將Logstash和Elasticsearch都經過集羣模式進行構建，完整架構如圖所示：

日誌採集器Logstash其功能雖然強大，可是它依賴java、在數據量大的時候，Logstash進程會消耗過多的系統資源，這將嚴重影響業務系統的性能，而filebeat就是一個完美的替代者，它基於Go語言沒有任何依賴，配置文件簡單，格式明瞭，同時filebeat比logstash更加輕量級，因此佔用系統資源極少，很是適合安裝在生產機器上。這就是推薦使用filebeat，也是 ELK Stack 在 Agent 的第一選擇。

此架構適合大型集羣、海量數據的業務場景，它經過將前端Logstash Agent替換成filebeat，有效下降了收集日誌對業務系統資源的消耗。同時，消息隊列使用kafka集羣架構，有效保障了收集數據的安全性和穩定性，然後端Logstash和Elasticsearch均採用集羣模式搭建，從總體上提升了ELK系統的高效性、擴展性和吞吐量。我所在的項目組採用的就是這套架構，因爲生產所需的配置較高，且涉及較多持久化操做，採用的都是性能高配的雲主機搭建方式而非時下流行的容器搭建。

source: ELK應用架構介紹

2、FileBeat服務搭建

---

日誌採集器選擇了Filebeat而不是Logstash，是因爲 Logstash 是跑在 JVM 上面，資源消耗比較大，後來做者用 GO 寫了一個功能較少可是資源消耗也小的輕量級的 Agent 叫 Logstash-forwarder，後來更名爲FileBeat。關於ELK中Filebeat的原理介紹和詳細配置我另外寫了一篇文章，戳它：《FileBeat原理與實踐指南》。

1.filebeat.yml配置

最核心的部分在於FileBeat配置文件的配置，須要指定paths（日誌文件路徑），fileds（日誌主題），hosts（kafka主機ip和端口），topic（kafka主題），version（kafka的版本），drop_fields（捨棄沒必要要的字段），name（本機IP）

filebeat.inputs:

- type: log

enabled: true

paths:

- /wls/applogs/rtlog/app.log

fields:

log_topic: appName

multiline:

# pattern for error log, if start with space or cause by

pattern: '^[[:space:]]+(at|.{3})|^Caused by:'

negate: false

match: after

output.kafka:

enabled: true

hosts: ["kafka-1:9092","kafka-2:9092"]

topic: applog

version: "0.10.2.0"

compression: gzip

processors:

- drop_fields:

fields: ["beat", "input", "source", "offset"]

logging.level: error

name: app-server-ip

2.經常使用運維指令

• 終端啓動（退出終端或ctrl+c會退出運行）

./filebeat -e -c filebeat.yml

• 之後臺守護進程啓動啓動filebeats

nohup ./filebeat -e -c filebeat.yml &

• 確認配置再也不修改，可用以下命令

//能夠防止日誌爆盤，將全部標準輸出及標準錯誤輸出到/dev/null空設備，即沒有任何輸出信息。

nohup ./filebeat -e -c filebeat.yml >/dev/null 2>&1 &

• 中止運行FileBeat進程

ps -ef | grep filebeat

Kill -9 線程號

3.FileBeat調試

當FileBeat在服務主機採集應用日誌並向Kafka輸出日誌時能夠經過兩個步驟驗證Filebeat的採集輸送是否正常：

• 採集驗證：終端執行命令，查看控制檯輸出，若是服務有異常會直接打印出來並自動中止服務。

./filebeat -e -c filebeat.yml

• 接收驗證：Kafka集羣控制檯直接消費消息，驗證接收到的日誌信息。

./kafka-console-consumer.sh --zookeeper zk-1:2181,zk-2:2181 --topic app.log

• ElasticSearch或者Kibana驗證。若是已經搭建了ELK平臺，可根據上傳的日誌關鍵屬性，於KB或者ES平臺查看是否有日誌流輸入或者在search框中根據host.name/log_topic關鍵屬性來查看是否有落庫。

source: FileBeat原理與實踐指南

3、Kafka集羣搭建

---

一個典型的Kafka集羣包含若干Producer，若干broker、若干Consumer Group，以及一個Zookeeper集羣。Kafka經過Zookeeper管理集羣配置，選舉leader，以及在Consumer Group發生變化時進行rebalance。Producer使用push模式將消息發佈到broker，Consumer使用pull模式從broker訂閱並消費消息。關於Kafka的原理和使用能夠參考《Kafka學習筆記》，下面就提供了一個典型的Kafka+ZooKeeper集羣：

1.Kafka配置

生產環境中 Kafka 集羣中節點數量建議爲（2N + 1 ）個，Zookeeper集羣一樣建議爲（2N+1）個，這邊就都以 3 個節點舉例，修改kafka集羣的配置文件，以broker1爲例進行配置：

$ vim ./config/server.properties

broker.id=1

port=9092

host.name=192.168.0.1

num.replica.fetchers=1

log.dirs=/opt/kafka_logs

num.partitions=3

zookeeper.connect=zk-1:2181,zk-2:2181,zk-3:2181

zookeeper.connection.timeout.ms=6000

zookeeper.sync.time.ms=2000

num.io.threads=8

num.network.threads=8

queued.max.requests=16

fetch.purgatory.purge.interval.requests=100

producer.purgatory.purge.interval.requests=100

delete.topic.enable=true

這裏比較重要的一個參數配置就是：num.partitions

Kafka中的topic是以partition的形式存放的，每個topic均可以設置它的partition數量，Partition的數量決定了組成topic的log的數量。推薦partition的數量必定要大於同時運行的consumer的數量。另外，建議partition的數量大於集羣broker的數量，這樣消息數據就能夠均勻的分佈在各個broker中。

-delete.topic.enable：在0.8.2版本以後，Kafka提供了刪除topic的功能，可是默認並不會直接將topic數據物理刪除。若是要從物理上刪除（即刪除topic後，數據文件也會一同刪除），就須要設置此配置項爲true。

2.Kafka運維命令

這裏涉及到topic主題的建立、與filebeats調試消息的狀態，須要掌握幾個有用的運維指令：

• 查看topic狀態
• ./kafka-topics.sh --describe --zookeeper zk-1:2181,zk-2:2181,zk-3:2181 --topic app.log
• 查看全部topic列表：
• sh kafka-topics.sh --zookeeper --zookeeper zk-1:2181,zk-2:2181,zk-3:2181 --list
• 建立topic
• sh kafka-topics.sh --zookeeper --zookeeper zk-1:2181,zk-2:2181,zk-3:2181 --create --topic app.log --partitions 5 --replication-factor 2
• 注意：server.properties 設置 delete.topic.enable=true
• 刪除主題數據
• ./bin/kafka-topics.sh --delete --zookeeper zk-1:2181,zk-2:2181,zk-3:2181 --topic app.log
• 生產topic的消息
• ./kafka-console-producer.sh --broker-list kafka-1:9092 kafka-2:9092 --topic app.log
• 消費topic的消息
• ./kafka-console-consumer.sh --zookeeper zk-1:2181,zk-2:2181,zk-3:2181 --topic app.log

3.Kafka服務監控

經過如下命令啓動了Kafka集羣服務之後，嘗試建立主題、打印主題列表查看服務狀態。

$ ./bin/kafka-server-start.sh -daemon ./config/server.properties

同時也能夠登陸Kafka集羣中任意兩臺broker主機，分別用統一的主題進行消息的生產與消費，若是服務正常則消息兩端能夠互通：

4、LogStash

---

Logstash是一個開源的、服務端的數據處理pipeline（管道），它能夠接收多個源的數據、而後對它們進行轉換、最終將它們發送到指定類型的目的地。Logstash是經過插件機制實現各類功能的，能夠在https://github.com/logstash-plugins 下載各類功能的插件，也能夠自行編寫插件。

Logstash的數據處理過程主要包括：Inputs, Filters, Outputs 三部分， 另外在Inputs和Outputs中可使用Codecs對數據格式進行處理。這四個部分均以插件形式存在，在logstash.conf配置文件中設置須要使用的input，filter，output, codec插件，以實現特定的數據採集，數據處理，數據輸出等功能

• Inputs：用於從數據源獲取數據，常見的插件如file, syslog, redis, beats 等[詳細參考]
• Filters：用於處理數據如格式轉換，數據派生等，常見的插件如grok, mutate, drop, clone, geoip等[詳細參考]
• Outputs：用於數據輸出，常見的插件如elastcisearch，file, graphite, statsd等[詳細參考]
• Codecs：Codecs不是一個單獨的流程，而是在輸入和輸出等插件中用於數據轉換的模塊，用於對數據進行編碼處理，常見的插件如json，multiline

本實例中input從kafka中獲取日誌數據，filter主要採用grok、date插件，outputs則直接輸出到elastic集羣中。logstash的配置文件是能夠自定義的，在啓動應用時須要制定相應的配置文件。

$ vim logstash.conf

input {

kafka {

type => "kafka"

bootstrap_servers => "kafka-1:9092,kafka-2:9092,kafka-3:9092"

topics => "app.log"

consumer_threads => 2

codec => "json"

}

}

filter {

grok {

match => [

#涉及多個正則匹配的寫法

"message","%{HTTPDATE:timestamp}",

"message","%{COMBINEDAPACHELOG}"

]

}

date {

match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]

}

}

output {

elasticsearch {

host => ["es-1:9300","es-2:9300"，"es-3:9300"]

index => "applogs-%{+YYYY.MM.dd}"

}

}

對上述參數進行說明：

input，須要指明是kafka來源，broker的ip和端口，主題，codec模式爲json（由於通過filebeat採集而來的數據都json化了）

filter，grok是一個十分強大的logstash filter插件，經過正則解析任意文本，將非結構化日誌數據弄成結構化和方便查詢的結構。

output，指定了輸出到ES集羣，host這裏寫ES集羣的客戶端節點便可，index則是對應ES裏的檢索，通常以【topic+日期】便可。

可是每每複雜的日誌系統這些仍是不夠，須要加一些特殊處理如：異常堆棧須要合併行、控制檯調試等。

• 蒐集日誌時涉及異常堆棧的合併行處理時，能夠加上；若是Filebeat已做合併處理此處則不須要了：

input {

stdin {

codec => multiline {

pattern => "^["

negate => true

what => "previous"

}

}

}

• 控制檯調試過濾器。
• 不少時候咱們須要調試本身的正則表達式是否可用，官方的在線調試並很差用，那麼能夠經過本身生成的json數據來校驗正則的效果，count指定重複生成的次數，message則是待調試的內容：

input {

generator {

count => 1

message => '{"key1":"value1","key2":[1,2],"key3":{"subkey1":"subvalue1"}}'

codec => json

}

}

rubydebug指明瞭輸出內容到控制檯：

output {

stdout {

codec => rubydebug

}

}

filter插件由用戶自定義填寫，啓動測試並檢查接口，每次調試都要啓動一次服務可能會須要等待幾秒鐘才輸出內容到控制檯。

./logstash -f /wls/logstash/config/logstash-test.conf

關於LogStash的語法詳解和實踐指南能夠參考：《Logstash 最佳實踐》

5、Elastic集羣搭建

---

在ElasticSearch的架構中，有三類角色，分別是Client Node、Data Node和Master Node，搜索查詢的請求通常是通過Client Node來向Data Node獲取數據，而索引查詢首先請求Master Node節點，而後Master Node將請求分配到多個Data Node節點完成一次索引查詢。

Master Node：主要用於元數據(metadata)的處理，好比索引的新增、刪除、分片分配等，以及管理集羣各個節點的狀態。因爲數據的存儲和查詢都不會走主節點，因此主節點的壓力相對較小，所以主節點的內存分配也能夠相對少些，可是主節點倒是最重要的，由於一旦主節點宕機，整個elasticsearch集羣將不可用。因此必定要保證主節點的穩定性。

Data Node：數據節點，這些節點上保存了數據分片。它負責數據相關操做，好比分片的CRUD、搜索和整合等操做。數據節點上面執行的操做都比較消耗CPU、內存和I/O資源，數據節點服務器要選擇較好的硬件配置。

Client Node：客戶端節點。client node存在的好處是能夠分擔data node的一部分壓力，由於elasticsearch的查詢是兩層匯聚的結果，第一層是在data node上作查詢結果匯聚，而後把結果發給client node，client node接收到data node發來的結果後再作第二次的匯聚，而後把最終的查詢結果返回給用戶。這樣，client node就替data node分擔了部分壓力。

1.集羣配置

第一步即定義修改es集羣的配置文件：

$ vim config/elasticsearch.yml

cluster.name: es

node.name: es-node1

node.master: true

node.data: true

network.host: 192.168.0.1

discovery.zen.ping.unicast.hosts: ["192.168.0.2","192.168.0.3"]

discovery.zen.minimum_master_nodes: 2

集羣重要配置項

• node.name 能夠配置每一個節點的名稱
• node.master 能夠配置該節點是否有資格成爲主節點。若是配置爲 true，則主機有資格成爲主節點，配置爲 false 則主機就不會成爲主節點，能夠去當數據節點或負載均衡節點
• node.data 能夠配置該節點是否爲數據節點，若是配置爲 true，則主機就會做爲數據節點，注意主節點也能夠做爲數據節點
• discovery.zen.ping.unicast.hosts 能夠配置集羣的主機地址，配置以後集羣的主機之間能夠自動發現，須要剔除本身。
• discovery.zen.minimum_master_nodes: 爲了防止集羣發生「腦裂」，一般須要配置集羣最少主節點數目，一般爲 (主節點數目 / 2) + 1

2.服務啓停

經過 -d 來後臺啓動

$ ./bin/elasticsearch -d

打開網頁 http://192.168.0.1:9200/, 若是出現下面信息說明配置成功

{

name: "es-node1",

cluster_name: "es",

cluster_uuid: "XvoyA_NYTSSV8pJg0Xb23A",

version: {

number: "6.2.4",

build_hash: "ccec39f",

build_date: "2018-04-12T20:37:28.497551Z",

build_snapshot: false,

lucene_version: "7.2.1",

minimum_wire_compatibility_version: "5.6.0",

minimum_index_compatibility_version: "5.0.0"

},

tagline: "You Know, for Search"

}

集羣服務健康情況檢查，能夠再任意節點經過執行以下命令，若是能發現列表展現的主節點、客戶端和數據節點都是一一對應的，那麼說明集羣服務都已經正常啓動了。

curl "http://ip:port/_cat/nodes"

關於ES的詳細原理與實踐，推薦《Elasticsearch 權威指南》

6、Kibana

---

Kibana是一個開源的分析和可視化平臺，設計用於和Elasticsearch一塊兒工做，能夠經過Kibana來搜索，查看，並和存儲在Elasticsearch索引中的數據進行交互。kibana使用JavaScript語言編寫，安裝部署十分簡單，能夠從elastic官網下載所需的版本，這裏須要注意的是Kibana與Elasticsearch的版本必須一致，另外，在安裝Kibana時，要確保Elasticsearch、Logstash和kafka已經安裝完畢。

1.Kibana的配置

將下載的gz包解壓

$ tar -zxvf kibana-6.2.4-darwin-x86_64.tar.gz

$ mv kibana-6.2.4-darwin-x86_64.tar.gz kibana

找到配置文件kibana.yml並修改

$ vim config/kibana.yml

server.port: 5601

server.host: "192.168.0.1"

elasticsearch.url: "http://192.168.0.1:9200"

涉及到三個關鍵參數配置：

server.port: kibana綁定的監聽端口，默認是5601

server.host: kibana綁定的IP地址

elasticsearch.url: 若是是ES集羣，則推薦綁定集羣中任意一臺ClientNode便可。

本人在項目過程當中是經過Nginx配置域名來訪問Kibana的，雖然配置了映射，且在Nginx主機上curl能訪問到服務，可是域名訪問始終報404異常，後來經過添加兩項配置便可訪問：

server.basePath: "/kibana"

server.rewriteBasePath: true

2.Kibana運維命令

啓動服務：

$ nohup ./bin/kibana &

中止服務

ps -ef | grep node

kill -9 線程號

服務啓動之後能夠經過訪問：http://192.168.0.1:5601/

3.查詢數據

打開discover菜單，這也是kibanan最經常使用的功能，選擇好時間維度來過濾數據範圍：

Kibana語法查詢，能夠直接在搜索框內輸入過濾條件進行查詢：

• response:200，將匹配response字段的值是200的文檔
• message:"Quick brown fox"，將在message字段中搜索"quick brown fox"這個短語。若是沒有引號，將會匹配到包含這些詞的全部文檔，而無論它們的順序如何。
• response:200 and extension:php or extension:css 將匹配response是200而且extension是php，或者匹配extension是css而response任意，括號能夠改變這種優先級
• >, >=, <, <= 都是有效的操做符
• response:* 將匹配全部存在response字段的文檔

點開數據行便可查看具體數據，支持table視圖和Json文本兩種方式，日誌數據都存儲在message屬性中，而前面定義的name能夠查看具體的主句，log_topic則指明是來源哪一個應用：

source: Kibana（一張圖片賽過千萬行日誌）

總結

---

綜上，經過上面部署命令來實現 ELK 的整套組件，包含了日誌收集、過濾、索引和可視化的所有流程，基於這套系統實現分析日誌功能。同時，經過水平擴展 Kafka、Elasticsearch 集羣，能夠實現日均億級的日誌實時存儲與處理，可是從細節方面來看，這套系統還存着許多能夠繼續優化和改進的點：

• 日誌格式需優化，每一個系統收集的日誌格式須要約定一個標準，好比各個業務系統在定義log4j或logback日誌partern時能夠按照【時間】【級別】【全局Traceid】【線程號】【方法名】【日誌信息】統一輸出。
• Logstash的正則優化，一旦約定了日誌模式，編寫Logstash的自定義grok正則就能過濾出關鍵屬性存放於ES，那麼基於時間、traceId以及方法名的查詢則不在堆積於message，大大優化查詢效率。
• TraceId埋點優化，分佈式與微服務架構中，一個Restful請求的發起可能會通過多達十幾個系統的處理流程，任何一個環節都有error可能，須要有一個全局ID進行全鏈路追蹤，這裏須要結合Java探針把tiraceId埋入日誌模板裏，現有PinPoint、SkyWalking與ZipKin都能爲全局ID提供成熟的解決方案。
• ES存儲優化，按照線上機器的業務量來看，天天TB級的日誌數據都寫入ES會形成較大的存儲壓力，時間越久的日誌利用價值則越低，能夠按照7天有效期來自動清理ES索引優化存儲空間，參考【ES清理腳本】。
• 運維優化，一個複雜日誌平臺在運維方面有着巨大的成本，這裏涉及到了Kafka、ZooKeeper、ELK等多個集羣環境的維護，除了提供統一的集羣操做指令之外，也須要造成對整套日誌平臺環境的監控視圖。
• 性能優化，多組件、混合語言、分佈式環境與集羣林立的複雜系統，性能問題老生常談，實踐出真知，遇到了再補充！