帶你瞭解zabbix整合ELK收集系統異常日誌觸發告警~
https://www.cnblogs.com/bixiaoyu/p/9595698.html
今天來了解一下關於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告警機制觸發；
[[email protected] ~]# /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端】
[[email protected] ~]# 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"欄位輸出的就是伺服器上具體的日誌內容。
}
}
}
[[email protected] 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非零值表示失敗
[[email protected] 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：指定鍵值
好了，到這裡就結束了，喜歡我的部落格作品，想要學習更多技術，關注我吧，後面將會更新更多幹貨，如果對部落格文章有更好的建議或者技術疑問歡迎在下方評論區留言，也可以在左側找我的聯絡方式和QQ群組進行交流互動！ PS：技術文章為原創作品，未經允許，請勿轉載，如果轉發分享，請標明出處 分類: 監控平臺, 日誌分析, 自動化工具https://www.cnblogs.com/bixiaoyu/p/9595698.html
今天來了解一下關於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告警機制觸發；
[[email protected] ~]# /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端】
[[email protected] ~]# 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"欄位輸出的就是伺服器上具體的日誌內容。
}
}
}
[[email protected] 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非零值表示失敗
[[email protected] 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：指定鍵值