ELK 大家應(yīng)該比較熟悉了,，zabbix 應(yīng)該也不陌生，那么將 ELK 和 zabbix 放到一起的話,，可能大家就會有疑問了,？這兩個放到一起是什么目的呢，聽我細(xì)細(xì)道來

　　ELK 是一套日志收集套件,，它其實是由 Elasticsearch,、Logstash 和 Kibana 三個軟件組成,，通過 ELK 可以收集系統(tǒng)日志,、網(wǎng)站日志、應(yīng)用系統(tǒng)日志等各種日志數(shù)據(jù),，并且還可以對日志進行過濾,、清洗，然后進行集中存放并可用于實時檢索,、分析,。這是 ELK 的基礎(chǔ)功能。

　　但是有些時候,，我們希望在收集日志的時候,，能夠?qū)⑷罩局械漠惓Ｐ畔ⅲň妗㈠e誤,、失敗等信息）及時地提取出來,，因為日志中的異常信息意味著操作系統(tǒng)、應(yīng)用程序可能存在故障,，如果能將日志中的故障信息及時的告知運維人員,，那么運維就可以第一時間去進行故障排查和處理，進而也就可以避免很多故障的發(fā)生,。

　　那么如何才能做到將 ELK 收集的日志數(shù)據(jù)中出現(xiàn)的異常信息及時地告知運維人員呢,，這就需要用到 zabbix 了，ELK（更確切的說應(yīng)該是 logstash）可以實時地讀取日志的內(nèi)容,，并且還可以過濾日志信息,，通過 ELK 的讀取和過濾功能，就可以將日志中的一些異常關(guān)鍵字（error,、failed,、OutOff、Warning）過濾出來,，然后通過 logstash 的 zabbix 插件將這個錯誤日志信息發(fā)送給 zabbix,，那么 zabbix 在接收到這個數(shù)據(jù)后,，結(jié)合自身的機制，然后發(fā)起告警動作,，這樣就實現(xiàn)了日志異常 zabbix 實時告警的功能了,。

　　Logstash 支持多種輸出介質(zhì),，比如 syslog,、HTTP、TCP,、elasticsearch,、kafka 等，而有時候我們想將收集到的日志中一些錯誤信息輸出,，并告警時,，就用到了 logstash-output-zabbix 這個插件，此插件可以將 Logstash 與 zabbix 進行整合,，也就是將 Logstash 收集到的數(shù)據(jù)進行過濾,，將有錯誤標(biāo)識的日志輸出到 zabbix 中，最后通過 zabbix 的告警機制進行觸發(fā),、告警,。

　　logstash-output-zabbix 是一個社區(qū)維護的插件，它默認(rèn)沒有在 Logstash 中安裝,但是安裝起來也很容易,，直接在 logstash 中運行如下命令即可：

　　[root@elk-master bin]# /usr/share/logstash/bin/logstash-plugin install logstash-output-zabbix

　　其中,，/usr/share/logstash/bin/是 Logstash 的 yum 默認(rèn)安裝目錄，如果是源碼安裝的需要自己修改

　　此外,，logstash-plugin 命令還有多種用法,，我們來看一下：

　　列出所有已安裝的插件

　　[root@elk-master ~]# /usr/share/logstash/bin/logstash-plugin list

　　列出已安裝的插件及版本信息

　　[root@elk-master ~]# /usr/share/logstash/bin/logstash-plugin list --verbose

　　列出包含 namefragment 所有已安裝插件

　　[root@elk-master ~]# /usr/share/logstash/bin/logstash-plugin list "*namefragment*"

　　列出特定組的所有已安裝插件（ input，filter,，codec,，output）

　　[root@elk-master ~]# /usr/share/logstash/bin/logstash-plugin --group output

　　要安裝某個插件，例如安裝 kafka 插件,，可執(zhí)行如下命令：

　　[root@elk-master ~]# /usr/share/logstash/bin/logstash-plugin install logstash-output-kafka

　　要使用此命令安裝插件,，需要你的電腦可以訪問互聯(lián)網(wǎng)。此插件安裝方法,，會檢索托管在公共存儲庫（RubyGems）上的插件,，然后下載到本地機器并在 Logstash 安裝之上進行自動安裝

　　每個插件有自己的發(fā)布周期和版本更新，這些更新通常是獨立于 Logstash 的發(fā)布周期的,。因此,，有時候需要單獨更新插件，可以使用 update 子命令獲得最新版本的插件

　　更新所有已安裝的插件

　　[root@elk-master ~]# /usr/share/logstash/bin/logstash-plugin update

　　僅更新指定的插件

　　[root@elk-master ~]# /usr/share/logstash/bin/logstash-plugin update logstash-output-kafka

　　如果需要從 Logstash 插件中刪除插件,，可執(zhí)行如下命令：

