萬字長文：ELK（V7）部署與架構分析

時間 2019-11-09

原文原文鏈接

ELK（7版本）部署與架構分析

1、ELK的背景介紹與應用場景

在項目應用運行的過程當中，每每會產生大量的日誌，咱們每每須要根據日誌來定位分析咱們的服務器項目運行狀況與BUG產生位置。通常狀況下直接在日誌文件中tailf、 grep、awk 就能夠得到本身想要的信息。但在規模較大的場景中，此方法效率低下，面臨問題包括日誌量過大、文本搜索太慢、如何多維度查詢。這就須要對服務器上的日誌收集彙總。常看法決思路是創建集中式日誌收集系統，將全部節點上的日誌統一收集，管理，訪問。前端

通常大型系統每每是一種分佈式部署的架構，不一樣的服務模塊部署在不一樣的服務器上，問題出現時，大部分狀況須要根據問題暴露的關鍵信息，定位到具體的服務器和服務模塊，因此構建一套集中式日誌系統，能夠提升定位問題的效率。一個完整的集中式日誌系統，須要包含如下幾個主要特色：java

收集－可以採集多種來源的日誌數據，服務日誌與系統日誌。
傳輸－可以穩定的把日誌數據傳輸到中央系統
存儲－如何存儲日誌數據，持久化數據。
分析－能夠支持 UI 分析，界面化定製查看日誌操做。

ELK提供了一整套解決方案，而且都是開源軟件，之間互相配合使用，完美銜接，高效的知足了不少場合的應用。是目前主流的一種日誌系統。node

2、ELK簡介：

ELK是三個開源軟件的縮寫，分別表示：Elasticsearch , Logstash, Kibana , 它們都是開源軟件。其中後來新增了一個FileBeat。nginx

即ELK主要由Elasticsearch（搜索）、Logstash（收集與分析）和Kibana（展現）三部分組件組成；git

其中各組件說明以下：github

Filebeat：輕量級數據收集引擎。早期的ELK架構中使用Logstash收集、解析日誌，可是Logstash對內存、cpu、io等資源消耗比較高。若是用它來對服務器進行日誌收集，將加劇服務器的負載。相比 Logstash，Beats所佔系統的CPU和內存幾乎能夠忽略不計，因此filebeat做爲一個輕量級的日誌收集處理工具(Agent)，它能夠用來替代Logstash，因爲其佔用資源少，因此更適合於在各個服務器上搜集日誌後傳輸給Logstash，這也是官方推薦的一種作法。【收集日誌】web

Logstash：數據收集處理引擎。支持動態的從各類數據源蒐集數據，並對數據進行過濾、分析、豐富、統一格式等操做，而後存儲以供後續使用。【對日誌進行過濾、分析】

Elasticsearch：分佈式搜索引擎。是基於Lucene的開源分佈式搜索服務器，具備高可伸縮、高可靠、易管理等特色。能夠用於全文檢索、結構化檢索和分析，並能將這三者結合起來。【蒐集、分析、存儲數據】正則表達式

Kibana：可視化平臺。它可以搜索、展現存儲在 Elasticsearch 中索引數據。使用它能夠很方便的用圖表、表格、地圖展現和分析數據。【圖形化展現日誌】redis

結合以上，常見的日誌系統架構圖以下：docker

如上圖所示，日誌文件分別由filebeat在服務器上進行收集，收集的日誌文件彙總到logstash上並對文件數據進行過濾、分析、豐富、統一格式等操做，而後發送到Elasticsearch，進一步對日誌進行結構化檢索和分析，並存儲下來，最後由kibana進行展現。

這只是日誌系統的一種最初級結構，生產環境中須要對此結構進行進一步的優化。

3、ELK系統的部署（配置將在生產環境架構中進行說明）

環境：CentOS7.5部署ELK 7版本

準備工做： CentOS7.5

elasticsearch7， logstash7, kibana7

關閉防火牆和SELinux並更新yum源（非必須）：yum -y update

本次分佈式部署的ELK版本爲2019年五月份左右最新發布的版本，更新了許多新特性，後面將詳細說明。在安裝上面本次極大簡化了安裝步驟，能夠源碼，組件安裝，本次採用YUM+插件docker安裝，能達到相同的效果。

3.1 Filebeat安裝與配置

#！/bin/bash
# 安裝Filebeat
rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch

cat >/etc/yum.repos.d/elk-elasticsearch.repo<<EOF
[elastic-7.x]
name=Elastic repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
EOF

yum -y install filebeat

systemctl enable filebeat
systemctl start filebeat

3.2 elasticsearch，Logstash，Kibana和Filebeat的安裝（分開部署請分別安裝三個組件）

1. java環境（7版本自帶Java）

yum -y install java

2. 導入elasticsearch PGP key文件
rpm --import https://artifacts.elastic.co/GPG-KEY-elasticsearch

3. 配置yum源
vim /etc/yum.repos.d/elasticsearch.repo
[elasticsearch-7.x]
name=Elasticsearch repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md

4. yum -y install elasticsearch logstash kibana

