ELK分佈式日誌處理系統搭建

組件介紹

ELK主要是由三部分組成，包括Elasticsearch、Logstash和Kibana。

1. Elasticsearch 是一個分佈式、RESTful 風格的搜索和數據分析引擎。它主要提供收集、分析、存儲數據三個功能。
2. Logstash 用於蒐集日誌，並對日誌進行分析、過濾、解析和轉換後，將處理後的日誌提供給其餘模塊使用，例如發到Elasticsearch中保存。
3. Kibana 是一個日誌可視化工具，既能夠搜索和展現存儲在Elasticsearch中的數據，也提供了一批經典的可視化功能：柱狀圖、線狀圖、餅圖、旭日圖等等。

ELK常見的幾種架構

Logstash+Elasticsearch+Kibana

這是最簡單的一種架構。

1. 經過logstash收集日誌，發送到Elasticsearch。

2. Elasticsearch分析和存儲日誌。

3. Kibana展現日誌。

這種架構雖然是官網介紹裏的方式，可是每每在生產中不多使用。

Filebeat+Logstash+Elasticsearch+Kibana

與上一種架構相比，這種架構增長了一個Filebeat模塊。

Filebeat是部署在客戶端的一個輕量的日誌收集代理。Filebeat收集日誌後，轉發到Logstash中。相對安裝多個Logstash，Filebeat消耗更少的資源，所以每每生產環境會採起這種架構方式。但缺點就是，當Logstash服務出問題，會形成日誌的丟失。

Filebeat屬於Beats工具中的一類，目前Beats包括且不限於如下5種：

(1) Packetbeat: 收集網絡流量數據。

(2) Metricbeat: 收集系統、進程和文件系統級別的CPU和內存使用狀況等數據。

(3) Filebeat: 從安全設備、雲、容器、主機上收集日誌或文件數據。

(4) Winlogbeat: 收集Window事件日誌數據。

(5) Auditbeat: 收集Linux審計框架的數據，監控文件完整性。

Filebeat+消息中間件(redis、kafka)+Logstash+Elasticsearch+Kibana

這種架構較上個架構更完善，經過增長中間件，來解耦處理過程。一方面能在流量突發增加時，起到削峯的做用，避免請求超負荷，致使系統崩潰。另外一方面，當Logstash出現故障，日誌數據仍是存在中間件中，當Logstash再次啓動，則會讀取中間件中積壓的日誌，避免了數據的丟失。

配置介紹

接下來會按照上述的第三種架構，構建系統。主機配置要求和每臺主機所安裝的軟件以下:

主機名	配置要求	系統版本
master	內存>=4G	CentOS7以上
node1	內存>=4G	CentOS7以上
node2	內存>=4G	CentOS7以上
node3	內存>=2G	CentOS7以上

軟件	軟件版本	主機名
Elasticsearch	7.4.2	master/node1/node2
Logstash	7.4.2	master
Kibana	7.4.2	master
Filtebeat	7.4.2	node3
Kafka(內置Zookeeper)	2.11-2.2.0	master/node1/node2

實施部署

準備工做

1. 肯定每臺機器上已安裝JRE8。
2. 設置域名解析。在每臺機器上編輯/etc/hosts文件，添加機器ip以及對應的域名，如如下內容：

192.168.133.174 master
192.168.133.148 node1
192.168.133.143 node2
192.168.133.153 node3

搭建Elasticsearch集羣

1. ES運行依賴JRE,在機器上安裝JRE 8。

2. 建立普通用戶，用於運行ES，官方推薦不使用root用戶啓動。

adduser esuser
passwd esuser

3. 將ES安裝包上傳到master,node1,node2機器上，並進行解壓。

4. 編輯部署包conf/elasticsearch.yml文件，添加如下配置信息：

cluster.name: es-cluster #集羣名稱，保證惟一
node.name: node-1 #節點名稱，每臺必須不同
node.master: true #該節點是否能夠成爲master節點
node.data: true #該節點是否能夠做爲data節點，通常master節點不配置爲data節點，防止master ES壓力過大
path.data: /opt/elk/elasticsearch-7.4.2/data #數據存儲目錄
path.logs: /opt/elk/elasticsearch-7.4.2/logs #日誌存儲目錄
network.host: master #本機的ip地址，因爲前面作了解析，所以能夠用名字代替
http.port: 9200 #服務端口號，若是在同一機器裏部署多個ES則必須不同
transport.tcp.port: 9300 #集羣間通訊端口號，若是在同一機器裏部署多個ES則必須不同
#設置集羣自動發現機器ip集合
discovery.zen.ping.unicast.hosts: ["master:9300", "node1:9300", "node2:9300"]