　　[root@elk-master ~]# /usr/share/logstash/bin/logstash-plugin remove logstash-output-kafka

　　這樣就刪除了 logstash-output-kafka 插件

　　logstash-output-zabbix 安裝好之后,，就可以在 logstash 配置文件中使用了,，下面是一個 logstash-output-zabbix 使用的例子：

　　zabbix {

　　zabbix_host=>"[@metadata][zabbix_host]"zabbix_key=>"[@metadata][zabbix_key]"zabbix_server_host=>"x.x.x.x"zabbix_server_port=>"xxxx"

　　zabbix_value="xxxx"

　　}

　　其中：

　　zabbix_host：表示 Zabbix 主機名字段名稱， 可以是單獨的一個字段,， 也可以是 @metadata 字段的子字段,， 是必需的設(shè)置，沒有默認(rèn)值,。

　　zabbix_key：表示 Zabbix 項目鍵的值,，也就是 zabbix 中的 item，此字段可以是單獨的一個字段,， 也可以是 @metadata 字段的子字段,，沒有默認(rèn)值。

　　zabbix_server_host：表示 Zabbix 服務(wù)器的 IP 或可解析主機名,，默認(rèn)值是 “l(fā)ocalhost”,，需要設(shè)置為 zabbix server 服務(wù)器所在的地址。

　　zabbix_server_port：表示 Zabbix 服務(wù)器開啟的監(jiān)聽端口,，默認(rèn)值是 10051,。

　　zabbix_value：表示要發(fā)送給 zabbix item 監(jiān)控項的值對應(yīng)的字段名稱，默認(rèn)值是 “message”,，也就是將”message”字段的內(nèi)容發(fā)送給上面 zabbix_key 定義的 zabbix item 監(jiān)控項，當(dāng)然也可以指定一個具體的字段內(nèi)容發(fā)送給 zabbix item 監(jiān)控項,。

　　這里我們以 logstash 收集日志,，然后對日志進行讀取，最后選擇關(guān)鍵字進行過濾并調(diào)用 zabbix 告警的流程,，來看看如何配置 logstash 實現(xiàn) zabbix 告警,。

　　先說明一下我們的應(yīng)用需求：通過對系統(tǒng)日志文件的監(jiān)控，然后去過濾日志信息中的一些關(guān)鍵字,，例如 ERROR,、Failed、WARNING 等,，將日志中這些信息過濾出來,，然后發(fā)送到 zabbix 上，最后借助 zabbix 的報警功能實現(xiàn)對系統(tǒng)日志中有上述關(guān)鍵字的告警,。

　　對于過濾關(guān)鍵字,，進行告警，不同的業(yè)務(wù)系統(tǒng),，可能關(guān)鍵字不盡相同,，例如對 http 系統(tǒng)，可能需要過濾 500,、403,、503 等這些錯誤碼,，對于 java 相關(guān)的系統(tǒng)，可能需要過濾 OutOfMemoryError,、PermGen,、Java heap 等關(guān)鍵字。在某些業(yè)務(wù)系統(tǒng)的日志輸出中,，可能還有一些自定義的錯誤信息,，那么這些也需要作為過濾關(guān)鍵字來使用。

　　接下來就是創(chuàng)建一個 logstash 事件配置文件,，這里將配置文件分成三個部分來介紹,，首先是 input 部分，內(nèi)容如下：

　　input {

　　file {

　　path="/var/log/secure"type="system"

　　start_position="beginning"

　　}

　　}

　　input 部分是從/var/log/secure 文件中讀取數(shù)據(jù),，start_position 表示從 secure 文件開頭讀取內(nèi)容,。接著是 filter 部分，內(nèi)容如下：

　　filter {

　　grok {

　　match=> { "message"=> "%{SYSLOGTIMESTAMP:message_timestamp} %{SYSLOGHOST:hostname} %{DATA:message_program}(?:\[%{POSINT:message_pid}\])?: %{GREEDYDATA:message_content}" }#這里通過 grok 對 message 字段的數(shù)據(jù)進行字段劃分,，這里將 message 字段劃分了 5 個子字段,。其中，message_content 字段會在 output 中用到

　　}

　　mutate {

　　add_field=> ["[zabbix_key]","oslogs"] #新增的字段,，字段名是 zabbix_key,，值為 oslogsadd_field=> ["[zabbix_host]","Zabbix server"] #新增的字段，字段名是 zabbix_host,，值可以在這里直接定義,，也可以引用字段變量來獲取。這里的%{host}獲取的就是日志數(shù)據(jù)的主機名,，這個主機名與 zabbix web 中“主機名稱”需要保持一致 remove_field=> ["@version","message"] #這里是刪除不需要的字段

　　}

　　date { #這里是對日志輸出中的日期字段進行轉(zhuǎn)換,，其中 message_timestamp 字段是默認(rèn)輸出的時間日期字段，將這個字段的值傳給 @timestamp 字段 match=> [ "message_timestamp","MMM d HH:mm:ss","MMM dd HH:mm:ss","ISO8601"]

　　}

　　}