##########關於安裝環境，也能夠參照如下二種方式，也比較簡便#################

----即經過RPM包的方式，此處給出Filebeat的方式，其餘組件雷同

curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-7.0.0-x86_64.rpm

sudo rpm -vi filebeat-7.0.0-x86_64.rpm

----Logstash實例另外一種方式（優先採用）

wget https://artifacts.elastic.co/downloads/logstash/logstash-7.0.0.rpm
yum install -y java
yum install -y logstash-7.0.0.rpm

##服務啓動前必看，因爲配置文件在下一章節給出，此處是前提準備，等配置完成後再啓動##

3.3 關於elasticsearch須要重點關注的地方

1.elasticsearch在分詞方面，須要添加中文分詞的插件

在其安裝代碼的plugins目錄，即/usr/share/elasticsearch/plugins，須要增長中文分詞插件。

[root@192-168-108-35 plugins]# ll
total 3192
[root@192-168-108-35 plugins]# pwd
/usr/share/elasticsearch/plugins
[root@192-168-108-35 plugins]# wget https://codeload.github.com/medcl/elasticsearch-analysis-ik/tar.gz/v7.0.0

下載後重啓時須要刪除插件源文件，不然報錯以下：

Caused by: java.nio.file.FileSystemException: /usr/share/elasticsearch/plugins/v7.0.0/plugin-descriptor.properties: Not a directory
因此安裝完成後必須刪除源文件v7.0.0

2.elasticsearch服務的啓動問題與前提準備

-- 首先在安裝完畢後會生成不少文件，包括配置文件日誌文件等等，下面幾個是最主要的配置文件路徑
/etc/elasticsearch/elasticsearch.yml # els的配置文件
/etc/elasticsearch/jvm.options # JVM相關的配置，內存大小等等
/etc/elasticsearch/log4j2.properties # 日誌系統定義

/usr/share/elasticsearch # elasticsearch 默認安裝目錄
/var/lib/elasticsearch # 數據的默認存放位置

-- 建立用於存放數據與日誌的目錄
數據文件會隨着系統的運行飛速增加，因此默認的日誌文件與數據文件的路徑不能知足咱們的需求，那麼手動建立日誌與數據文件路徑。

mkdir -p /data/elkdata
mkdir -p /data/elklogs

-- JVM配置（7.0版本針對此作出優化，能夠基本保障溢出問題，但最好設置一下）
因爲Elasticsearch是Java開發的，因此能夠經過/etc/elasticsearch/jvm.options配置文件來設定JVM的相關設定。若是沒有特殊需求按默認便可。
不過其中仍是有兩項最重要的-Xmx1g與-Xms1gJVM的最大最小內存。若是過小會致使Elasticsearch剛剛啓動就馬上中止。太大會拖慢系統自己。
vim /etc/elasticsearch/jvm.options # JVM最大、最小使用內存

-Xms1g

-Xmx1g

-- 使用ROOT帳戶執行命令
elasticsearch的相關配置已經完成，下面須要啓動elasticsearch集羣。可是因爲安全的考慮，elasticsearch不容許使用root用戶來啓動，因此須要建立一個新的用戶，併爲這個帳戶賦予相應的權限來啓動elasticsearch集羣。

建立ES運行用戶
# 建立用戶組 group add elk

# 建立用戶並添加至用戶組 useradd elk -g elk

# （可選）更改用戶密碼

passwd elasticsearch

同時修改ES目錄權限，如下操做都是爲了賦予es用戶操做權限，

-- 安裝源碼文件目錄
[root@192-168-108-35 share]# chown -R elk :elk /usr/share/elasticsearch/
[root@192-168-108-35 share]# chown -R elk :elk /var/log/elasticsearch/
[root@192-168-108-35 share]# chown -R elk :elk /data

-- 運行常見的報錯信息

[1]文件數目不足
# 修改系統配置文件屬性

# vim /etc/security/limits.conf 添加

elk soft memlock unlimited

elk hard memlock unlimited

elk soft nofile 65536

elk hard nofile 131072
退出用戶從新登陸，使配置生效

[2]: max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]
解決辦法：
在 /etc/sysctl.conf文件最後添加一行
vm.max_map_count=262144
便可永久修改

3. 啓動elasticsearch服務

前臺啓動服務
# 需切換爲es用戶

su elk

# 啓動服務（當前的路徑爲：/usr/share/elasticsearch/）

./bin/elasticsearch

後臺運行ES
能夠加入-p 命令讓es在後臺運行， -p 參數記錄進程ID爲一個文件
# 設置後臺啓動
./bin/elasticsearch -p ./elasticsearch-pid -d

通常狀況下，直接執行一下命令便可

./bin/elasticsearch -d

結束進程
# 查看運行的pid，並查殺 cat /tmp/elasticsearch-pid && echo

kill -SIGTERM {pid}

# 暴力結束進程（另外一種方式）
kill -9 `ps -ef |grep elasticsearch|awk '{print $2}' `

驗證一下服務是否正常
curl -i "http://192.168.60.200:9200"

