ELK實時日誌分析平臺環境部署

時間 2020-01-09

標籤 elk 實時日誌分析平臺環境部署简体版

原文原文鏈接

1、概念介紹

日誌主要包括系統日誌、應用程序日誌和安全日誌。系統運維和開發人員能夠經過日誌瞭解服務器軟硬件信息、檢查配置過程當中的錯誤及錯誤發生的緣由。常常分析日誌能夠了解服務器的負荷，性能安全性，從而及時採起措施糾正錯誤。html

一般，日誌被分散的儲存不一樣的設備上。若是你管理數十上百臺服務器，你還在使用依次登陸每臺機器的傳統方法查閱日誌。這樣是否是感受很繁瑣和效率低下。當務之急咱們使用集中化的日誌管理，例如：開源的syslog，將全部服務器上的日誌採集彙總。前端

集中化管理日誌後，日誌的統計和檢索又成爲一件比較麻煩的事情，通常咱們使用grep、awk和wc等Linux命令能實現檢索和統計，可是對於要求更高的查詢、排序和統計等要求和龐大的機器數量依然使用這樣的方法不免有點力不從心。java

經過咱們須要對日誌進行集中化管理，將全部機器上的日誌信息採集、彙總到一塊兒。完整的日誌數據具備很是重要的做用：node

信息查找。經過檢索日誌信息，定位相應的bug，找出解決方案。
服務診斷。經過對日誌信息進行統計、分析，瞭解服務器的負荷和服務運行狀態，找出耗時請求進行優化等等。
數據分析。若是是格式化的log，能夠作進一步的數據分析，統計、聚合出有意義的信息，好比根據請求中的商品id，找出TOP10用戶感興趣商品。

那麼，有沒有一種方法能解決日誌採集、統計和檢索？開源實時日誌分析ELK平臺就可以完美的解決咱們上述的問題。mysql

簡單地來講，ELK 就是 Elasticsearch、Logstash 和 Kibana 三個開源工具的首字母簡寫。linux

一、Elasticsearch

Elasticsearch是一個開源的基於Lucene的開源實時的分佈式搜索分析引擎工具。它能讓你以一個以前從未有過的速度和規模，去探索你的數據。它的特色有：分佈式，零配置，自動發現，索引自動分片，集羣配置方便等。索引副本機制，restful風格接口，多數據源，自動搜索負載等。它提供了一個分佈式多用戶能力的全文搜索引擎，基於RESTful web接口。Elasticsearch是用Java開發的，並做爲Apache許可條款下的開放源碼發佈，是第二流行的企業搜索引擎。設計用於雲計算中，可以達到實時搜索，穩定，可靠，快速，安裝使用方便。在Elasticsearch中，全部節點的數據是均等的。ios

（1）關於集羣配置nginx

discovery.zen.ping.unicast.hosts，Elasticsearch默認使用Zen Discovery來作節點發現機制，推薦使用unicast來作通訊方式，在該配置項中列舉出Master節點。
discovery.zen.minimum_master_nodes，該參數表示集羣中可工做的具備Master節點資格的最小數量，默認值是1。爲了提升集羣的可用性，官方推薦設置爲(N/2)+1，其中N是具備Master資格的節點的數量。
discovery.zen.ping_timeout，表示節點在發現過程當中的等待時間，默認值是3秒，能夠根據自身網絡環境進行調整，必定程度上提供可用性。
```
discovery.zen.ping.unicast.hosts: ["host1", "host2"]
discovery.zen.minimum_master_nodes: 2
discovery.zen.ping_timeout: 10
```

（2）關於集羣節點git

節點類型包括：候選Master節點、數據節點和Client節點。經過設置兩個配置項node.master和node.data爲true或false，來決定將一個節點分配爲何類型的節點。
儘可能將候選Master節點和Data節點分離開，一般Data節點負載較重，須要考慮單獨部署。

（3）關於內存
Elasticsearch默認設置的內存是1GB，對於任何一個業務部署來講，這個都過小了。經過指定ES_HEAP_SIZE環境變量，能夠修改其堆內存大小，服務進程在啓動時候會讀取這個變量，並相應的設置堆的大小。建議設置系統內存的一半給Elasticsearch，可是不要超過32GB。github

（4）關於硬盤空間
Elasticsearch將數據存儲在自定義的路徑下，隨着數據的增加，必定會出現硬盤空間不夠用的情形，此時就須要給機器掛載新的硬盤，並將Elasticsearch的路徑配置到新硬盤的路徑下。經過「path.data」配置項來進行設置，好比「path.data: /elk/elasticsearch/data」。須要注意的是，同一分片下的數據只能寫入到一個路徑下，所以仍是須要合理的規劃。

（5）關於Index的劃分和分片的個數
這個須要根據數據量來作權衡了，Index能夠按時間劃分，好比每個月一個或者天天一個，在Logstash輸出時進行配置，shard的數量也須要作好控制。

（6）關於監控
這裏我使用head和kopf兩個監控插件。

二、Logstash

Logstash是一個開源的日誌採集工具，是一個數據管道。主要用來採集、解析和分析大量結構化和非結構化的數據以及各類系統產生的事件。它能夠對你的日誌進行採集、過濾、分析，並將其存儲供之後使用（如，搜索），咱們可使用它。說到搜索，logstash帶有一個web界面，搜索和展現全部日誌。

Logstash工做原理展現圖：

（1）經常使用的輸入input

file：文件輸入插件用於將事件和日誌文件輸入給logstash。它會自動檢測日誌文件輪轉，並從上次讀取的位點讀取數據。
syslog：在514端口上監聽系統日誌消息，並根據RFC3164標準進行解析
redis：從redis service中讀取
beats：從filebeat中讀取
Filters：數據中間處理，對數據進行操做。

（2）經常使用的過濾器filter

grok：解析任意文本數據，Grok 是 Logstash 最重要的插件。它的主要做用就是將文本格式的字符串，轉換成爲具體的結構化的數據，配合正則表達式使用。內置120多個解析語法。
mutate：對字段進行轉換。例如對字段進行刪除、替換、修改、重命名等。
drop：丟棄一部分events不進行處理。
clone：拷貝 event，這個過程當中也能夠添加或移除字段。
geoip：添加地理信息（爲前臺kibana圖形化展現使用）。
Outputs：outputs是logstash處理管道的最末端組件。一個event能夠在處理過程當中通過多重輸出，可是一旦全部的outputs都執行結束，這個event也就完成生命週期。

（3）常見的輸出output

elasticsearch：elasticsearch插件是ELK技術棧中用到的最重要的插件，它用於將輸出的結果存儲到elasticsearch搜索引擎中供Kibana分析。
file：將event數據保存到文件中。
graphite：將event數據發送到圖形化組件中，一個很流行的開源存儲圖形化展現的組件。
codecs：codecs 是基於數據流的過濾器，它能夠做爲input，output的一部分配置。Codecs能夠幫助你輕鬆的分割發送過來已經被序列化的數據。

（4）常見的編解碼codecs

json：使用json格式對數據進行編碼/解碼。
line：用於將每行輸入日誌做爲一個事件，將每一個事件解碼成一行。
multiline：將匯多個事件中數據彙總爲一個單一的行。好比：java異常信息和堆棧信息。
plain：插件用於指示輸入或輸出事件時，不須要額外的編解碼操做，對應的輸入輸出插件會本身處理編解碼。對不少插件如redis、mongondb等而言，plain是默認的編解碼類型。
rubydebug：只在輸出事件時使用，它使用Ruby Awesome打印庫打印輸出事件。

（5）採集日誌信息

