1 ELK與ZABBIX有什么關(guān)系?

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

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

但是有些時(shí)候，我們希望在收集日志的時(shí)候，能夠?qū)⑷罩局械漠惓Ｐ畔?警告、錯(cuò)誤、失敗等信息)及時(shí)的提取出來，因?yàn)槿罩局械漠惓Ｐ畔⒁馕吨僮飨到y(tǒng)、應(yīng)用程序可能存在故障，如果能將日志中的故障信息及時(shí)的告知運(yùn)維人員，那么運(yùn)維就可以第一時(shí)間去進(jìn)行故障排查和處理，進(jìn)而也就可以避免很多故障的發(fā)生。

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

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

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

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

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

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

2.1 列出目前已經(jīng)安裝的插件

列出所有已安裝的插件

[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

2.2 安裝插件

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

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

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

2.3 更新插件

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

更新所有已安裝的插件

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

僅更新指定的插件

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

2.4 刪除插件

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

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

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

3 logstash-output-zabbix插件的使用

logstash-output-zabbix安裝好之后，就可以在logstash配置文件中使用了，下面是一個(gè)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主機(jī)名字段名稱， 可以是單獨(dú)的一個(gè)字段， 也可以是 @metadata 字段的子字段， 是必需的設(shè)置，沒有默認(rèn)值。
• zabbix_key：表示Zabbix項(xiàng)目鍵的值，也就是zabbix中的item，此字段可以是單獨(dú)的一個(gè)字段， 也可以是 @metadata 字段的子字段，沒有默認(rèn)值。
• zabbix_server_host：表示Zabbix服務(wù)器的IP或可解析主機(jī)名，默認(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)控項(xiàng)的值對應(yīng)的字段名稱，默認(rèn)值是 “message”，也就是將”message”字段的內(nèi)容發(fā)送給上面zabbix_key定義的zabbix item監(jiān)控項(xiàng)，當(dāng)然也可以指定一個(gè)具體的字段內(nèi)容發(fā)送給zabbix item監(jiān)控項(xiàng)。

4 將logstash與zabbix進(jìn)行整合

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

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

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

4.1 配置logstash事件配置文件

接下來就是創(chuàng)建一個(gè)logstash事件配置文件，這里將配置文件分成三個(gè)部分來介紹，首先是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ù)進(jìn)行字段劃分，這里將message字段劃分了5個(gè)子字段。其中，message_content字段會在output中用到
         }
    mutate {
           add_field => ["[zabbix_key]","oslogs"] #新增的字段，字段名是zabbix_key，值為oslogs
           add_field => ["[zabbix_host]","Zabbix server"] #新增的字段，字段名是zabbix_host，值可以在這里直接定義，也可以引用字段變量來獲取。這里的%{host}獲取的就是日志數(shù)據(jù)的主機(jī)名，這個(gè)主機(jī)名與zabbix web中“主機(jī)名稱”需要保持一致
           remove_field => ["@version","message"] #這里是刪除不需要的字段
        }
    date { #這里是對日志輸出中的日期字段進(jìn)行轉(zhuǎn)換，其中message_timestamp字段是默認(rèn)輸出的時(shí)間日期字段，將這個(gè)字段的值傳給 @timestamp字段
           match => [ "message_timestamp","MMM  d HH:mm:ss","MMM dd HH:mm:ss","ISO8601"]
        }
}

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

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

output {
        elasticsearch{  #這部分是為把日志次發(fā)送到es中便于kibana中進(jìn)行展示，如果沒有部署的可以去掉
                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ā)送給zabbix
                   zabbix {
                            zabbix_host =>"[zabbix_host]" #這個(gè)zabbix_host將獲取上面filter部分定義的字段變量%{host}的值
                            zabbix_key =>"[zabbix_key]" #這個(gè)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" #這個(gè)很重要，指定要傳給zabbix監(jiān)控項(xiàng)item（oslogs）的值， zabbix_value默認(rèn)的值是"message"字段，因?yàn)樯厦嫖覀円呀?jīng)刪除了"message"字段，因此，這里需要重新指定，根據(jù)上面filter部分對"message"字段的內(nèi)容劃分，這里指定為"message_content"字段，其實(shí),"message_content"字段輸出的就是服務(wù)器上具體的日志內(nèi)容
                        }
       }
        #stdout{ codec => rubydebug }
} #這里是開啟調(diào)試模式，當(dāng)?shù)谝淮闻渲玫臅r(shí)候，建議開啟，這樣過濾后的日志信息直接輸出的屏幕，方便進(jìn)行調(diào)試，調(diào)試成功后，即可關(guān)閉

將上面三部分內(nèi)容合并到一個(gè)文件file_to_zabbix.conf中，然后啟動(dòng)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進(jìn)程的數(shù)據(jù)存儲目錄，用于在一個(gè)服務(wù)器上啟動(dòng)多個(gè)logstash進(jìn)程的環(huán)境中

4.2 zabbix平臺配置日志告警

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

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

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

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

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

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

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

下面我們模擬一個(gè)故障，在任意主機(jī)通過ssh登錄192.168.73.135主機(jī)，然后輸入一個(gè)錯(cuò)誤密碼，讓系統(tǒng)的/var/log/secure文件產(chǎn)生錯(cuò)誤日志，然后看看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平臺，點(diǎn)擊監(jiān)測中——->最新數(shù)據(jù)，如果zabbix能夠接收到日志，就可以看到下圖的最新數(shù)據(jù)：

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

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

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

如果接收到logstash發(fā)送過來的數(shù)據(jù)，就進(jìn)行告警，也就是說接收到的數(shù)據(jù)，如果長度大于0就告警 觸發(fā)器配置完成后，如果配置正常，就會進(jìn)行告警了，使用的是釘釘進(jìn)行圖文告警，告警內(nèi)容如下圖所示：

Kibana上查看安全日志

總結(jié)

首先我們來捋一下思路：

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

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

分享一個(gè)小技巧：通過該命令可以測試定義在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：指定鍵值