微信公眾號：運維開發故事，作者：double冬

1 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實時告警的功能了。

2 Logstash與zabbix插件的使用

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

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

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

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

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

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

列出所有已安裝的插件

[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 安裝插件

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

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

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

2.3 更新插件

每個插件有自己的發布周期和版本更新，這些更新通常是獨立于Logstash的發布周期的。因此，有時候需要單獨更新插件，可以使用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插件中刪除插件，可執行如下命令：

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

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

3 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監控項。

4 將logstash與zabbix進行整合

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

先說明一下我們的應用需求：通過對系統日志文件的監控，然后去過濾日志信息中的一些關鍵字，例如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:message_pid}\])?: %{GREEDYDATA:message_content}" }#這里通過grok對message字段的數據進行字段劃分，這里將message字段劃分了5個子字段。其中，message_content字段會在output中用到}mutate {add_field => ["[zabbix_key]","oslogs"] #新增的字段，字段名是zabbix_key，值為oslogsadd_field => ["[zabbix_host]","Zabbix server"] #新增的字段，字段名是zabbix_host，值可以在這里直接定義，也可以引用字段變量來獲取。這里的%{host}獲取的就是日志數據的主機名，這個主機名與zabbix web中“主機名稱”需要保持一致remove_field => ["@version","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 {elasticsearch{ #這部分是為把日志次發送到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字段中，需要過濾的關鍵字信息，也就是在message_content字段中出現給出的這些關鍵字，那么就將這些信息發送給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的監聽端口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 /tmp/ &

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

4.2 zabbix平臺配置日志告警

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

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

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

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

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

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

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

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

[root@k8s-node03 ~]# tail /var/log/secureDec 25 22:57:44 k8s-node03 sshd[49803]: Failed password for root from 192.168.73.1 port 60256 ssh2Dec 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=rootDec 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 ssh2Dec 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]

這里面有我們要過濾的關鍵字Failed，因此logstash會將此內容過濾出來，發送到zabbix上
接著，登錄zabbix web平臺，點擊監測中——->最新數據，如果zabbix能夠接收到日志，就可以看到下圖的最新數據：

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

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

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

這里注意觸發器創建中，表達式的寫法，這里觸發器的觸發條件是：

如果接收到logstash發送過來的數據，就進行告警，也就是說接收到的數據，如果長度大于0就告警

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

Kibana上查看安全日志

總結

首先我們來捋一下思路：

我們的架構基本不變，仍然是filebat收集日志推送到kibana消息隊列，然后由Logstash前去拉取日志數據，經過處理最后中轉出去；只不過是中轉輸出到zabbix上面而已；能夠實現這個功能的，最核心的功臣就是Logsatsh的插件（logstash-output-zabbix）；

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

分享一個小技巧：通過該命令可以測試定義在zabbix上的鍵值；出現以下輸出變為正常~，如果failed非零值表示失敗

#在server端使用zabbix_sender向zabbix發送測試[root@localhost zabbix_sender]# /usr/local/zabbix/bin/zabbix_sender -s 192.168.73.135 -z 192.168.73.133 -k "oslogs" -o 1info from server: "processed: 1; failed: 0; total: 1; seconds spent: 0.000081"sent: 1; skipped: 0; total: 1

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

-z：指定zabbix服務端

-k：指定鍵值

總結

以上是生活随笔為你收集整理的Zabbix与ELK整合实现对安全日志数据的实时监控告警的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。