Zabbix與ELK整合實現對安全日志數據的實時監控告警
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，值為oslogs
add_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字段中出現給出的這些關鍵字，那么就將這些信息發送給zabbix
zabbix {
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/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]
這里面有我們要過濾的關鍵字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 1
info from server: "processed: 1; failed: 0; total: 1; seconds spent: 0.000081"
sent: 1; skipped: 0; total: 1
詳解：-s：指定本地agent端
-z：指定zabbix服務端
-k：指定鍵值