4.安裝elasticsearch-head插件（支持前端界面查看數據）
安裝docker鏡像或者經過github下載elasticsearch-head項目都是能夠的，1或者2兩種方式選擇一種安裝使用便可

【1】使用docker的集成好的elasticsearch-head
# docker run -p 9100:9100 mobz/elasticsearch-head:5
docker容器下載成功並啓動之後，運行瀏覽器打開http://localhost:9100/

【2】使用git安裝elasticsearch-head
   # yum install -y npm
   # git clone git://github.com/mobz/elasticsearch-head.git
   # cd elasticsearch-head
   # npm install
   # npm run start
   檢查端口是否起來
   netstat -antp |grep 9100
   瀏覽器訪問測試是否正常
   http://IP:9100/

【注意】因爲elasticsearch-head:5鏡像對elasticsearch的7版本適配性未經檢測。這裏也推薦推薦另一個鏡像lmenezes/cerebro，下載後執行

docker run -d -p 9000:9000 lmenezes/cerebro

就能夠在9000端口查看了。

最後，filebeat logstash kibana能夠在配置文件後，正常啓動，若是須要切換用戶的話，也能夠參照上面。本次這三個組件所有默認用root用戶啓動。因爲分開部署，與ES互不影響。

附：kafka三節點集羣搭建

環境準備

1.zookeeper集羣環境，本次採用Kafka自帶的Zookeeper環境。
kafka是依賴於zookeeper註冊中心的一款分佈式消息對列，因此須要有zookeeper單機或者集羣環境。

2.三臺服務器：
192.168.108.200 ELK-kafka-cluster

192.168.108.165 ELK-kafka-salve1

192.168.108.103 ELK-kafka-salve2

3.下載kafka安裝包

在 http://kafka.apache.org/downloads 中下載，目前最新版本的kafka已經到2.2.0,這裏下載的是kafka_2.11-2.2.0.tgz

安裝kafka集羣

1.上傳壓縮包到三臺服務器解壓縮到/opt/目錄下

tar -zxvf kafka_2.11-2.2.0.tgz -C /opt/

ls -s kafka_2.11-2.2.0 kafka

2.修改 server.properties （/opt/kafka/config目錄下）

############################# Server Basics #############################

broker.id=0

######################## Socket Server Settings ########################

listeners=PLAINTEXT://192.168.108.200:9092
advertised.listeners=PLAINTEXT://192.168.108.200:9092

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

############################# Log Basics #############################

log.dirs=/var/log/kafka-logs

num.partitions=1

num.recovery.threads.per.data.dir=1

######################## Internal Topic Settings #########################
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1

######################### Log Retention Policy ########################

# The minimum age of a log file to be eligible for deletion due to age
log.retention.hours=168

# The maximum size of a log segment file. When this size is reached a new log segment will be created.
log.segment.bytes=1073741824

# The interval at which log segments are checked to see if they can be deleted according
# to the retention policies
log.retention.check.interval.ms=300000

############################# Zookeeper #############################

zookeeper.connect=192.168.108.200:2181,192.168.108.165:2181,192.168.108.103:2181

# Timeout in ms for connecting to zookeeper
zookeeper.connection.timeout.ms=6000
delete.topic.enable=true

######################## Group Coordinator Settings ##########################

group.initial.rebalance.delay.ms=0

3.拷貝兩份到192.168.108.165和192.168.108.103

[root@192.168.108.165 config]# cat server.properties

broker.id=1

listeners=PLAINTEXT://192.168.108.165:9092

advertised.listeners=PLAINTEXT://192.168.108.165:9092

[root@k8s-n3 config]# cat server.properties

broker.id=2

listeners=PLAINTEXT://192.168.108.103:9092

advertised.listeners=PLAINTEXT://192.168.108.103:9092

而後添加環境變量在/etc/profile 中添加
export ZOOKEEPER_HOME=/opt/kafka

export PATH=$PATH:$ZOOKEEPER_HOME/bin
source /etc/profile 重載生效

4.修改zookeeper.properties：
一、設置鏈接參數，添加以下配置
maxClientCnxns=100
tickTime=2000
initLimit=10
syncLimit=5

二、設置broker Id的服務地址
server.0=192.168.108.200:2888:3888
server.1=192.168.108.165:2888:3888
server.2=192.168.108.103:2888:3888

總結就在Kafka源碼目錄的/kafka/config/zookeeper.properties文件設置以下

dataDir=/opt/zookeeper # 這時須要在/opt/zookeeper文件夾下，新建myid文件，把broker.id填寫進去，本次本節點爲0。
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=100
tickTime=2000
initLimit=10
syncLimit=5
server.0=192.168.108.200:2888:3888
server.1=192.168.108.165:2888:3888
server.2=192.168.108.103:2888:3888

而後在zookeeper數據目錄添加id配置（dataDir=/opt/zookeeper）
在zookeeper.properties的數據目錄中建立myid文件
在各臺服務的zookeeper數據目錄添加myid文件，寫入服務broker.id屬性值，如這裏的目錄是/usr/local/zookeeper/data
第一臺broker.id爲0的服務到該目錄下執行：echo 0 > myid