pattern => "^\["             # 採集以"["開頭的日誌信息
pattern => "^2018"           # 採集以"2018"開頭的日誌信息
pattern => "^[a-zA-Z0-9]"      # 採集以字母（大小寫）或數字開頭的日誌信息
pattern => "^[a-zA-Z0-9]|[^ ]+"  # 採集以字母（大小寫）或數字或空格的日誌信息

三、Kibana

Kibana是一個開源的基於瀏覽器頁面的Elasticsearch前端展現工具。它Kibana能夠爲 Logstash 和 ElasticSearch 提供的日誌分析友好的 Web 界面，能夠幫助咱們彙總、分析和搜索重要數據日誌，也能夠經過直方圖、地圖、餅圖、其它圖形和表格等方式來表達數據。只須要點擊鼠標，就能夠完成搜索、聚合功能，生成炫麗的儀表板。
Kibana提供的是數據查詢和顯示的Web服務，有豐富的圖表樣板，能知足大部分的數據可視化需求，這也是不少人選擇ELK的主要緣由之一。

而在5.0版本之後，Elastic公司將原來的 ELK Stack 稱之爲 Elastic Stack，緣由是引入了 Beats 套件。

官網：https://www.elastic.co/products

ELK工做原理展現圖：

如上圖所示：多個應用服務器上的日誌經過Logstash採集器傳輸到一個集中化的索引器中，索引器將處理後的數據結果輸出到ElasticSearch集羣，而後Kibana經過下查詢ElasticSearch集羣中的日誌數據建立儀表盤，作可視化展示。

2、ELK總體方案

ELK中的三個系統分別扮演着不一樣的角色，組成了一個總體的解決方案。Logstash是一個ETL工具，負責從每臺機器抓取日誌數據，對數據進行格式轉換和處理後，輸出到Elasticsearch中存儲。Elasticsearch是一個分佈式搜索引擎和分析引擎，用於數據存儲，可提供實時的數據查詢。Kibana是一個數據可視化服務，根據用戶的操做從Elasticsearch中查詢數據，造成相應的分析結果，以圖表的形式展示給用戶。
ELK的安裝很簡單，能夠按照「下載->修改配置文件->啓動」方法分別部署三個系統，也可使用Docker來快速部署。下面來看一個常見的部署方案，以下圖所示：

在每臺生成日誌文件的機器上，部署Logstash，做爲Shipper的角色，負責從日誌文件中提取數據，可是不作任何處理，直接將數據輸出到Redis隊列(list)中；
須要一臺機器部署Logstash，做爲Indexer的角色，負責從Redis中取出數據，對數據進行格式化和相關處理後，輸出到Elasticsearch中存儲；
部署Elasticsearch集羣，固然取決於你的數據量了，數據量小的話可使用單臺服務，若是作集羣的話，最好是有3個以上節點，同時還須要部署相關的監控插件；
部署Kibana服務，提供Web可視化展示服務。

在前期部署階段，主要工做是Logstash節點和Elasticsearch集羣的部署，而在後期使用階段，主要工做就是Elasticsearch集羣的監控和使用Kibana來檢索、分析日誌數據了，固然也能夠直接編寫程序來消費Elasticsearch中的數據。

在上面的部署方案中，咱們將Logstash分爲Shipper和Indexer兩種角色來完成不一樣的工做，中間經過Redis作數據管道，爲何要這樣作？爲何不是直接在每臺機器上使用Logstash提取數據、處理、存入Elasticsearch？

首先，採用這樣的架構部署，有三點優點：

下降對日誌所在機器的影響，這些機器上通常都部署着反向代理或應用服務，自己負載就很重了，因此儘量的在這些機器上少作事。
若是有不少臺機器須要作日誌採集，那麼讓每臺機器都向Elasticsearch持續寫入數據，必然會對Elasticsearch形成壓力，所以須要對數據進行緩衝，同時，這樣的緩衝也能夠必定程度的保護數據不丟失。
將日誌數據的格式化與處理放到Indexer中統一作，能夠在一處修改代碼、部署，避免須要到多臺機器上去修改配置。

其次，咱們須要作的是將數據放入一個消息隊列中進行緩衝，因此Redis只是其中一個選擇，也能夠是RabbitMQ、Kafka等等，在實際生產中，Redis與Kafka用的比較多。因爲Redis集羣通常都是經過key來作分片，沒法對list類型作集羣，在數據量大的時候必然不合適了，而Kafka天生就是分佈式的消息隊列系統。

3、ELK環境部署

一、基礎環境介紹

# 系統： Centos7.3
[root@elk-01 ~]# cat /etc/redhat-release 
CentOS Linux release 7.3.1611 (Core)

# 防火牆： 關閉
[root@elk-01 ~]# systemctl stop firewalld && systemctl disable firewalld
Removed symlink /etc/systemd/system/multi-user.target.wants/firewalld.service.
Removed symlink /etc/systemd/system/dbus-org.fedoraproject.FirewallD1.service.

# Sellinux： 關閉
[root@elk-01 ~]# setenforce 0
[root@elk-01 ~]# sed -i 's/SELINUX=enforcing/SELINUX=disabled/g' /etc/selinux/config

# 機器環境： 兩臺
elk-01: 192.168.8.55       # master機器
elk-02: 192.168.8.66       # slave機器

說明：
master-slave模式：master採集到日誌後，會把一部分數據碎片到salve上（隨機的一部分數據）；同時，master和slave又都會各自作副本，並把副本放到對方機器上，這樣就保證了數據不會丟失。
若是master宕機了，那麼客戶端在日誌採集配置中將Elasticsearch主機指向改成slave，就能夠保證ELK日誌的正常採集和web展現

二、Elasticsearch安裝配置

Elasticsearch是java程序，並且要跑在jdk1.8版本以上
elk-01和elk-02同時操做

（1）寫hosts文件

[root@elk-01 ~]# echo "192.168.8.55 elk-01" >> /etc/hosts
[root@elk-02 ~]# echo "192.168.8.66 elk-02" >> /etc/hosts

（2）安裝jdk1.8版本

[root@elk-01 ~]# yum -y install java-1.8.0
[root@elk-01 ~]# java -version
openjdk version "1.8.0_102"
OpenJDK Runtime Environment (build 1.8.0_102-b14)
OpenJDK 64-Bit Server VM (build 25.102-b14, mixed mode)

（3）安裝Elasticsearch

[root@elk-01 ~]# wget https://download.elastic.co/elasticsearch/release/org/elasticsearch/distribution/tar/elasticsearch/2.4.6/elasticsearch-2.4.6.tar.gz
[root@elk-01 ~]# mkdir /elk
[root@elk-01 ~]# tar zxvf elasticsearch-2.4.6.tar.gz -C /elk
[root@elk-01 ~]# cd /elk/
[root@elk-01 elk]# ls
elasticsearch-2.4.6
[root@elk-01 elk]# mv ./elasticsearch-2.4.6 ./elasticsearch

（4）修改Elasticsearch配置文件

[root@elk-01 ~]# vim /elk/elasticsearch/config/elasticsearch.yml     #將配置文件內容修改以下，打開註釋
17 cluster.name: test-ELK     # 配置集羣名，兩臺服務器名保持一致
23 node.name: elk-01   # 配置單一節點名稱，每一個節點惟一標識
33 path.data: /elk/elasticsearch/data   # data存儲路徑
37 path.logs: /elk/elasticsearch/logs   #log存儲路徑
43 bootstrap.memory_lock: true      # 鎖住內存，不被使用到交換分區去
54 network.host: 0.0.0.0    # 監聽地址
58 http.port: 9200          # 監聽端口
68 discovery.zen.ping.unicast.hosts: ["192.168.8.66"]   # 集羣節點發現列表，寫另一臺的ip
72 discovery.zen.minimum_master_nodes: 2   # 集羣可作master的最小節點數