　　filter 部分是個重點,，在這個部分中,，重點關(guān)注的是 message_timestamp 字段、message_content 字段

　　最后是 output 部分,，內(nèi)容如下

　　output {

　　elasticsearch{ #這部分是為把日志次發(fā)送到 es 中便于 kibana 中進行展示,，如果沒有部署的可以去掉 index=> "oslogs-%{+YYYY.MM.dd}"hosts=> ["192.168.73.133:9200"]

　　user=> "elastic"password=> "Goldwind@2019"sniffing=> false

　　}

　　if [message_content]=~ /(ERR|error|ERROR|Failed)/ { #定義在 message_content 字段中，需要過濾的關(guān)鍵字信息,，也就是在 message_content 字段中出現(xiàn)給出的這些關(guān)鍵字,，那么就將這些信息發(fā)送給 zabbixzabbix {

　　zabbix_host=>"[zabbix_host]" #這個 zabbix_host 將獲取上面 filter 部分定義的字段變量%{host}的值 zabbix_key=>"[zabbix_key]" #這個 zabbix_key 將獲取上面 filter 部分中給出的值 zabbix_server_host=>"192.168.73.133" #這是指定 zabbix server 的 IP 地址 zabbix_server_port=>"10051" #這是指定 zabbix server 的監(jiān)聽端口 zabbix_value=>"message_content" #這個很重要，指定要傳給 zabbix 監(jiān)控項 item（oslogs）的值,， zabbix_value 默認(rèn)的值是"message"字段,，因為上面我們已經(jīng)刪除了"message"字段，因此,，這里需要重新指定,，根據(jù)上面 filter 部分對"message"字段的內(nèi)容劃分,，這里指定為"message_content"字段，其實,"message_content"字段輸出的就是服務(wù)器上具體的日志內(nèi)容

　　}

　　}

　　#stdout{ codec=> rubydebug }

　　} #這里是開啟調(diào)試模式,，當(dāng)?shù)谝淮闻渲玫臅r候,，建議開啟，這樣過濾后的日志信息直接輸出的屏幕,，方便進行調(diào)試,，調(diào)試成功后，即可關(guān)閉

　　將上面三部分內(nèi)容合并到一個文件 file_to_zabbix.conf 中,，然后啟動 logstash 服務(wù)：

　　[root@logstashserver ~]#cd /usr/local/logstash

　　[root@logstashserver logstash]#nohup bin/logstash -f config/file_to_zabbix.conf --path.data /tmp/ &

　　這里的—path.data 是指定此 logstash 進程的辦統(tǒng)招文憑數(shù)據(jù)存儲目錄,，用于在一個服務(wù)器上啟動多個 logstash 進程的環(huán)境中

　　登錄 zabbix web 平臺，選擇配置—->模板—->創(chuàng)建模板,，名稱定為 logstash-output-zabbix,，如下圖所示：

　　接著,，在此模塊下創(chuàng)建一個應(yīng)用集,，點擊應(yīng)用集——->創(chuàng)建應(yīng)用集，如下圖所示：

　　然后,，在此模塊下創(chuàng)建一個監(jiān)控項，點擊監(jiān)控項——->創(chuàng)建監(jiān)控項,，如下圖所示：

　　到此為止，zabbix 監(jiān)控 logstash 的日志數(shù)據(jù)配置完成

　　這里我們以客戶端 192.168.73.135 主機為例,，也就是監(jiān)控 192.168.73.135 主機上的系統(tǒng)日志數(shù)據(jù)，當(dāng)發(fā)現(xiàn)日志異常就進行告警

　　在上面創(chuàng)建了一個模板和監(jiān)控項,，接著還需要將此模板鏈接到 192.168.73.135 主機上,，選擇“配置”—-“主機”，然后選中 192.168.73.135 主機,，選擇“模板”標(biāo)簽,，將上面創(chuàng)建的模板加入到 192.168.73.135 主機上，如下圖所示：

　　這樣,，上面創(chuàng)建的監(jiān)控項在 192.168.73.135 主機上就自動生效了

　　下面我們模擬一個故障，在任意主機通過 ssh 登錄 192.168.73.135 主機,，然后輸入一個錯誤密碼,，讓系統(tǒng)的/var/log/secure 文件產(chǎn)生錯誤日志，然后看看 logstash 是否能夠過濾到,，是否能夠發(fā)送到 zabbix 中