[root@192-168-108-165 kafka]# cd /opt/zookeeper/

[root@192-168-108-165 zookeeper]# ls

myid version-2

[root@192-168-108-165 zookeeper]# cat myid

[root@192-168-108-165 zookeeper]#

集羣啓動

集羣啓動：cd /opt/kafka
先分別啓動zookeeper
kafka使用到了zookeeper，所以你要首先啓動一個zookeeper服務，若是你沒有zookeeper服務。kafka中打包好了一個簡潔版的單節點zookeeper實例。
kafka啓動時先啓動zookeeper，再啓動kafka；關閉時相反，先關閉kafka，再關閉zookeeper

前臺啓動：bin/zookeeper-server-start.sh config/zookeeper.properties

[root@192-168-108-200 ~]# lsof -i:2181

COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME

java 5197 root 98u IPv6 527171 0t0 TCP *:eforward (LISTEN)
能夠看到已經啓動
後臺啓動：bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

再分別啓動kafka

前臺啓動：bin/kafka-server-start.sh config/server.properties
[root@192-168-108-200 ~]# lsof -i:9092
COMMAND PID USER FD TYPE DEVICE SIZE/OFF NODE NAME
java 5519 root 157u IPv6 528511 0t0 TCP 192-168-108-200:XmlIpcRegSvc (LISTEN)
java 5519 root 167u IPv6 528515 0t0 TCP 192-168-108-200:XmlIpcRegSvc->192.168.108.191:56656 (ESTABLISHED)
java 5519 root 169u IPv6 528516 0t0 TCP 192-168-108-200:XmlIpcRegSvc->192.168.108.187:49368 (ESTABLISHED)
java 5519 root 176u IPv6 528096 0t0 TCP 192-168-108-200:48062->192-168-108-200:XmlIpcRegSvc (ESTABLISHED)
java 5519 root 177u IPv6 528518 0t0 TCP 192-168-108-200:XmlIpcRegSvc->192-168-108-200:48062 (ESTABLISHED)
java 5519 root 178u IPv6 528097 0t0 TCP 192-168-108-200:XmlIpcRegSvc->192.168.108.187:49370 (ESTABLISHED)

後臺啓動：bin/kafka-server-start.sh -daemon config/server.properties

服務啓動腳本

cd /opt/kafka
kill -9 `ps -ef |grep kafka|awk '{print $2}' `
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
bin/kafka-server-start.sh -daemon config/server.properties

Zookeeper+Kafka集羣測試

/opt/kafka/bin
1.建立topic:
kafka-topics.sh --create --zookeeper 192.168.108.200:2181,192.168.108.165:2181,192.168.108.103:2181 --replication-factor 3 --partitions 3 --topic test

2.顯示topic
kafka-topics.sh --describe --zookeeper 192.168.108.200:2181,192.168.108.165:2181,192.168.108.103:2181 --topic test
[root@192-168-108-200 bin]# kafka-topics.sh --describe --zookeeper 192.168.108.200:2181,192.168.108.165:2181,192.168.108.103:2181 --topic tyun
Topic:tyun PartitionCount:3 ReplicationFactor:3 Configs:
Topic: tyun Partition: 0 Leader: 1 Replicas: 1,0,2 Isr: 1,0,2
Topic: tyun Partition: 1 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0
Topic: tyun Partition: 2 Leader: 0 Replicas: 0,2,1 Isr: 0,2,1

PartitionCount：partition個數
ReplicationFactor：副本個數
Partition：partition編號，從0開始遞增
Leader：當前partition起做用的broker.id
Replicas: 當前副本數據所在的broker.id，是一個列表，排在最前面的其做用
Isr：當前kakfa集羣中可用的broker.id列表

3.列出topic
kafka-topics.sh --list --zookeeper 192.168.108.200:2181,192.168.108.165:2181,192.168.108.103:2181
test

建立 producer(生產者);
kafka-console-producer.sh --broker-list 192.168.108.200:9092 --topic test
hello

建立 consumer（消費者）
kafka-console-consumer.sh --bootstrap-server 192.168.108.200:9092 --topic test --from-beginning
hello

4.查看寫入kafka集羣中的消息（重要命令，判斷日誌是否寫入Kafka的重要依據）

bin/kafka-console-consumer.sh --bootstrap-server 192.168.108.200:9092 --topic tiops --from-beginning

5.刪除 Topic----命令標記刪除後，再次刪除對應的數據目錄
bin/kafka-topics.sh --delete --zookeeper master:2181,slave1:2181,slave2:2181 --topic topic_name
若 delete.topic.enable=true
直接完全刪除該 Topic。

若 delete.topic.enable=false
若是當前 Topic 沒有使用過即沒有傳輸過信息：能夠完全刪除。

