今天來了解一下關于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: logenabled: truepaths:- /var/log/secure- /var/log/messages- /var/log/cronfields:log_topic: system_log processors:- drop_fields:fields: ["beat", "input", "source", "offset", "prospector"] #這里在filebeat中直接去掉不需要的字段。 filebeat.config.modules:path: ${path.config}/modules.d/*.ymlreload.enabled: false name: 192.168.37.147 #這是日志輸出標識，表明日志來自哪個主機，后面再logstash會用到。 output.kafka:enabled: truehosts: ["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: trueworker: 2required_acks: 1compression: gzipmax_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：指定鍵值