[root@elk-01 ~]# mkdir -p /elk/elasticsearch/{data,logs}

（5）修改系統參數

[root@elk-01 ~]# vim /etc/security/limits.conf     --文件最後添加系統參數，讓系統用戶打開文件無限制
# 添加配置
*        soft    nofile  65536
*        hard    nofile  131072
*        soft    nproc   2048
*        hard    nproc   4096
*        soft    memlock unlimited
*        hard    memlock unlimited

[root@elk-01 ~]# vim /etc/sysctl.conf   --在此文件里加入如下參數
# 添加配置
vm.max_map_count= 262144

[root@elk-01 ~]# sysctl -p  --刷新
vm.max_map_count = 262144

（6）開戶啓動Elasticsearch服務

當使用root帳戶啓動Elasticsearch會出現錯誤信息，這是由於處於系統安裝考慮的設置，因爲Elasticsearch能夠接收用戶輸入的腳本而且執行，爲了系統安全考慮，不容許root帳號啓動，因此建議給Elasticsearch單首創建一個用戶來運行Elasticsearch。

# 建立elk用戶組及elk用戶
[root@elk-01 ~]# groupadd elk
[root@elk-01 ~]# useradd elk -g elk -p 123
[root@elk-01 ~]# chown -R elk.elk /elk/elasticsearch
[root@elk-01 ~]# su - elk
上一次登陸：六 5月  5 11:08:30 CST 2018pts/5 上
[elk@elk-01 ~]$ cd /elk/elasticsearch/
[elk@elk-01 elasticsearch]$ nohup ./bin/elasticsearch &
[1] 3503
[elk@elk-01 elasticsearch]$ nohup: 忽略輸入並把輸出追加到"nohup.out"

[root@elk-01 ~]# netstat -antlp |egrep "9200|9300"
tcp6       0      0 :::9200                 :::*                    LISTEN      1608/java
tcp6       0      0 :::9300                 :::*                    LISTEN      1608/java
tcp6       0      1 192.168.8.66:47992      192.168.8.55:9300       SYN_SENT    1608/java
tcp6       0      0 192.168.8.66:9200       192.168.8.1:2015        ESTABLISHED 1608/java

若是啓動失敗，在系統日誌（咱們的示例中爲/ var/log/messages）中將會遇到與如下相似的錯誤：

Apr  9 15:39:09 elk-02 kernel: [ 3727]  1000  3727   784748   214244     617    70749             0 java
Apr  9 15:39:09 elk-02 kernel: Out of memory: Kill process 3727 (java) score 368 or sacrifice child
Apr  9 15:39:09 elk-02 kernel: Killed process 3727 (java) total-vm:3138992kB, anon-rss:849016kB, file-rss:7960kB, shmem-rss:0kB

這個錯誤的緣由是low memory耗盡。「內核使用low memory來跟蹤全部的內存分配，一旦low memory耗盡，就會查殺進程，以保持系統的正常運轉。說白了 OOM Killer 就是一層保護機制，用於避免 Linux 在內存不足的時候不至於出太嚴重的問題，把可有可無的進程殺掉。
最簡單的解決方法就是增長內存空間。

（7）訪問Elasticsearch

http://192.168.8.55:9200/
http://192.168.8.66:9200/

瀏覽器訪問結果（訪問時建議使用chrome、firefox瀏覽器）

三、安裝Elasticsearch-head插件

elasticsearch-head是用於瀏覽Elasticsearch集羣並與其進行交互的Web前端。也就是用於顯示集羣節點和數據信息。

Elasticsearch 5及以上版本（可自行研究）中的head插件已經不經過elasticsearch-plugin 來進行安裝，已經成爲了一個獨立的服務，須要單獨進行安裝，安裝方法可參考GitHub網站。

在Elasticsearch 5以前的版本，安裝起來比較簡單，在安裝成功後能夠在瀏覽器中經過http://ip:9200/_plugin/head 就能夠訪問了，這裏的IP地址即爲你的Elasticsearch服務器地址。查看官方文檔。具體安裝以下：

[root@elk-01 ~]# /elk/elasticsearch/bin/plugin install mobz/elasticsearch-head
-> Installing mobz/elasticsearch-head...
Trying https://github.com/mobz/elasticsearch-head/archive/master.zip ...
Downloading .............................................................................................................................DONE
Verifying https://github.com/mobz/elasticsearch-head/archive/master.zip checksums if available ...
NOTE: Unable to verify checksum for downloaded plugin (unable to find .sha1 or .md5 file to verify)
Installed head into /elk/elasticsearch/plugins/head

插件安裝的目錄：/elk/elasticsearch/plugins
在線安裝完成以後，進行授予權限

[root@elk-01 ~]# chown -R elk:elk /elk/elasticsearch/plugins
[root@elk-01 ~]# ll /elk/elasticsearch/plugins/head/

插件訪問（最好提早將elk-02節點的配置和插件都安裝後，再來進行訪問和數據插入測試）

http://192.168.8.55:9200/_plugin/head/
http://192.168.8.66:9200/_plugin/head/

注：ES集羣健康度分爲：紅、黃、綠三種顏色
紅色：個別分片副本不可用
×××：個別副本不可用
綠色：爲健康

兩臺機器看到的內容徹底一致，test-ELK是集羣名稱，集羣健康值爲綠色。

測試一
插入數據實例，看界面能不能正常顯示
以下：點擊「複合查詢」，在POST選項下，任意輸入如/2018/test而後在下面輸入數據（注意內容之間換行的逗號不要漏掉）；
數據輸入好以後(以下輸入"user":"xiaozuo","mess":"test Elasticsearch"內容)，下面點擊「驗證JSON」->「提交請求」，提交成功後，觀察右欄內出現的信息：有index、type、version等信息，failed:0（成功消息）

測試二
以下：點擊「複合查詢」，選擇GET選項，在/2018/test/後面輸入上面POST結果中的id號，不輸入內容，即{}括號裏爲空！
而後點擊「驗證JSON」->「提交請求」，觀察右欄內就有了上面插入的數據了（即xiaozuo，test Elasticsearch）

點擊「基本查詢」，查看數據，以下圖所示，便可查詢到上面插入的數據：

點擊「數據瀏覽」，也能查看到插入的數據：

點擊「概覽」，主界面也會更新，顯示剛纔建立類型，也會顯示數據碎片存取的位置：

每一個索引都有5個分片，粗線寬的分片是主節點分片，細線寬的是副本。這樣kibana在搜索的時候就能夠從多個Easticsearch服務器上讀取，壓力也按比例分佈在各個集羣節點中。

說明：
必定要提早在elk-02節點上也完成配置（配置內容同上）。若是你尚未創建集羣，那麼只能看到一個節點，並且上面插入數據後，Elasticsearch集羣狀態會呈現×××yellow狀態，elk-02完成配置加入到集羣裏後就會恢復到正常的綠色狀態。
當數據很少的時候經過一臺Elasticsearch服務器也能完成任務。可是數據隨着時間的推移而增多，尤爲須要查詢6個月、一年甚至更長時間跨度的數據時，你會發現集羣有多麼重要。

四、安裝kopf監控插件

Kopf展現了ElasticSearch的節點分片、文檔、所佔用空間、大小等信息，而且能夠查詢文檔的索引。

