帶你瞭解zabbix整合ELK收集系統異常日誌觸發告警~

今天來了解一下關於ELK的「L」-Logstash,沒錯，就是這個神奇小組件，咱們都知道，它是ELK不可缺乏的組件，完成了輸入（input），過濾（fileter），output（輸出）工做量，也是咱們做爲運維人員須要掌握的難點，說到這裏 ，又愛又恨；「愛之好，恨之難」；這個Logstash擁有這強大的插件功能，除了幫咱們過濾，高效的輸出日誌，還能幫咱們與Zabbix監控相結合？

由於咱們的Logstash支持多種輸出類型，可以收集web服務日誌，系統日誌，內核日誌；可是；居然是有日誌輸出，確定避免不了錯誤（error）日誌的出現；當Error日誌出現的時候，雖然能夠經過ELK查找出來，可是ELK不能實時提供報警，這就有點尷尬了，咱們要作的就是可以像zabbix，nagios監控那樣，不能要作到監控，還要作到報警，這一點，ELK只作到了監控，可是沒有作到報警；不過不要緊，咱們的Logstash插件可以與zabbix結合起來，將須要告警 的日誌收集起來（好比說有錯誤標識的日誌）完成日誌監控觸發告警~

logstash支持多種輸出介質，好比說syslog，http，tcp，elasticsearch,kafka等，若是咱們將logstash收集的日誌輸出到zabbix告警，就必需要用到logstash-output-zabbix插件，經過這個插件將logstash與zabbix整合，logstash收集到的數據過濾出錯誤信息的日誌輸出到zabbix中，最後經過zabbix告警機制觸發；

[root@localhost ~]# /usr/local/logstash/bin/logstash-plugin install logstash-output-zabbix      #安裝logstash-output-zabbix插件
Validating logstash-output-zabbix
Installing logstash-output-zabbix
Installation successful

環境案例需求：

經過讀系統日誌文件的監控，過濾掉日誌信息中的異常關鍵詞，如ERR,error，Failed，warning等信息，將這些帶有異常關鍵詞的異常日誌信息過濾出來，而後輸出到zabbix，經過zabbix告警機制實現觸發告警；下面環境是filebeat做爲採集端；輸出到kafaka消息隊列，最後由logsatsh拉取日誌並過濾，輸出到zabbix

【filebeat】日誌採集端

filebeat.inputs:
- type: log
  enabled: true
  paths:
   - /var/log/secure
   - /var/log/messages
   - /var/log/cron
  fields:
    log_topic: system_log
processors:
 - drop_fields:
    fields: ["beat", "input", "source", "offset", "prospector"] #這裏在filebeat中直接去掉不須要的字段。
filebeat.config.modules:
  path: ${path.config}/modules.d/*.yml
  reload.enabled: false
name: 192.168.37.147        #這是日誌輸出標識，代表日誌來自哪一個主機，後面再logstash會用到。
output.kafka:
  enabled: true
  hosts: ["192.168.37.147:9092", "192.168.37.148:9092", "192.168.37.149:9092"]    #日誌輸出到kafka集羣
  version: "0.10"
  topic: '%{[fields.log_topic]}' 
  partition.round_robin:
    reachable_only: true
  worker: 2
  required_acks: 1
  compression: gzip
  max_message_bytes: 10000000
logging.level: debug

 

【Logstash端】

 

[root@localhost ~]# vim /usr/local/logstash/config/etc/system_log.conf

input {
        kafka {
        bootstrap_servers => "192.168.37.147:9092,192.168.37.148:9092,192.168.37.149:9092"
        topics => ["system_log"]
        codec => "json"
        }
}

filter {
    if [fields][log_topic] == "system_log" {    #指定filebeat產生的日誌主題
             mutate {
             add_field => [ "[zabbix_key]", "oslogs" ]      #新增的字段，字段名是zabbix_key，值爲oslogs。
             add_field => [ "[zabbix_host]", "%{[host][name]}" ]   #新增的字段，字段名是zabbix_host，值能夠在這裏直接定義，也能夠引用字段變量來獲取。這裏的%{[host][name]獲取的就是日誌數據的來源IP，這個來源IP在filebeat配置中的name選項進行定義。
             }
         }
    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 {        #這裏是刪除不須要的字段
            remove_field => "@version"
            remove_field => "message"
        }
        date {      #這裏是對日誌輸出中的日期字段進行轉換，其中message_timestamp字段是默認輸出的時間日期字段，將這個字段的值傳給 @timestamp字段。
                match => [ "message_timestamp","MMM  d HH:mm:ss", "MMM dd HH:mm:ss", "ISO8601"]
        }
}

output {
        if [message_content]  =~ /(ERR|error|ERROR|Failed)/  {   #定義在message_content字段中，須要過濾的關鍵字信息，也就是在message_content字段中出現給出的這些關鍵字，那麼就將這些信息發送給zabbix。
              zabbix {
                        zabbix_host => "[zabbix_host]"      #這個zabbix_host將獲取上面filter部分定義的字段變量%{[host][name]的值
                        zabbix_key => "[zabbix_key]"        #這個zabbix_key將獲取上面filter部分中給出的值
                        zabbix_server_host => "192.168.37.149"  #這是指定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"字段輸出的就是服務器上具體的日誌內容。
                        }
                    }
}

 

[root@localhost logstash]# nohup /usr/local/logstash/bin/logstash -f config/etc/system_log.conf --path.data=/tmp/          #這裏的--path.data指定詞logstash進程的數據存儲目錄，用於在一個服務器上啓動多個logstash進程環境

[測試]不肯定事件配置文件是否正確，咱們能夠經過前臺運行，輸出stdout；驗證filebeat收集的日誌是夠成功的過濾~

stdout {codec => rubydebug}    #咱們將這條指令加入output輸出端，前臺運行測試，看是夠可以過濾出來錯誤日誌輸出，效果以下~（ok以後記得將這條指令註釋掉並後臺運行哦）

# /usr/local/logstash/bin/logstash -f config/etc/system_log.conf --path.data=/tmp/

【zabbix-監控模板建立到 告警一觸即發】

1.建立模板

將詞模板連接到192.168.37.147上，建立的模板上的監控項就會在192.168.37.147上自動生效了，

 

2.建立應用集，點擊應用集-建立應用集

3.建立監控項，點擊監控項，建立監控項

4.告警觸發，建立 觸發器

 

將我們建立的收集日誌的模板鏈接到 須要收集日誌的主機，驗證告警觸發效果

【模擬告警】

ssh鏈接192.168.37.147日誌收集主機，故意輸錯密碼，讓系統產生錯誤日誌，驗證是夠發送到zabbix端，下面即是咱們過濾的錯誤日誌信息，如「error」，"Failed"等~到目前爲止，已經成功的收集到錯誤日誌輸出了~

 【總結】

首先咱們來捋一下思路：

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

在這裏須要注意的是:filebeat收集端的IP必定要與zabbix監控主機的IP相對應，不然日誌是過不來的~

分享一個小技巧：經過該命令能夠測試定義在zabbix上的鍵值；出現如下輸出變爲正常~，若是failed非零值表示失敗

[root@localhost zabbix_sender]# /usr/local/zabbix/bin/zabbix_sender -s 192.168.37.147 -z 192.168.37.149 -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：指定鍵值