5. 編輯部署包中config/jvm.options文件，設置堆內存大小，默認爲1G，能夠根據狀況修改:

-Xms1g
-Xmx1g

須要注意的是：

確保堆內存的最大值(Xmx)和最小值(Xms)相同，防止程序在運行時改變堆內存的大小，影響性能。

若是系統內存足夠，在64位系統中，也不能超過32G，最多可設置爲31G。

堆內存的大小，不要超過系統內存的50%。

6. 系統配置修改

（1）配置內存最大映射數,編輯/etc/sysctl.conf文件，添加如下配置後，運行sysctl -p,使配置生效：

vm.max_map_count=655360

（2） 增長最大文件打開數，編輯/etc/security/limits.conf，添加如下內容：

* - nofile 65536

（3）增長最大進程數等，編輯/etc/security/limits.conf，添加如下內容：

* soft nofile 65536
* hard nofile 131072
* soft nproc 2048
* hard nproc 4096

7. 切換到esuser用戶，運行./bin/elasticsearch，啓動master,node1,node2的ES。若是須要後臺啓動，則使用./bin/elasticsearch -d。

部署Kibana

1. 將Kibana安裝包上傳到master機器上，並進行解壓。

2. 修改配置文件，添加如下配置：

server.host: "master" #kibana主機地址，默認爲localhost
server.port: 5601 #kibana服務端口，默認爲5601
elasticsearch.hosts: ["http://master:9200"] #用於作查詢的ES節點的URL
kibana.index: ".kibana" #kibana在ES中使用縮影來存儲保存的searches,visualizations和dashboards，默認爲.kibana

3. 切換到esuser用戶，運行./bin/kibana，啓動服務，若是須要後臺啓動，則使用 nohup ./bin/kibana &。

4. 訪問kibana的地址，進入Stack monitoring,能夠看到ES的服務狀況：

ES集羣用顏色表示當前節點和分片狀態，意思以下：

green:主分片和副本分片均可用
yellow:部分副本分片不可用
red:部分主分片和副本分片不可用

部署Kafka和Zookeeper

1. 將Kafka安裝包上傳到master、node一、node2機器上，並進行解壓。

2. Kafka使用Zookeeper來管理配置，以及master選舉。這裏使用Kafka自帶的Zookeeper來安裝配置。

3. 配置Zookeeper，修改kafka部署包下的config/zookeeper.properties文件。

#zk數據存放目錄
dataDir=/opt/kafka_cluster/kafka_2.11-2.2.0/data
#zk日誌存放目錄
dataLogDir=/opt/kafka_cluster/kafka_2.11-2.2.0/logs
#客戶端鏈接zk服務的端口
clientPort=2181
#zk服務器之間，或者客戶端與服務器之間維持心跳的間隔時間
tickTime=2000
#容許follower(相對於Leader而言的"客戶端")鏈接並同步到Leader的初始化鏈接時間，以tickTime爲單位。超過該值，表示鏈接失敗
initLimit=20
 # Leader與follower之間發送消息時，請求與應答時間長度，若是follower在設置時間內不能與leader通訊，那麼該follower會被丟棄
syncLimit=10
 #kafka集羣的ip和port，2888是follower與leader交換信息的端口，3888是當leader故障，用於執行選舉時，服務器相互通訊的端口
server.1=master:2888:3888
server.2=node1:2888:3888
server.3=node2:2888:3888

4. 在zk數據存放目錄，建立一個myid文件，每臺kafka機器都必須惟一，能夠按順序，在每份文件中，分別錄入連續的數字便可。

5. 配置kafka， 修改kafka部署包下的config/server.properties文件。