若是當前 Topic 有使用過即有過傳輸過信息：並無真正刪除 Topic 只是把這個 Topic 標記爲刪除(marked for deletion)，重啓 Kafka Server 後刪除。
注：delete.topic.enable=true 配置信息位於配置文件 config/server.properties 中(較新的版本中無顯式配置，默認爲 true)。

4. 日誌系統架構解析

整個系統一共含有10臺主機（filebeat部署在客戶端，不計算在內），其中Logstash有四臺，Elasticsearch有二臺，Kafka集羣三臺，kibana一臺並配置Nginx代理。

架構解釋：

（1）首先用戶經過nginx代理訪問ELK日誌統計平臺，這裏的Nginx能夠設置界面密碼。
（2）Nginx將請求轉發到kibana
（3）kibana到Elasticsearch中去獲取數據，這裏的Elasticsearch是兩臺作的集羣，日誌數據會隨機保存在任意一臺Elasticsearch服務器。
（4）Logstash1從Kafka中取出數據併發送到Elasticsearch中。
（5）Kafka服務器作日誌數據的持久化保存，避免web服務器日誌量過大的時候形成的數據收集與保存不一致而致使日誌丟失，其中Kafka能夠作集羣，而後再由Logstash服務器從Kafka持續的取出數據。
（6）logstash2從Filebeat取出的日誌信息，並放入Kafka中進行保存。
（7）Filebeat在客戶端進行日誌的收集。

注1：【Kafka的加入緣由與做用】

整個架構加入Kafka，是爲了讓整個系統更好的分層，Kafka做爲一個消息流處理與持久化存儲軟件，可以幫助咱們在主節點上屏蔽掉多個從節點之間不一樣日誌文件的差別，負責管理日誌端（從節點）的人能夠專一於向 Kafka裏生產數據，而負責數據分析聚合端的人則能夠專一於從 Kafka內消費數據。因此部署時要把Kafka加進去。

並且使用Kafka進行日誌傳輸的緣由還在於其有數據緩存的能力，而且它的數據可重複消費，Kafka自己具備高可用性，可以很好的防止數據丟失，它的吞吐量相對來講比較好而且使用普遍。能夠有效防止日誌丟失和防止logsthash掛掉。綜合來講：它均衡了網絡傳輸，從而下降了網絡閉塞，尤爲是丟失數據的可能性，

注2：【雙層的Logstash做用】

這裏爲何要在Kafka前面增長二臺logstash呢？是由於在大量的日誌數據寫入時，容易致使數據的丟失和混亂，爲了解決這一問題，增長二臺logstash能夠經過類型進行彙總分類，下降數據傳輸的臃腫。

若是隻有一層的Logstash，它將處理來自不一樣客戶端Filebeat收集的日誌信息彙總，而且進行處理分析，在必定程度上會形成在大規模日誌數據下信息的處理混亂，並嚴重加深負載，因此有二層的結構進行負載均衡處理，而且職責分工，一層匯聚簡單分流，一層分析過濾處理信息，而且內層都有二臺Logstash來保障服務的高可用性，以此提高整個架構的穩定性。

接下來分別說明原理與各個組件之間的交互（配置文件）。

1】Filebeat與Logstash-collect鏈接配置

這裏爲了方便記憶，把此處的Logstash稱爲Logstash-collect，首先看Filebeat的配置文件：

其中，filebeat配置輸出到logstash：5044端口，在logstash上啓動5044端口做爲logstash與filebeat的通訊agent，而logstash自己服務起在9600端口

[root@192-168-108-191 logstash]# ss -lnt
State Recv-Q Send-Q Local Address:Port Peer Address:Port
LISTEN 0 128 :::5044 :::*
LISTEN 0 50 ::ffff:192.168.108.191:9600 :::*

vim /etc/filebeat/filebeat.yml # 對幾個重要配置進行設置以下

- type: log

