Zabbix與ELK整合實現對日志數據的實時監控

• 4.2、zabbix平台配置日志告警

一、 ELK與zabbix有什么關系？

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

ELK是一套日志收集套件，它其實有由Elasticsearch、Logstash和Kibana三個軟件組成，通過ELK可以收集系統日志、網站日志、應用系統日志等各種日志數據，並且還可以對日志進行過濾、清洗，然后進行集中存放並可用於實時檢索、分析。這是ELK的基礎功能。

但是有些時候，我們希望在收集日志的時候，能夠將日志中的異常信息（警告、錯誤、失敗等信息）及時的提取出來，因為日志中的異常信息意味着操作系統、應用程序可能存在故障，如果能將日志中的故障信息及時的告知運維人員，那么運維就可以第一時間去進行故障排查和處理，進而也就可以避免很多故障的發生。

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

二、Logstash與zabbix插件的使用

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

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

/usr/local/logstash/bin/logstash-plugin install logstash-output-zabbix

其中，/usr/local/logstash是Logstash的安裝目錄。

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

2.1、列出目前已經安裝的插件

將列出所有已安裝的插件

/usr/local/logstash/bin/logstash-plugin list #將列出已安裝的插件及版本信息 /usr/local/logstash/bin/logstash-plugin list --verbose #將列出包含namefragment的所有已安裝插件 /usr/local/logstash/bin/logstash-plugin list "http" #將列出特定組的所有已安裝插件（ input，filter，codec，output） /usr/local/logstash/bin/logstash-plugin list --group input

2.2、安裝插件

要安裝某個插件，例如安裝kafka插件，可執行如下命令：

/usr/local/logstash/bin/logstash-plugin install logstash-output-kafka
要使用此命令安裝插件，需要你的電腦可以訪問互聯網。此插件安裝方法，會檢索托管在公共存儲庫（RubyGems.org）上的插件，然后下載到本地機器並在Logstash安裝之上進行自動安裝。

2.3、更新插件

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

將更新所有已安裝的插件

/usr/local/logstash/bin/logstash-plugin update

將僅更新指定的插件

/usr/local/logstash/bin/logstash-plugin update logstash-output-kafka

2.4、刪除插件

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

/usr/local/logstash/bin/logstash-plugin remove logstash-output-kafka

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

三、logstash-output-zabbix插件的使用

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 字段的子字段， 是必需的設置，沒有默認值。

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

zabbix_server_host：表示Zabbix服務器的IP或可解析主機名，默認值是 "localhost"，需要設置為zabbix server服務器所在的地址。 zabbix_server_port：表示Zabbix服務器開啟的監聽端口，默認值是10051。 zabbix_value：表示要發送給zabbix item監控項的值對應的字段名稱，默認值是 "message"，也就是將"message"字段的內容發送給上面zabbix_key定義的zabbix item監控項，當然也可以指定一個具體的字段內容發送給zabbix item監控項。

四、將logstash與zabbix進行整合

這里我們以logstash收集日志，然后對日志進行讀取，最后選擇關鍵字進行過濾並調用zabbix告警的流程，來看看如何配置logstash實現zabbix告警。

先說明一下我們的應用需求：通過對系統日志文件的監控，然后去過濾日志信息中的一些關鍵字，例如ERR、error、ERROR、Failed、WARNING等，將日志中這些信息過濾出來，然后發送到zabbix上，最后借助zabbix的報警功能實現對系統日志中有上述關鍵字的告警。

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

4.1、配置logstash事件配置文件

接下來就是創建一個logstash事件配置文件，這里將配置文件分成三個部分來介紹，首先是input部分，內容如下：

input {
        file {
        path => ["/var/log/secure"] type => "system" start_position => "beginning" } }

input部分是從/var/log/secure文件中讀取數據，start_position 表示從secure文件開頭讀取內容。

接着是filter部分，內容如下：