[root@elk-01 ~]# /elk/elasticsearch/bin/plugin install lmenezes/elasticsearch-kopf
-> Installing lmenezes/elasticsearch-kopf...
Trying https://github.com/lmenezes/elasticsearch-kopf/archive/master.zip ...
Downloading .............................................................................................................................DONE
Verifying https://github.com/lmenezes/elasticsearch-kopf/archive/master.zip checksums if available ...
NOTE: Unable to verify checksum for downloaded plugin (unable to find .sha1 or .md5 file to verify)
Installed kopf into /elk/elasticsearch/plugins/kopf

# 授予權限
[root@elk-01 ~]# chown -R elk:elk /elk/elasticsearch/plugins
[root@elk-01 ~]# ll /elk/elasticsearch/plugins/kopf/

訪問插件：（以下，一樣要提早安裝好elk-02節點上的插件，不然訪問時會出現集羣節點爲×××的yellow告警狀態）

http://192.168.8.55:9200/_plugin/kopf/#!/cluster
http://192.168.8.66:9200/_plugin/kopf/#!/cluster

到此Elasticsearch軟件包安裝完成。

五、Logstash安裝配置

elk-01和elk-02都要安裝。同時，Logstash部署在每臺生成日誌文件的機器上，採集到的數據寫入到Elasticsearch裏，就能夠登錄Logstash界面查看到了）

（1）安裝Logstash

[root@elk-01 ~]# wget https://download.elastic.co/logstash/logstash/logstash-2.4.1.tar.gz
[root@elk-01 ~]# tar zxvf logstash-2.4.1.tar.gz -C /elk/
[root@elk-01 ~]# mv /elk/logstash-2.4.1/ /elk/logstash/
[root@elk-01 ~]# useradd logstash -s /sbin/nologin 
[root@elk-01 ~]# chown -R logstash.logstash /elk/logstash/

（2）數據的測試

測試一
基本的輸入輸出，驗證logstash可否接收到數據。

[root@elk-01 ~]# /elk/logstash/bin/logstash -e "input { stdin{ } } output { stdout{} }"
Settings: Default pipeline workers: 1
Pipeline main started
xiaozuo      # 輸入的內容
2018-05-06T07:15:50.140Z elk-01.com xiaozuo   # 輸出的內容
hello        # 輸入的內容
2018-05-06T07:16:09.850Z elk-01.com hello     # 輸出的內容

在這個例子中，咱們經過輸入插件stdin和輸出插件stdout來運行Logstash，因此無論你輸入什麼，都會輸出一樣的信息。使用-e參數能夠在命令行中快速地測試配置是否正確。

測試二
使用rubydebug詳細輸出

[root@elk-01 ~]# /elk/logstash/bin/logstash -e "input { stdin{} } output { stdout{ codec => rubydebug} }"
Settings: Default pipeline workers: 1
Pipeline main started
hello    # 輸入的內容
{       # 輸出下面信息
       "message" => "hello",
      "@version" => "1",
    "@timestamp" => "2018-05-06T07:23:16.933Z",
          "host" => "elk-01.com"
}
xiaozuoxiansen   # 輸入的內容
{       # 輸出下面信息
       "message" => "xiaozuoxiansen",
      "@version" => "1",
    "@timestamp" => "2018-05-06T07:23:48.631Z",
          "host" => "elk-01.com"
}

測試三
把內容寫到Elasticsearch中

[root@elk-01 ~]# /elk/logstash/bin/logstash -e "input { stdin{} } output { elasticsearch { hosts => ['192.168.8.55:9200']} }"
Settings: Default pipeline workers: 1
Pipeline main started
123456     # 隨意輸入內容
xiaozuo
hello

說明：
使用rubydebug和寫到elasticsearch中的區別：其實就在於後面標準輸出的區別，前者使用codec；後者使用elasticsearch。

寫到Elasticsearch中在Logstash中查看，以下圖所示：

注意：
master採集到日誌後，會把一部分數據碎片到salve上（隨機的一部分數據），master和slave又都會各自作副本，並把副本放到對方機器上，這樣就保證了數據不會丟失。
以下，master採集到的數據放到了本身的第0、二、4分片上，其餘的放到了slave的第一、3分片上。

再點擊「數據瀏覽」，能夠看到剛纔輸入的內容：

（3） Logstash的配置和文件的編寫

1）Logstash的配置
簡單的配置方式：

[root@elk-01 ~]# mkdir /elk/logstash/conf.d/
[root@elk-01 ~]# vim /elk/logstash/conf.d/01-logstash.conf
input { stdin { } }
output {
        elasticsearch { hosts => ["192.168.8.55:9200"]}
        stdout { codec => rubydebug }
}

執行結果：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /elk/logstash/conf.d/01-logstash.conf 
Settings: Default pipeline workers: 1
Pipeline main started
ShenZhen       # 隨意輸入內容
{       # 輸出下面的信息
       "message" => "ShenZhen",
      "@version" => "1",
    "@timestamp" => "2018-05-06T07:59:40.650Z",
          "host" => "elk-01.com"
}

再擊「數據瀏覽」，能夠看到剛纔輸入的內容：

參考官方文檔：
https://www.elastic.co/guide/en/logstash/current/configuration.html
https://www.elastic.co/guide/en/logstash/current/configuration-file-structure.html

2）採集系統日誌

[root@elk-01 ~]# vim /elk/logstash/conf.d/file.conf
input {            # 指定輸出
    file {         # 輸出的內容爲文件
      path => "/var/log/messages"   # 文件路徑
      type => "system"            # 給該類日誌內容定一個名稱，可自定義
      start_position => "beginning"  # 表示從哪裏開始讀取日誌,beginning是全部都讀
    }
}

output {               # 指定輸出到哪裏
    elasticsearch {      # 指定輸出到elasticaearch服務裏
       hosts => ["192.168.8.55:9200"]     # 指定鏈接elasticaearch服務的IP和端口
       index => "system-%{+YYYY.MM.dd}"   # 把日誌按日期進行分類
    }
}

執行上面日誌信息的採集，以下圖所示，這個命令會一直在執行中，表示日誌在監控採集中；若是中斷，就表示日誌不在採集！因此須要放在後臺執行&

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /elk/logstash/conf.d/file.conf &

登錄Elasticsearch界面，查看本機系統日誌的信息：

參考官方文檔：
https://www.elastic.co/guide/en/logstash/current/plugins-outputs-elasticsearch.html
https://www.elastic.co/guide/en/logstash/current/event-dependent-configuration.html

從上面採集系統日誌的「數據瀏覽」中能夠看出，每一個日誌都給採集成一行了，不是按照一個日誌，一個事件模塊採集的。

（4）將行換成事件的方式展現

[root@elk-01 ~]# vim /elk/logstash/conf.d/multiline.conf
input {
    stdin {
       codec => multiline {
          pattern => "^\["
          negate => true
          what => "previous"
        }
    }
}
output {
    stdout {
      codec => "rubydebug"
     }
}

執行命令：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /elk/logstash/conf.d/multiline.conf