#當前機器在集羣中的惟一標識，和zookeeper的myid性質同樣,每臺kafka機器都必須惟一
broker.id=1
#服務地址，注意每一個都要不一樣
listeners=PLAINTEXT://master:9092
# 接收和發送網絡信息的線程數
num.network.threads=3
#borker進行I/O處理的線程數
num.io.threads=8
#發送緩衝區buffer大小，數據不是一會兒就發送的，先回存儲到緩衝區了到達必定的大小後在發送，能提升性能
socket.send.buffer.bytes=102400
#kafka接收緩衝區大小，當數據到達必定大小後在序列化到磁盤
socket.receive.buffer.bytes=102400
#向kafka請求消息或者向kafka發送消息的請請求的最大數，這個值不能超過java的堆棧大小，防止OOM
socket.request.max.bytes=104857600
#消息存放的目錄，這個目錄能夠配置爲「，」逗號分割的表達式，需小於num.io.threads
log.dirs=/opt/kafka_cluster/kafka_2.11-2.2.0/kafkalogs/
#分區數，一個topic默認1個分區數
num.partitions=1
#在啓動時恢復日誌、關閉時刷盤日誌每一個數據目錄的線程數量，默認1
num.recovery.threads.per.data.dir=1
transaction.state.log.replication.factor=1
#日誌文件刪除以前的保存時間(單位小時)，168小時，7天
log.retention.hours=168
#單個日誌文件的大小，因爲kafka的消息是以追加的形式落地到文件，當超過這個值的時候，kafka會新起一個文件
log.segment.bytes=1073741824
#每隔300000毫秒去檢查上面配置的log失效時間（log.retention.hours=168 ），到目錄查看是否有過時的消息若是有，則根據保留策略進行刪除
log.retention.check.interval.ms=300000
#是否啓用log壓縮，通常不用啓用，啓用的話能夠提升性能
log.cleaner.enable=false
 #設置zookeeper的鏈接地址，用逗號分隔
zookeeper.connect=master:2181,node1:2181,node2:2181
#鏈接到zookeeper的超時時間
group.initial.rebalance.delay.ms=0

6. 運行如下命令，啓動ZK集羣。

nohup ./bin/zookeeper-server-start.sh config/zookeeper.properties &

7. 能夠運行命令lsof -i:2181，查看端口是否監聽。

8. 運行如下命令，啓動Kafka集羣。

nohup ./bin/kafka-server-start.sh config/server.properties &

9. 能夠運行如下命令，進行驗證：

在其中一臺機器上，運行如下命令，建立topic。

./bin/kafka-topics.sh --create --zookeeper master:2181 --replication-factor 1 --partitions 1 --topic testtopic

在每臺機器上，運行如下命令，看是否查出相同的topic。

#列出topic
./bin/kafka-topics.sh --zookeeper master:2181 --list

10. 使用Kafka Manager 工具，監控Kafka 集羣。

部署Logstash

1. 將Logstash安裝包上傳到master機器上，並進行解壓。

2. 配置Logstash，建立my_conf目錄，添加input、filter、output等配置文件到my_conf目錄下。

# input.conf文件內容,用於從kafka的`nginx`這個topic中獲取數據
input{
    kafka{
        type => "nginx_kafka"
        codec => "json"
        topics => "nginx"
        decorate_events => true
        bootstrap_servers => "node1:9092,node2:9092,master:9092"
    }
}

# output.conf文件內容，用於寫入數據到ES的index中，index的前綴爲`logstash-nginx-`,後半部分爲日期。
output {
    if[type] == "nginx_kafka" {
        elasticsearch {
            hosts => ["node1","node2","master"]
            index => 'logstash-nginx-%{+YYYY-MM-dd}'
        }
    }
}

3. 啓動Logstash。

./bin/logstash -f ./my_conf/ --config.reload.automatic

部署Filebeat

Filebeat須要部署在每臺應用服務器上。

1. 將Filebeat安裝包上傳到node3機器上，並進行解壓。

2. 修改配置文件filebeat.yml，此處使用nginx服務器日誌（搭建nginx不在本文贅述），配置nginx日誌路徑做爲輸入源，並輸出日誌到kafka集羣中，保存在topic "nignx"下。

filebeat.inputs:
- input_type: log
  paths:
    - /opt/logs/server/access.log

output.kafka:
  hosts: ["node1:9092","node2:9092","master:9092"]
  topic: 'nginx'

3. 啓動Filebeat。

./filebeat -e -c filebeat.yml

配置Kibana的Index

爲了將ES中保存的nginx日誌取出，還須要配置Kibana的索引，用於查看ES日誌。

1. 使用瀏覽器登陸Kibana, 在"Management"中，配置Kibana的"Index Patterns」，根據保存在ES中的Index格式，咱們能夠配置爲"logstash-nginx*"。

2. 配置完成後，便可在"Discover"中，使用該Kibana的index，查詢和過濾ES中的日誌數據了。

後續進階

目前僅完成從nginx中獲取日誌，後續能夠對日誌信息進行切割、字段過濾，格式化成多個字段再保存。這樣在Kibana上查看日誌能更加方便。