帶你了解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：指定鍵值