說明：
在沒有遇「[」的時候，系統不會採集，只有碰見「[」的時候，纔算是一個事件，才採集起來。

參考官方文檔：
https://www.elastic.co/guide/en/logstash/current/plugins-codecs-multiline.html

六、Kibana安裝配置

（1）安裝Kibana

[root@elk-01 ~]# wget https://download.elastic.co/kibana/kibana/kibana-4.6.6-linux-x86_64.tar.gz
[root@elk-01 ~]# tar zxvf kibana-4.6.6-linux-x86_64.tar.gz -C /elk/
[root@elk-01 ~]# mv /elk/kibana-4.6.6-linux-x86_64/ /elk/kibana/

（2）修改配置文件

[root@elk-01 ~]# cd /elk/kibana/config/
[root@elk-01 config]# pwd
/elk/kibana/config
[root@elk-01 config]# ls
kibana.yml
[root@elk-01 config]# mv kibana.yml kibana.yml.bak
[root@elk-01 config]# vim kibana.yml
server.port: 5601
server.host: "0.0.0.0"
elasticsearch.url: "http://192.168.8.55:9200"
kibana.index: ".kibana"  # 注意這個.Kibana的index索引配置必定要有，它是將ES數據經過kibana進行web展現的關鍵。這個配置後，在ES的web界面裏就會看到這個.kibana索引。

（3）啓動Kibana

由於Kibana一直運行在前臺，因此能夠放在後臺執行&

[root@elk-01 ~]# cd /elk/kibana/
[root@elk-01 kibana]# nohup ./bin/kibana &
[1] 10314
[root@elk-01 kibana]# nohup: 忽略輸入並把輸出追加到"nohup.out"
[root@elk-01 kibana]# lsof -i:5601
COMMAND   PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
node    10314 root   11u  IPv4 107473      0t0  TCP *:esmagent (LISTEN)

（4）訪問Kibana

http://192.168.8.55:5601

（5）建立日誌採集項

建立一個日誌採集項，該名稱要和前面寫的file.conf配置文件裏的type字段一致。好比添加system系統日誌。注意後面的*不要忘了。

建立完成後，再點擊「Discover」，在Discover中查看：

查看日誌登錄，須要點擊「Discover」-->「message」，點擊它後面的「add」
說明：
須要右邊查看日誌內容時帶什麼屬性，就在左邊點擊相應屬性後面的「add」
以下圖，添加了message和path的屬性：

這樣，右邊顯示的日誌內容的屬性就帶了message和path。

點擊右邊日誌內容屬性後面隱藏的<<，就可將內容向前縮進：

添加新的日誌採集項，點擊「Settings」->「 +Add New」，好比添加logstash-日誌。注意後面的*不要忘了。

刪除Kibana裏的日誌採集項，點擊刪除圖標便可，以下圖所示：

若是打開Kibana查看日誌，發現沒有日誌內容，出現「No results found」，以下圖所示，這說明要查看的日誌在當前時間沒有日誌信息輸出，能夠點擊右上角的時間鍾來調試日誌信息的查看。

（6）採集nginx的訪問日誌

[root@elk-01 ~]# wget https://nginx.org/download/nginx-1.14.0.tar.gz
[root@elk-01 ~]# tar zxf nginx-1.14.0.tar.gz -C /usr/src/
[root@elk-01 ~]# cd /usr/src/nginx-1.14.0/
[root@elk-01 ~]# ./configure --prefix=/usr/local/nginx
[root@elk-01 ~]# make && make install
[root@elk-01 ~]# useradd nginx -s /sbin/nologin  --建立nginx用戶，-s制定shell，nginx用戶不須要登陸系統
[root@elk-01 ~]# id nginx
uid=1002(nginx) gid=1002(nginx) 組=1002(nginx)
[root@elk-01 ~]# chown -R nginx.nginx /usr/local/nginx/
[root@elk-01 ~]# vim /usr/local/nginx/conf/nginx.conf  --打開如下注釋（根據需求）
  2 user nginx nginx;  # 運行的用戶和組，打開註釋，使用nginx身份
  7 error_log  logs/error.log  info;  # 錯誤日誌以及日誌等級，打開等級爲info級別的日誌
  9 pid        logs/nginx.pid;  # pid文件
33     gzip  on;  # 在服務器壓縮數據，進行傳輸
37         server_name  192.168.8.55;  # 主機名或者IP
39         charset utf8;  # 字符集,改爲utf8
48         error_page  404              /404.html;  # 錯誤頁面

Nginx安裝配置完成後，先查看一下80端口是否被佔用，佔用的話先停掉Apache（因Apache默認是80端口，若是是其它程序佔用，則停掉佔用80端口的程序）

[root@elk-01 ~]# lsof -i:80   --80端口沒有佔用
[root@elk-01 ~]# /usr/local/nginx/sbin/nginx 
[root@elk-01 ~]# lsof -i:80   --Nginx啓動成功
COMMAND   PID   USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
nginx   13375   root    6u  IPv4 166972      0t0  TCP *:http (LISTEN)
nginx   13376 nobody    6u  IPv4 166972      0t0  TCP *:http (LISTEN)

瀏覽器訪問Nginx

http://192.168.8.55/

Nginx安裝成功以後，修改Nginx的配置文件，分別在nginx.conf的http和server配置區域添加下面內容：

# http 標籤中
    log_format json '{"@timestamp":"$time_iso8601",'
                      '"@version":"1",'
                      '"client":"$remote_addr",'
                      '"url":"$uri",'
                      '"status":"$status",'
                      '"domain":"$host",'
                      '"host":"$server_addr",'
                      '"size":$body_bytes_sent,'
                      '"responsetime":$request_time,'
                      '"referer": "$http_referer",'
                      '"ua": "$http_user_agent"'
                      '}';

# server標籤中
        access_log /var/log/nginx/access_json.log json;

截圖以下：

啓動Nginx服務：

[root@elk-01 ~]# mkdir /var/log/nginx/
[root@elk-01 ~]# touch /var/log/nginx/access_json.log
[root@elk-01 ~]# /usr/local/nginx/sbin/nginx -s stop
[root@elk-01 ~]# /usr/local/nginx/sbin/nginx

編寫日誌採集文件：
使用的是json的方式採集：

[root@elk-01 ~]# vim json.conf
input {
   file {
      path => "/var/log/nginx/access_json.log"
      codec => "json"
   }
}

output {
   stdout {
      codec => "rubydebug"
   }
}

啓動日誌採集程序：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/json.conf &   --加個&放在後臺執行

訪問nginx頁面（在elk-01的主機上執行訪問頁面的命令：curl http://192.168.8.55）就會出現如下圖所示：

注意：
上面的json.conf配置只是將nginx日誌輸出，尚未輸入到Elasticsearch裏，因此這個時候在Elasticsearch界面裏是採集不到nginx日誌的。
須要配置一下，將Nginx日誌輸入到Elasticsearch中，將其彙總到總文件file.conf裏，以下也將nginx-log日誌輸入到Elasticserach裏：（後續就能夠只用這個彙總文件，把要追加的日誌彙總到這個總文件裏便可）

[root@elk-01 ~]# cat file.conf 
input {
    file {
      path => "/var/log/messages"
      type => "system"
      start_position => "beginning"
    }

    file {
       path => "/var/log/nginx/access_json.log"
       codec => json
       start_position => "beginning"
       type => "nginx-log"
    }
}

output {

    if [type] == "system"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "system-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "nginx-log"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "nignx-log-%{+YYYY.MM.dd}"
        }
    }
}

在執行日誌採集命令時，能夠加上--configtest參數，測試下配置文件是否有語法錯誤或配置不當的地方，這個很重要！！

[root@elk-01 ~]# /elk/logstash/bin/logstash -f file.conf --configtest
Configuration OK

而後接着執行Logstash命令（因爲上面已經將這個執行命令放到了後臺，因此這裏其實不用執行，也能夠先kill以前的，再放後臺執行），而後能夠再訪問Nginx界面測試下

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/file.conf &

登錄Elasticsearch主界面查看：

登錄Kibana主界面查看：

（7）採集系統日誌

編寫日誌採集文件：

[root@elk-01 ~]# cat syslog.conf 
input {
    syslog {
        type => "system-syslog"
        host => "192.168.8.55"
        port => "514"
    }
}