filter {
    grok {
             match => { "message" => "%{SYSLOGTIMESTAMP:message_timestamp} %{SYSLOGHOST:hostname} %{DATA:message_program}(?:\[%{POSINT:messag e_pid}\])?: %{GREEDYDATA:message_content}" } #這里通過grok對message字段的數據進行字段划分，這里將message字段划分了5個子字段。其中，message_content字段會在output中用到。 } mutate { add_field => [ "[zabbix_key]", "oslogs" ] #新增的字段，字段名是zabbix_key，值為oslogs。 add_field => [ "[zabbix_host]", "%{host}" ] #新增的字段，字段名是zabbix_host，值可以在這里直接定義，也可以引用字段變量來獲取。這里的%{host}獲取的就是日志數據的主機名，這個主機名與zabbix web中“主機名稱”需要保持一致。 } mutate { #這里是刪除不需要的字段 remove_field => "@version" remove_field => "message" } date { #這里是對日志輸出中的日期字段進行轉換，其中message_timestamp字段是默認輸出的時間日期字段，將這個字段的值傳給 @timestamp字段。 match => [ "message_timestamp","MMM d HH:mm:ss", "MMM dd HH:mm:ss", "ISO8601"] } }

filter部分是個重點，在這個部分中，重點關注的是message_timestamp字段、message_content字段。

最后是output部分，內容如下：

output {
        if [message_content] =~ /(ERR|error|ERROR|Failed)/ { #定義在message_content字段中，需要過濾的關鍵字信息，也就是在message_content字段中出現給出的這些關鍵字，那么就將這些信息發送給zabbix。 zabbix { zabbix_host => "[zabbix_host]" #這個zabbix_host將獲取上面filter部分定義的字段變量%{host}的值 zabbix_key => "[zabbix_key]" #這個zabbix_key將獲取上面filter部分中給出的值 zabbix_server_host => "172.16.213.140" #這是指定zabbix server的IP地址 zabbix_server_port => "10051" #這是指定zabbix server的監聽端口 zabbix_value => "message_content" #這個很重要，指定要傳給zabbix監控項item（oslogs）的值， zabbix_value默認的值是"message"字段，因為上面我們已經刪除了"message"字段，因此，這里需要重新指定，根據上面filter部分對"message"字段的內容划分，這里指定為"message_content"字段，其實，"message_content"字段輸出的就是服務器上具體的日志內容。 } } #stdout { codec => rubydebug } #這里是開啟調試模式，當第一次配置的時候，建議開啟，這樣過濾后的日志信息直接輸出的屏幕，方便進行調試，調試成功后，即可關閉。 }

將上面三部分內容合並到一個文件file_to_zabbix.conf中，然后啟動logstash服務：

[root@logstashserver ~]#cd /usr/local/logstash [root@logstashserver logstash]#nohup bin/logstash -f config/file_to_zabbix.conf --path.data /data/osdata &

這里的–path.data是指定此logstash進程的數據存儲目錄，用於在一個服務器上啟動多個logstash進程的環境中。

4.2、zabbix平台配置日志告警

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

接着，在此模塊下創建一個應用集，點擊應用集—–>創建應用集，如下圖所示：

然后，在此模塊下創建一個監控項，點擊監控項—–>創建監控項，如下圖所示：

到此為止，zabbix監控logstash的日志數據配置完成。

這里我們以客戶端172.16.213.157主機為例，也就是監控172.16.213.157主機上的系統日志數據，當發現日志異常就進行告警。

在上面創建了一個模板和監控項，接着還需要將此模板鏈接到172.16.213.157主機上，選擇“配置”—“主機”，然后選中172.16.213.157主機，選擇“模板”標簽，將上面創建的模板加入到172.16.213.157主機上，如下圖所示：

這樣，上面創建的監控項在172.16.213.157主機上就自動生效了。

下面我們模擬一個故障，在任意主機通過ssh登錄172.16.213.157主機，然后輸入一個錯誤密碼，讓系統的/var/log/secure文件產生錯誤日志，然后看看logstash是否能夠過濾到，是否能夠發送到zabbix中。

首先讓系統文件/var/log/secure產生類似如下內容：

Sep  5 16:01:04 localhost sshd[27159]: pam_unix(sshd:auth): authentication failure; logname= uid=0 euid=0 tty=ssh ruser= rhost=172.16.213.127 user=root Sep 5 16:01:06 localhost sshd[27159]: Failed password for root from 172.16.213.127 port 59913 ssh2

這里面有我們要過濾的關鍵字Failed，因此logstash會將此內容過濾出來，發送到zabbix上。

接着，登錄zabbix web平台，點擊監測中—–>最新數據，如果zabbix能夠接收到日志，就可以看到下圖的最新數據：

點擊歷史記錄，可以查看詳細內容，如下圖所示：

可以看到，紅框中的內容就是在logstash中定義的message_content字段的內容。

到這里為止，zabbix已經可以收到logstash的發送過來的數據了，但是要實現報警，還需要在zabbix中創建一個觸發器，進入配置—–>模板，選擇logstash-output-zabbix這個模板，然后點擊上面的觸發器，繼續點擊右上角的創建觸發器，如下圖所示：

這里注意觸發器創建中，表達式的寫法，這里觸發器的觸發條件是：如果接收到logstash發送過來的數據，就進行告警，也就是說接收到的數據，如果長度大於0就告警。

觸發器配置完成后，如果配置正常，就會進行告警了，告警內容如下圖所示：