# Change to true to enable this input configuration.
enabled: true

  # Paths that should be crawled and fetched. Glob based paths.
  paths:
     - /var/log/*.log
     - /var/log/tiops/**/*.log      # 即tiops平臺的全部日誌位置, 指定數據的輸入路徑爲/tiops/**/*.log結尾的全部文件，注意/tiops/子目錄下的日誌不會被讀取，孫子目錄下的日誌能夠
     #- c:\programdata\elasticsearch\logs\*
   fields:
     service: filebeat
   multiline: # 多行日誌合併爲一行，適用於日誌中每一條日誌佔據多行的狀況，好比各類語言的報錯信息調用棧。
     pattern: ‘^\[‘
     negate: true
     match: after

# Exclude lines. A list of regular expressions to match. It drops the lines that are
# 刪除以DBG開頭的行：
exclude_lines: ['^DBG']

# filebeat的日誌可發送到logstash、elasticsearch、kibana，選擇一個便可，本次默認選擇logstash，其餘二個能夠註釋
  output.logstash:
   # The Logstash hosts
     hosts: ["192.168.108.191:5044", "192.168.108.87:5044"]       # 發往二臺Logstash-collect
     loadbalance: true
     worker: 2

  #output.elasticsearch:
     #hosts: ["http://192.168.108.35:9200"]
     #username: "elk"
     #password: ""

  #setup.kibana:
     #host: "192.168.108.182:5601"

注1：【Filebeat與Logstash-collect的日誌消息流轉】

注意事項：Logstash-collect怎麼接收到Filebeat發送的消息，並再次發向下一級呢？

不只僅配置發往二臺Logstash-collect的ip地址加端口，還要注意fields字段，能夠理解爲是一個Key，當Logstash-collect收到多個Key時，它能夠選擇其中一個或者多個Key來進行下一級發送，達到日誌消息的轉發。

在使用了6.2.3版本的ELK之後，若是使用if [type]配置，則匹配不到在filebeat裏面使用document_type定義的字符串。由於6.0版本以上已經取消了document_type的定義。若是要實現以上的配置只能使用以下配置：

fields:
service: filebeat

service : filebeat都是本身定義的，定義完成後使用Logstash的if 判斷，條件爲if [fields][service] == "filebeat".就能夠了，具體能夠看下面的轉發策略。

注2：【Filebeat負載平衡主機的輸出】

這裏還有一點須要說明一下，就是關於Filebeat與Logstash-collect鏈接的負載均衡設置。

logstash是一個無狀態的流處理軟件。logstash怎麼集羣配置只能橫向擴展，而後本身用配置管理工具分發，由於他們內部並無交流的。

Filebeat提供配置選項,可使用它來調整負載平衡時發送消息到多個主機。要啓用負載均衡,您指定loadbalance的值爲true。

output.logstash:
   # The Logstash hosts
     hosts: ["192.168.108.191:5044", "192.168.108.87:5044"]       # 發往二臺Logstash-collect
     loadbalance: true
     worker: 2

loadbalance: false # 消息只是往一個logstash裏發，若是這個logstash掛了，就會自動將數據發到另外一個logstash中。（主備模式）

loadbalance: true # 若是爲true，則將數據均分到各個logstash中，掛了就不發了，往存活的logstash裏面發送。

即logstash地址若是爲一個列表，若是loadbalance開啓，則負載到裏表中的服務器，當一個logstash服務器不可達，事件將被分發到可到達的logstash服務器（雙活模式）

loadbalance選項可供 Redis,Logstash, Elasticsearch輸出。Kafka的輸出能夠在其自身內部處理負載平衡。
同時每一個主機負載平衡器還支持多個workers。默認是1。若是你增長workers的數量, 將使用額外的網絡鏈接。workers參與負載平衡的總數=主機數量*workers。

在這個小節，配置整體結構，與日誌數據流向以下圖：

2】Logstash-collect與Kafka鏈接配置

首先看/etc/logstash/logstash.yml文件，做爲Logstash-collect的公共配置文件，咱們須要作一下的更改：

vim /etc/logstash/logstash.yml

path.data: /var/lib/logstash #數據存放路徑
path.config: /etc/logstash/conf.d #配置文件的讀取路徑
path.logs: /var/log/logstash #日誌文件的保存路徑
pipeline.workers: 2 # 默認爲CPU的核數
http.host: "192.168.108.186"
http.port: 9600-9700 # logstash will pick up the first available ports

注意上面配置文件的讀取路徑 /etc/logstash/conf.d，在這個文件夾裏面，咱們能夠設置Logstash的輸入輸出。logstash支持把配置寫入文件/etc/logstash/conf.d/xxx.conf，而後經過讀取配置文件來採集數據。

logstash收集日誌基本流程: input–>codec–>filter–>codec–>output ，因此配置文件能夠設置輸入輸出與過濾的基本格式以下：（這裏暫時忽略filter，由於這一層的Logstash主要匯聚數據，暫不分析匹配）

關於codec => json # json處理

logstash最終會把數據封裝成json類型，默認會添加@timestamp時間字段、host主機字段、type字段。原消息數據會整個封裝進message字段。若是數據處理過程當中，用戶解析添加了多個字段，則最終結果又會多出多個字段。也能夠在數據處理過程當中移除多個字段，總之，logstash最終輸出的數據格式是json格式。因此數據通過Logstash會增長額外的字段，能夠選擇過濾。
Logstash主要由 input，filter，output三個組件去完成採集數據。

input {

指定輸入

}

filter {

}

output {

指定輸出

}

解釋說明以下：

input
input組件負責讀取數據，能夠採用file插件讀取本地文本文件，stdin插件讀取標準輸入數據，tcp插件讀取網絡數據，log4j插件讀取log4j發送過來的數據等等。本次用 beats插件讀取Filebeat發送過來的日誌消息。

filter
filter插件負責過濾解析input讀取的數據，能夠用grok插件正則解析數據，date插件解析日期，json插件解析json等等。

output
output插件負責將filter處理過的數據輸出。能夠用elasticsearch插件輸出到es，redis插件輸出到redis，stdout插件標準輸出，kafka插件輸出到kafka等等

在實際的Tiops平臺日誌處理流程中，配置以下：

二個Logstash-collect均採用此相同的配置

在/etc/logstash/conf.d目錄下，建立filebeat-kafka.conf文件，內容以下：

input {

beats {

port => 5044 # logstash上啓動5044端口做爲logstash與filebeat的通訊agent

codec => json # json處理

}

#此處附加Redis做爲緩存消息的配置，本次採用Kafka，因此此部分暫時註釋

#output {

# if [fields][service] == "filebeat"{

# redis {

# data_type => "list"

# host => "192.168.108.200"

# db => "0"

# port => "6379"

# key => "tiops-tcmp"

# }

output {

if [fields][service] == "filebeat"{ # 參照第一部分，Logstash-collect怎麼接收到Filebeat發送的消息，並再次發向下一級呢？這裏就是取filebeat的Fields字段的Key值，進行向下發送

kafka {

bootstrap_servers => "192.168.108.200:9092,192.168.108.165:9092,192.168.108.103:9092" # kafka集羣配置，所有IP：端口都要加上

topic_id => "tiops" # 設置Kafka的topic，這樣發送到kafka中時，kafka就會自動建立該topic

}

在這個小節，配置整體結構與日誌數據流向以下圖：

3】Kafka與Logstash-grok鏈接配置

在上面第二部分，咱們已經成功把數據從Logstash-collect發送到Kafka，而Kafka自己的輸入輸出並無配置。而是Logstash-collect指定了一個topic，這並不表明kafka不須要配置參數，只是kafka自己不須要配置pull/push目標參數，它被動接受

別的生產者發送過來的數據，由於Logstash-collect指定了Kafka集羣的IP地址和端口。那麼數據將會被髮送到Kafka中，那麼Kafka如何處理這些日誌數據，那麼這時候就須要Kafka配置消息處理參數了。

這個咱們如今暫時先給出參數，至於具體緣由涉及到MQ的高可用、消息過時策略、如何保證消息不重複消費、如何保證消息不丟失、如何保證消息按順序執行以及消息積壓在消息隊列裏怎麼辦等問題，咱們以後詳細分析並解決。

如下是Kafka集羣中，某一臺的配置文件，在Kafka源碼目錄的/kafka/config/server.properties文件設置以下：（加粗的爲重要配置文件）

broker.id=0
listeners=PLAINTEXT://192.168.108.200:9092
advertised.listeners=PLAINTEXT://192.168.108.200:9092
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/var/log/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=192.168.108.200:2181,192.168.108.165:2181,192.168.108.103:2181 # zookeeper 集羣
delete.topic.enable=true

同時kafka是依賴於zookeeper註冊中心的一款分佈式消息隊列，因此須要有zookeeper單機或者集羣環境。本次採用Kafka自帶的Zookeeper環境。

在Kafka源碼目錄的/kafka/config/zookeeper.properties文件設置以下

dataDir=/opt/zookeeper # 這時須要在/opt/zookeeper文件夾下，新建myid文件，把broker.id填寫進去，本次本節點爲0。
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=100
tickTime=2000
initLimit=10
syncLimit=5
server.0=192.168.108.200:2888:3888
server.1=192.168.108.165:2888:3888
server.2=192.168.108.103:2888:3888

其餘注意事項，請看以前的Kafka集羣安裝步驟，這裏主要給出集羣某一節點的配置文件參數。

到目前爲止，咱們配置了kafka的參數，對消息進行了處理（持久化...）,接下來就須要對Kafka中的消息進行下一步的傳輸，即傳送到Logstash-grok。

這裏須要注意的是：並非Kafka主動把消息發送出去的，固然Kafka也支持這種操做，可是在這裏，採用的是Logstash-grok做爲消費者，主動拉取Kafka中的消息，來進行消費。

因此就須要像第二步Logstash-collect那樣進行輸入輸出設置，這裏Logstash-grok不只僅配置輸入輸出，最重要的是它的過濾filter做用。

在/etc/logstash/conf.d目錄下，建立logstash-es.conf文件，這裏給出Logstash-grok中一個的內容以下（二個Logstash-grok配置同樣）：

#input {

# redis {

# data_type => "list"

# host => "192.168.108.200"

# db => "0"

# port => "6379"

# key => "tiops-tcmp"

# # password => "123456"

# }

input{

kafka{

bootstrap_servers => ["192.168.108.200:9092,192.168.108.165:9092,192.168.108.103:9092"] # 配置拉取的kafka集羣消息地址

group_id => "tyun" # 設置 group_id，對於這個消費組而言會有一個消費 offset，消費掉是 auto_commit 部分控制的，這個參數會按期將你消費的 offset 提交到 kafka，下次啓動的時候已經消費過的就不會重複消費了；

# 若是想從新消費，能夠換一個 group_id 便可

auto_offset_reset => "earliest"

consumer_threads => "3" # 多個實例的consumer_threads數之和應該等於topic分區數近似達到logstash多實例的效果。

decorate_events => "false"

topics => ["tiops"] # kafka topic 名稱，獲取指定topic內的消息

type => "log"

codec => json

}

filter { # 此處Logstash最主要的過濾做用，能夠自定義正則表達式，選擇性輸出消息

grok {

match => {

#截取

"message" => "(?<LOG>(?<=architecture=x86_64}|containerized=true})(.*)/?)"

}

output { # 過濾後的消息，發送到ES集羣

elasticsearch {

hosts => ["192.168.108.35:9200","192.168.108.194:9200"]

codec => json

index => "tiops-%{+YYYY.MM.dd}" # 自定義索引名稱

}

在這個小節，配置整體結構與日誌數據流向以下圖：

4】Logstash-grok與Elasticsearch（ES）集羣鏈接配置

ElasticSearch集羣
在第三步中，咱們已經將Logstash-grok過濾後的消息，發送到了ES集羣，這裏的ES集羣被動接受發送過來的消息，即ES集羣自己不須要配置輸入源，已經由其餘組件發送決定了。

ES的配置將集中體如今集羣自身的配置上。

Elasticsearch 能夠橫向擴展至數百（甚至數千）的服務器節點，同時能夠處理PB級數據
Elasticsearch 天生就是分佈式的，而且在設計時屏蔽了分佈式的複雜性。

ES功能：（1）分佈式的搜索引擎和數據分析引擎

（2）全文檢索，結構化檢索，數據分析

（3）對海量數據進行近實時的處理

安裝後的ES集羣，配置在/etc/elasticsearch/elasticsearch.yml文件，集羣中，各個節點配置大致相同，不一樣之處下面有紅字標出，主要就是節點名稱與網絡監聽地址，須要每一個節點自行按實際填寫。

cluster.name: elk-cluster #集羣名稱
node.name: elk-node1 #節點名稱，一個集羣以內節點的名稱不能重複
path.data: /data/elkdata #數據路徑
path.logs: /data/elklogs #日誌路徑
# bootstrap.memory_lock: true #鎖住es內存，保證內存不分配至交換分區。
network.host: 192.168.108.35 #網絡監聽地址，能夠訪問elasticsearch的ip, 默認只有本機
http.port: 9200 #用戶訪問查看的端口，9300是ES組件訪問使用
discovery.seed_hosts: ["192.168.108.35","192.168.108.194"] 單播（配置一臺便可，生產可使用組播方式）
cluster.initial_master_nodes: ["192.168.108.35"] # Master
http.cors.enabled: true # 這二個是ES插件的web顯示設置，head插件防止跨域
http.cors.allow-origin: "*"

可見集羣配置中最重要的兩項是node.name與network.host，每一個節點都必須不一樣。其中node.name是節點名稱主要是在Elasticsearch本身的日誌加以區分每個節點信息。
discovery.seed_hosts是集羣中的節點信息，可使用IP地址、可使用主機名(必須能夠解析)。

當ElasticSearch的節點啓動後，它會利用多播(multicast)(或者單播），若是用戶更改了配置)尋找集羣中的其它節點，並與之創建鏈接。（節點發現）

這裏沒有配置分片的數目，在ES的7版本中，默認爲1，這個包括replicas可後來自定義設置。

在這個小節，配置整體結構與日誌數據流向以下圖：

5】Elasticsearch（ES）集羣與Kibana鏈接配置

Kibana是一個開源的分析和可視化平臺，和Elasticsearch一塊兒工做時，就能夠用Kibana來搜索，查看，並和存儲在Elasticsearch索引中的數據進行交互。
能夠輕鬆地執行高級數據分析，而且以各類圖標、表格和地圖的形式可視化數據。
同時Kibana使得理解大量數據變得很容易。它簡單的、基於瀏覽器的界面使你可以快速建立和共享動態儀表板，實時顯示Elasticsearch查詢的變化。

在第四步中，咱們已經將ES集羣中的數據進行了處理，創建了索引（至關於MySQL的database概念），而且分片、副本等都存在，並且處理後的數據更加方便於檢索。

Kibana的配置在/etc/kibana/kibana.yml文件中，以下：

server.port: 5601 #監聽端口
server.host: "192.168.108.182" #監聽IP地址，建議內網ip
#kibana.index: ".newkibana"
elasticsearch.hosts: ["http://192.168.108.35:9200","http://192.168.108.194:9200"] # ES機器的IP，這裏的全部host必須來自於同一個ES集羣
#xpack.security.enabled: false #添加這條，這條是配置kibana的安全機制，暫時關閉。

能夠看出，這裏的Kibana會主動去ES集羣中去拉取數據，最後在瀏覽器Web端進行實時展現。

這時，咱們打開瀏覽器界面以下，看看最終的結果：

注：Nginx代理配置，能夠經過自定義域名訪問並設置密碼，提高安全性。（可選）

upstream kibana_server {
        server 192.168.108.182:5601 weight=1 max_fails=3 fail_timeout=60;
    }
    server {
        listen 80;
        server_name www.kibana.com;
        auth_basic "Restricted Access";
        auth_basic_user_file /etc/nginx/conf.d/htpasswd.users;
        location / {
                proxy_pass http://kibana_server;
                proxy_http_version 1.1;
        }
    }

在這個小節，配置整體結構與日誌數據流向以下圖：