output {
    stdout {
        codec => "rubydebug"
    }
}

對上面的採集文件進行執行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f syslog.conf

從新開啓一個窗口，查看服務是否啓動：

[root@elk-01 ~]# netstat -antlp | grep 514
tcp6       0      0 192.168.8.55:514        :::*                    LISTEN      14101/java          
[root@elk-01 ~]# vim /etc/rsyslog.conf
[root@elk-01 ~]# systemctl restart rsyslog

回到原來的窗口（即上面採集文件的執行終端），就會出現數據：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f syslog.conf
Settings: Default pipeline workers: 1
Pipeline main started
{
           "message" => "[origin software=\"rsyslogd\" swVersion=\"7.4.7\" x-pid=\"14155\" x-info=\"http://www.rsyslog.com\"] start\n",
          "@version" => "1",
        "@timestamp" => "2018-05-07T08:30:05.000Z",
              "type" => "system-syslog",
              "host" => "192.168.8.55",
          "priority" => 46,
         "timestamp" => "May  7 16:30:05",
         "logsource" => "elk-01",
           "program" => "rsyslogd",
          "severity" => 6,
          "facility" => 5,
    "facility_label" => "syslogd",
    "severity_label" => "Informational"
}
........
........

再次把syslog.conf文件的內容添加到總文件file.conf中：

[root@elk-01 ~]# cat file.conf 
input {
    file {
      path => "/var/log/messages"
      type => "system"
      start_position => "beginning"
    }

    file {
       path => "/var/log/nginx/access_json.log"
       codec => json
       start_position => "beginning"
       type => "nginx-log"
    }
    syslog {
        type => "system-syslog"
        host => "192.168.8.55"
        port => "514"
    }
}

output {
    if [type] == "system"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "system-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "nginx-log"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "nignx-log-%{+YYYY.MM.dd}"
        }
    }
        ``
    if [type] == "system-syslog"{
        elasticsearch {
           hosts => ["192.168.8.55:9200"]
           index => "system-syslog-%{+YYYY.MM.dd}"
        }
    }
}

執行總文件（先測試下總文件file.conf配置是否有誤，而後先kill以前在後臺啓動的file.conf文件，再次執行）：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/file.conf --configtest
Configuration OK
[root@elk-01 ~]# jobs -l
[1]  13511 運行中               nohup ./bin/kibana &(工做目錄：/elk/kibana)
[2]- 13645 運行中               /elk/logstash/bin/logstash -f /root/json.conf &
[3]+ 13815 運行中               /elk/logstash/bin/logstash -f /root/file.conf &
[root@elk-01 ~]# kill -9 13815
[root@elk-01 ~]# 
[3]+  已殺死               /elk/logstash/bin/logstash -f /root/file.conf
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/file.conf &

測試：
向日志中添加數據，查看Elasticsearch和Kibana的變化：

[root@elk-01 ~]# logger "Hello World_01"
[root@elk-01 ~]# logger "Hello World_02"
[root@elk-01 ~]# logger "Hello World_03"
[root@elk-01 ~]# logger "Hello World_04"
[root@elk-01 ~]# logger "Hello World_05"

（8）TCP日誌的採集

編寫日誌採集文件，並執行：（有須要的話，能夠將下面採集文件的配置彙總到上面的總文件file.conf裏，進而輸入到Elasticsearch界面裏和Kibana裏查看）

[root@elk-01 ~]# cat tcp.conf 
input {
    tcp {
        host => "192.168.8.55"
        port => "6666"
    }
}

output {
    stdout {
        codec => "rubydebug"
    }
}
# 執行
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/tcp.conf

從新開啓一個窗口，執行以下命令：
測試一（安裝nc命令）：

[root@elk-01 ~]# yum install -y nc
[root@elk-01 ~]# nc 192.168.8.55 6666 < /etc/resolv.conf

回到原來的窗口（即上面採集文件的執行終端），就會出現數據：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/tcp.conf 
Settings: Default pipeline workers: 1
Pipeline main started
{                  
       "message" => "# Generated by NetworkManager",
      "@version" => "1",
    "@timestamp" => "2018-05-07T08:59:11.599Z",
          "host" => "192.168.8.55",
          "port" => 58622
}
{
       "message" => "search elk-01.com",
      "@version" => "1",
    "@timestamp" => "2018-05-07T08:59:11.601Z",
          "host" => "192.168.8.55",
          "port" => 58622
}
{
       "message" => "nameserver 192.168.8.1",
      "@version" => "1",
    "@timestamp" => "2018-05-07T08:59:11.614Z",
          "host" => "192.168.8.55",
          "port" => 58622
}

測試二

[root@elk-01 ~]# echo "hello" | nc 192.168.8.55 6666
[root@elk-01 ~]# echo "hello" > /dev/tcp/192.168.8.55/6666

回到以前的執行端口，再去查看，就會顯示出數據：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/tcp.conf 
Settings: Default pipeline workers: 1
Pipeline main started
{
       "message" => "hello",
      "@version" => "1",
    "@timestamp" => "2018-05-07T09:00:51.130Z",
          "host" => "192.168.8.55",
          "port" => 58970
}
{
       "message" => "hello",
      "@version" => "1",
    "@timestamp" => "2018-05-07T09:01:13.536Z",
          "host" => "192.168.8.55",
          "port" => 59048
}

（9）使用filter

編寫日誌採集文件：

[root@elk-01 ~]# cat grok.conf 
input {
    stdin { }
}
filter {
    grok {
        match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" }
    }
}

output {
    stdout{
        codec => "rubydebug"
    }
}

對上面的採集文件進行執行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f grok.conf 
Settings: Default pipeline workers: 1
Pipeline main started
55.3.244.1 GET /index.html 15824 0.043   # 輸入左邊內容，下面就會自動生成字典的形式
{
       "message" => "55.3.244.1 GET /index.html 15824 0.043",
      "@version" => "1",
    "@timestamp" => "2018-05-07T09:14:56.201Z",
          "host" => "elk-01.com",
        "client" => "55.3.244.1",
        "method" => "GET",
       "request" => "/index.html",
         "bytes" => "15824",
      "duration" => "0.043"
}

其實上面使用的那些正則在程序中都有定義：

[root@elk-01 patterns]# cd /elk/logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.5/patterns
[root@elk-01 patterns]# ls
aws     bro   firewalls      haproxy  junos         mcollective           mongodb  postgresql  redis
bacula  exim  grok-patterns  java     linux-syslog  mcollective-patterns  nagios   rails       ruby
[root@elk-01 patterns]# cat grok-patterns

（10）mysql慢查詢

編寫日誌採集文件：

[root@elk-01 ~]# cat mysql-slow.conf 
input {
    file {
        path => "/var/lib/mysql/elk-01-slow.log"
        type => "mysql-slowlog"
        codec => multiline {
            pattern => "^# User@Host"
            negate => true
            what => "previous"
        }
    }
}

filter {
    # drop sleep events
    grok {
        match => { "message" =>"SELECT SLEEP" }
        add_tag => [ "sleep_drop" ]
        tag_on_failure => [] # prevent default _grokparsefailure tag on real records
    }
    if "sleep_drop" in [tags] {
        drop {}
    }
    grok {
        match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s+Id: %{NUMBER:row_id:int}\s*# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*(?:use %{DATA:database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<query>(?<action>\w+)\s+.*)\n#\s*" ]
    }
    date {
        match => [ "timestamp", "UNIX" ]
        remove_field => [ "timestamp" ]
    }
}

output {
    stdout {
       codec =>"rubydebug"
    }
}