　　首先讓系統(tǒng)文件/var/log/secure 產(chǎn)生類似如下內(nèi)容：

　　[root@k8s-node03 ~]# tail /var/log/secure

　　Dec 25 22:57:44 k8s-node03 sshd[49803]: Failed password for root from 192.168.73.1 port 60256 ssh2

　　Dec 25 22:57:48 k8s-node03 sshd[49803]: error: Received disconnect from 192.168.73.1 port 60256:13: The user canceled authentication. [preauth]

　　Dec 25 22:57:48 k8s-node03 sshd[49803]: Disconnected from 192.168.73.1 port 60256 [preauth]

　　Dec 25 23:02:40 k8s-node03 sshd[49908]: reverse mapping checking getaddrinfo for bogon [192.168.73.1] failed - POSSIBLE BREAK-IN ATTEMPT!

　　Dec 25 23:02:43 k8s-node03 unix_chkpwd[49911]: password check failed for user (root)

　　Dec 25 23:02:43 k8s-node03 sshd[49908]: pam_unix(sshd:auth): authentication failure; logname= uid=0 euid=0 tty=ssh ruser= rhost=192.168.73.1 user=root

　　Dec 25 23:02:43 k8s-node03 sshd[49908]: pam_succeed_if(sshd:auth): requirement "uid >=1000" not met by user "root"

　　Dec 25 23:02:44 k8s-node03 sshd[49908]: Failed password for root from 192.168.73.1 port 60757 ssh2

　　Dec 25 23:02:46 k8s-node03 sshd[49908]: error: Received disconnect from 192.168.73.1 port 60757:13: The user canceled authentication. [preauth]

　　Dec 25 23:02:46 k8s-node03 sshd[49908]: Disconnected from 192.168.73.1 port 60757 [preauth]

　　這里面有我們要過濾的關(guān)鍵字 Failed,，因此 logstash 會將此內(nèi)容過濾出來,，發(fā)送到 zabbix 上

　　接著，登錄 zabbix web 平臺,，點擊監(jiān)測中——->最新數(shù)據(jù),，如果 zabbix 能夠接收到日志，就可以看到下圖的最新數(shù)據(jù)：

　　點擊歷史記錄,，可以查看詳細(xì)內(nèi)容，如下圖所示：

　　可以看到,，紅框中的內(nèi)容就是在 logstash 中定義的 message_content 字段的內(nèi)容,到這里為止，zabbix 已經(jīng)可以收到 logstash 的發(fā)送過來的數(shù)據(jù)了,，但是要實現(xiàn)報警,，還需要在 zabbix 中創(chuàng)建一個觸發(fā)器，進入配置——->模板,，選擇 logstash-output-zabbix 這個模板,，然后點擊上面的觸發(fā)器，繼續(xù)點擊右上角的創(chuàng)建觸發(fā)器,，如下圖所示：

　　這里注意觸發(fā)器創(chuàng)建中，表達式的寫法,，這里觸發(fā)器的觸發(fā)條件是：

　　如果接收到 logstash 發(fā)送過來的數(shù)據(jù),，就進行告警，也就是說接收到的數(shù)據(jù),，如果長度大于 0 就告警

　　觸發(fā)器配置完成后,，如果配置正常，就會進行告警了,，使用的是釘釘進行圖文告警,，告警內(nèi)容如下圖所示：

　　Kibana 上查看安全日志

　　首先我們來捋一下思路：

　　我們的架構(gòu)基本不變,，仍然是 filebat 收集日志推送到 kibana 消息隊列，然后由 Logstash 前去拉取日志數(shù)據(jù),，經(jīng)過處理最后中轉(zhuǎn)出去,；只不過是中轉(zhuǎn)輸出到 zabbix 上面而已；能夠?qū)崿F(xiàn)這個功能的,，最核心的功臣就是 Logsatsh 的插件（logstash-output-zabbix）,；

　　在這里需要注意的是:filebeat 收集端的 IP 一定要與 zabbix 監(jiān)控主機的 IP 相對應(yīng)，否則日志是過不來的~

　　分享一個小技巧：通過該命令可以測試定義在 zabbix 上的鍵值；出現(xiàn)以下輸出變?yōu)檎,，如果 failed 非零值表示失敗

　　#在 server 端使用 zabbix_sender 向 zabbix 發(fā)送測試

　　[root@localhost zabbix_sender]# /usr/local/zabbix/bin/zabbix_sender -s 192.168.73.135 -z 192.168.73.133 -k "oslogs" -o 1

　　info from server: "processed: 1; failed: 0; total: 1; seconds spent: 0.000081"

　　sent: 1; skipped: 0; total: 1

　　詳解：-s：指定本地 agent 端

　　-z：指定 zabbix 服務(wù)端

　　-k：指定鍵值