安裝好mysql數據庫以後，在配置文件my.cnf

[root@elk-01 ~]# vim /etc/my.cnf   --在[mysqld]組裏面添加如下內容
slow_query_log      # 打開慢查詢日誌
slow_query_log_file=/var/lib/mysql/elk-01-slow.log   # 指定慢查詢日誌所在文件
long_query_time=2               # 設一個閾值，要大於這個值纔會記錄，等於該值時不記錄。
log_queries_not_using_indexes   # 若是運行的SQL語句沒有使用索引，則MySQl數據庫一樣會將這條SQL語句記錄到慢查詢日誌文件

對上面的採集文件進行執行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/mysql-slow.conf
Settings: Default pipeline workers: 1
Pipeline main started
{
    "@timestamp" => "2018-05-08T04:19:49.871Z",
       "message" => "/usr/libexec/mysqld, Version: 5.5.52-MariaDB (MariaDB Server). started with:\nTcp port: 0  Unix socket: /var/lib/mysql/mysql.sock\nTime                 Id Command    Argument\n# Time: 180508 12:19:47",
      "@version" => "1",
          "tags" => [
        [0] "multiline",
        [1] "_grokparsefailure"
    ],
          "path" => "/var/lib/mysql/elk-01-slow.log",
          "host" => "elk-01.com",
          "type" => "mysql-slowlog"
}

再次把mysql-slow.conf文件的內容添加到總文件file.conf中，或者修改mysql-slow.conf文件，讓數據展現在ES和Kibana中：

[root@elk-01 ~]# cat mysql-slow.conf 
input {
    file {
        path => "/var/lib/mysql/elk-01-slow.log"
        type => "mysql-slowlog"
        start_position => "beginning"
        codec => multiline {
            pattern => "^# User@Host"
            negate => true
            what => "previous"
        }
    }
}

filter {
    # drop sleep events
    grok {
        match => { "message" =>"SELECT SLEEP" }
        add_tag => [ "sleep_drop" ]
        tag_on_failure => [] # prevent default _grokparsefailure tag on real records
    }
    if "sleep_drop" in [tags] {
        drop {}
    }
    grok {
        match => [ "message", "(?m)^# User@Host: %{USER:user}\[[^\]]+\] @ (?:(?<clienthost>\S*) )?\[(?:%{IP:clientip})?\]\s+Id: %{NUMBER:row_id:int}\s*# Query_time: %{NUMBER:query_time:float}\s+Lock_time: %{NUMBER:lock_time:float}\s+Rows_sent: %{NUMBER:rows_sent:int}\s+Rows_examined: %{NUMBER:rows_examined:int}\s*(?:use %{DATA:database};\s*)?SET timestamp=%{NUMBER:timestamp};\s*(?<query>(?<action>\w+)\s+.*)\n#\s*" ]
    }
    date {
        match => [ "timestamp", "UNIX" ]
        remove_field => [ "timestamp" ]
    }
}

output {
    if [type] == " mysql-slowlog"{
        elasticsearch {
            hosts => ["192.168.8.55:9200"]
            index => "mysql-slowlog-%{+YYYY.MM.dd}"
        }
    }
}

對上面的採集文件進行執行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/mysql-slow.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/mysql-slow.conf &

登錄Elasticsearch主界面查看：

登錄Kibana主界面查看：

接下來，咱們試想一個問題：
若是咱們的Elasticsearch出現了問題，那就不能進行日誌採集處理了！
這種狀況下該怎麼辦呢？
解決方案：
能夠在Client和Elasticsearch之間添加一箇中間件做爲緩存，先將採集到的日誌內容寫到中間件上，而後再從中間件輸入到Elasticsearch中。
這樣，就完美的解決了上述的問題了。

七、ELK中使用redis做爲中間件，緩存日誌採集數據

（1）redis的配置和啓動

# 下載解壓
[root@elk-01 ~]# wget http://download.redis.io/releases/redis-4.0.8.tar.gz
[root@elk-01 ~]# tar zxvf redis-4.0.8.tar.gz -C /usr/src/
[root@elk-01 ~]# cd /usr/src/redis-4.0.8/
# 進入解壓後的文件目錄，以後直接編譯便可（redis安裝相對簡單）
[root@elk-01 redis-4.0.8]# make
[root@elk-01 redis-4.0.8]# make install
[root@elk-01 redis-4.0.8]# cd utils
[root@elk-01 utils]# ./install_server.sh
Selected config:
Port           : 6379
Config file    : /etc/redis/6379.conf
Log file       : /var/log/redis_6379.log
Data dir       : /var/lib/redis/6379
Executable     : /usr/local/bin/redis-server
Cli Executable : /usr/local/bin/redis-cli

# 修改配置文件和啓動
[root@elk-01 ~]# vim /etc/redis/6379.conf   --修改下面內容
# bind 127.0.0.1   # 註釋掉
protected-mode no  # 保護模式爲no
daemonize yes  # 守護進程模式爲yes

# 啓動
[root@elk-01 ~]# redis-server /etc/redis/6379.conf
[root@elk-01 ~]# lsof -i:6379
COMMAND     PID USER   FD   TYPE DEVICE SIZE/OFF NODE NAME
redis-ser 27502 root    6u  IPv6 380301      0t0  TCP *:6379 (LISTEN)
redis-ser 27502 root    7u  IPv4 380302      0t0  TCP *:6379 (LISTEN)
# 登錄
[root@elk-01 ~]# redis-cli -h 192.168.8.55 
192.168.8.55:6379> info
# Server
redis_version:4.0.8
……

（2）編寫從Client端採集數據的文件

編寫日誌採集文件：

[root@elk-01 ~]# cat redis-out.conf
input {
   stdin {}
}

output {
   redis {
      host => "192.168.8.55"
      port => "6379"
      db => "6"
      data_type => "list"
      key => "demo"
   }
}

（3）執行採集數據的文件，並輸入數據hello redis

[root@elk-01 ~]# /elk/logstash/bin/logstash -f redis-out.conf
Settings: Default pipeline workers: 1
Pipeline main started    # 下面輸入數據hello redis
hello redis

（4）在redis中查看數據

從新開啓一個窗口，在redis中查看數據：

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> info  # 輸入內容
# Server
redis_version:4.0.8
……
……
# Keyspace   # 在最下面一行，顯示是db6
db6:keys=1,expires=0,avg_ttl=0
192.168.8.55:6379> select 6  # 輸入內容
OK
192.168.8.55:6379[6]> keys *  # 輸入內容
1) "demo"
192.168.8.55:6379[6]> LINDEX demo -1  # 輸入內容
"{\"message\":\"hello redis\",\"@version\":\"1\",\"@timestamp\":\"2018-05-08T08:53:16.806Z\",\"host\":\"elk-01.com\"}"

（5）繼續自定義輸入數據

[root@elk-01 ~]# /elk/logstash/bin/logstash -f redis-out.conf 
Settings: Default pipeline workers: 1
Pipeline main started
hello redis
123456
xiaozuo
51cto
test
haha

（6）在redis中查看

在redis中查看長度：

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> info  # 輸入內容
# Server
redis_version:4.0.8
……
……
# Keyspace
db6:keys=1,expires=0,avg_ttl=0  # 顯示是db6
192.168.8.55:6379> select 6  # 輸入內容
OK
192.168.8.55:6379[6]> keys *  # 輸入內容
1) "demo"
192.168.8.55:6379[6]> LLEN demo  # 輸入內容
(integer) 6  # 剛纔自定義輸入了6個字符串

（7）將redis中的內容寫到ES中

編寫日誌採集文件：

[root@elk-01 ~]# cat redis-in.conf 
input {
    redis {
        host => "192.168.8.55"
        port => "6379"
        db => "6"
        data_type => "list"
        key => "demo"
    }
}

output {
    elasticsearch {
        hosts => ["192.168.8.55:9200"]
        index => "redis-in-%{+YYYY.MM.dd}"
    }
}

對上面的採集文件進行執行：

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/redis-in.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/redis-in.conf &

在redis中查看，發現數據已被讀出：

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> LLEN demo
(integer) 0

登錄Elasticsearch主界面查看：

（8）將採集到的全部日誌寫入到redis中

我這裏從新定義一個添加redis緩存後的總文件shipper.conf。（能夠將以前執行的總文件file.conf停掉）
編寫總採集shipper.conf文件：

[root@elk-01 ~]# cat shipper.conf 
input {
    file {
        path => "/var/log/messages"
        type => "system"
        start_position => "beginning"
    }
    file {
        path => "/var/log/nginx/access_json.log"
        codec => json
        start_position => "beginning"
        type => "nginx-log"
    }
    syslog {
        type => "system-syslog"
        host => "192.168.8.55"
        port => "514"
    }
}

output {
    if [type] == "system"{
        redis {
            host => "192.168.8.55"
            port => "6379"
            db => "6"
            data_type => "list"
            key => "system"
        }
    }
    if [type] == "nginx-log"{   
        redis {
            host => "192.168.8.55"
            port => "6379"
            db => "6"
            data_type => "list"
            key => "nginx-log"
        }
    }
    if [type] == "system-syslog"{
        redis {
            host => "192.168.8.55"
            port => "6379"
            db => "6"
            data_type => "list"
            key => "system-syslog"
        }   
    }
}

對上面的採集文件進行執行（提早將上面以前啓動的file.conf文件的執行給結束掉！）

[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/shipper.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /root/shipper.conf &

在redis中查看：

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> info  # 輸入內容
# Server
redis_version:4.0.8
……
……
# Keyspace
db6:keys=1,expires=0,avg_ttl=0  # 顯示是db6
192.168.8.55:6379> select 6   # 輸入內容
OK
192.168.8.55:6379[6]> keys *  # 輸入內容
1) "system"
192.168.8.55:6379[6]> keys *  # 輸入內容
1) "nginx-log"
2) "system"

從新開啓一個窗口，自定義添加日誌數據：

[root@elk-01 ~]# logger "Hello World"
[root@elk-01 ~]# logger "123456"
[root@elk-01 ~]# logger "test"
[root@elk-01 ~]# logger "hello"
[root@elk-01 ~]# logger "hello"
[root@elk-01 ~]# logger "hello"

回到原來的窗口（即上面執行redis命令的終端），查看內容：

192.168.8.55:6379[6]> keys *  # 輸入內容
1) "system-syslog"   # 新增的日誌
2) "nginx-log"
3) "system"

其實能夠在任意的一臺ES中，將數據從redis讀取到ES中。
下面咱們在elk-02節點上，將數據從redis讀取到ES中：
編寫日誌採集文件：

[root@elk-02 ~]# cat file.conf 
input {
    redis {
        type => "system"
        host => "192.168.8.55"
        port => "6379"
        db => "6"
        data_type => "list"
        key => "system"
    }
    redis {
        type => "nginx-log"
        host => "192.168.8.55"
        port => "6379"
        db => "6"
        data_type => "list"
        key => "nginx-log"
    }
    redis {
        type => "system-syslog"
        host => "192.168.8.55"
        port => "6379"
        db => "6"
        data_type => "list"
        key => "system-syslog"
    }   
}

output { 
    if [type] == "system"{
        elasticsearch {
            hosts => ["192.168.8.55:9200"]
            index => "system-%{+YYYY.MM.dd}"
        }
    }
    if [type] == "nginx-log"{
        elasticsearch {
            hosts => ["192.168.8.55:9200"]
            index => "nignx-log-%{+YYYY.MM.dd}"
        }
    }
    if [type] == "system-syslog"{
        elasticsearch {
            hosts => ["192.168.8.55:9200"]
            index => "system-syslog-%{+YYYY.MM.dd}"
        }
    }
}

對上面的日誌採集文件進行執行：

[root@elk-02 ~]# /elk/logstash/bin/logstash -f /root/file.conf --configtest
Configuration OK
[root@elk-02 ~]# /elk/logstash/bin/logstash -f /root/file.conf &

在redis中查看，發現數據已經被讀出到Elasticsearch中了。

[root@elk-01 ~]# redis-cli -h 192.168.8.55
192.168.8.55:6379> keys *
(empty list or set)

同時登錄Logstash和Kibana看，發現能夠正常採集到日誌了。

也能夠啓動多個redis寫到ES中，具體根據本身的實際狀況而定。

當客戶機的日誌信息收集後，通過redis剛讀到ES數據庫裏後，若是沒有新數據寫入，則默認在ES的訪問界面裏是看不到數據的。只有當日志文件裏有新的日誌寫入後纔會觸發數據展現的動做，即ES的訪問界面（http://192.168.8.55:9200/_plugin/head/）裏才能看到日誌數據的展現效果。

4、配置示例

# 192.168.8.55爲ELK的master節點，同時也是redis節點
# 編寫日誌採集文件
[root@elk-01 ~]# cat /conf/tomacat-log.conf 
input {
    file {
        path => "/usr/local/tomcat/logs/catalina.out"
        type => "tomcat-logs"
        start_position => "beginning"
        codec => multiline {
            pattern => "^\["           # 表示採集以「[」開頭的日誌信息
            negate => true
            what => "previous"
        }
    }
}

output {
    if [type] == "tomcat-logs"{
        redis {
            host => "192.168.8.55"
            port => "6379"
            db => "1"
            data_type => "list"
            key => "tomcat-logs"
        }
    }
}

[root@elk-01 ~]# cat /conf/system-log.conf 
input {
    file {
        path => "/var/log/messages"
        type => "systemlog"
        start_position => "beginning"
        stat_interval => "2"
    }
}

output {
    if [type] == "systemlog" {
        redis {
            data_type => "list"
            host => "192.168.10.44"
            db => "2"
            port => "6379"
            key => "systemlog"
        }
    }
}

[root@elk-01 ~]# cat /conf/file.conf 
input {
    redis {
        type => "tomcat8-logs"
        host => "192.168.10.44"
        port => "6379"
        db => "1"
        data_type => "list"
        key => "tomcat8-logs"
    }

    redis {
        type => "systemlog"
        host => "192.168.10.44"
        port => "6379"
        db => "2"
        data_type => "list"
        key => "systemlog"
    }
}

output {
    if [type] == "tomcat8-logs"{
        elasticsearch {
            hosts => ["192.168.10.44:9200"]
            index => "elk-node2-tomcat8-logs-%{+YYYY.MM.dd}"
        }
    }

    if [type] == "systemlog"{
        elasticsearch {
            hosts => ["192.168.10.44:9200"]
            index => "elk-node2-systemlog-%{+YYYY.MM.dd}"
        }
    }
}

# 驗證
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/tomacat-log.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/system-log.conf --configtest
Configuration OK
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/file.conf --configtest
Configuration OK

# 執行
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/tomacat-log.conf &
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/system-log.conf &
[root@elk-01 ~]# /elk/logstash/bin/logstash -f /conf/file.conf &

當/usr/local/tomcat/logs/catalina.out和/var/log/messages文件裏有新日誌信息寫入時，就會觸發動做，在redis裏就能查看到相關信息，並查看寫入到ES裏。

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。