ELK-使用nxlog+filebeat采集不同類型的日志輸出到logstash

文章目錄

• • ELK-使用nxlog+filebeat采集不同類型的日志輸出到logstash
• 前言
• 正常的采集配置
• 開啟多個端口監(jiān)聽
• 優(yōu)化nxlog（關(guān)鍵字放在開頭）
• 再優(yōu)化nxlog（關(guān)鍵字放在結(jié)尾）
• 使用filebeat

前言

網(wǎng)絡(luò)上有很例子給出一示例是采集一種類型的日志輸出到logstash，但一個系統(tǒng)上日志種類很多，同一個采集端能區(qū)分不同日志類型嗎？
下面的結(jié)構(gòu)是nxlog做客戶端采集，通過tcp協(xié)議發(fā)送到logstash，然后logstash傳輸?shù)絜lasticsearch。
前提條件必需：

安裝好elasticsearch,logstash,nxlog,filebeat

nxlog/filebeats安裝在windows系統(tǒng)，elasticsearch/logstash安裝在linux系統(tǒng)

正常的采集配置

nxlog配置nxlog.conf:

<Input in_donglilog>Module im_fileFile "D:\\jar\\dongli\\logs\\spring-boot.log"SavePos TRUE </Input><Output out_donglitcp>Module om_tcpHost 192.168.1.238Port 514 </Output><Route 1>Path in_donglilog => out_donglitcp </Route>

采集D:\jar\dongli\logs\spring-boot.log日志，輸出到192.168.1.238:514上

logstash配置：

input {tcp {port => 514type=>"plm"} } output{if [type] == "plm"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelian-%{+YYYY.MM.dd}"}} }

通過tcp協(xié)議監(jiān)控514端口，這時候logstash工作mode是server（另一個工作mode是client，用于采集并發(fā)送數(shù)據(jù)），是監(jiān)控514端口數(shù)據(jù)的。
網(wǎng)上例子多數(shù)止步于些，監(jiān)聽一個端口，然后接收數(shù)據(jù)，發(fā)送到elasticsearch。
假設(shè)，我們不僅僅監(jiān)控dongli的日志D:\jar\dongli\logs\spring-boot.log日志，我還監(jiān)聽另一個系統(tǒng)日志，假設(shè)應(yīng)用為kelian。這2個日志格式不一樣。nxlog配置相對簡單，主要是logstash怎么能區(qū)分接收的不同日志，在elasticsearch創(chuàng)建不同的index。總不能把2個應(yīng)用日志輸出到同一個index吧。

開啟多個端口監(jiān)聽

這個方法最簡單，不同應(yīng)用開啟不同端口監(jiān)控
nxlog配置

<Input in_donglilog>Module im_fileFile "D:\\jar\\dongli\\logs\\spring-boot.log"SavePos TRUE </Input><Output out_donglitcp>Module om_tcpHost 192.168.1.238Port 514 </Output><Route 1>Path in_donglilog => out_donglitcp </Route><Input in_kelianlog>Module im_fileFile "D:\\jar\\kelaien\\logs\\spring-boot.log"SavePos TRUE </Input><Output out_keliantcp>Module om_tcpHost 192.168.1.238Port 515 </Output> <Route 2>Path in_kelianlog => out_keliantcp </Route>

logstash配置：

input {tcp {port => 514type=>"dongli"}tcp {port => 515type=>"kelian"} } output{if [type] == "dongli"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "dongli-%{+YYYY.MM.dd}"}}if [type] == "kelian"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelian-%{+YYYY.MM.dd}"}} }

最簡單，但我不希望這么做，因?yàn)槊吭黾右粋€應(yīng)用就會增加一個端口，而增加一個端口就要增加這個端口對外開放，如果是阿里云ECS，還要修改安全組規(guī)則。個人覺得麻煩，但是這也不失為一種可選方式

優(yōu)化nxlog（關(guān)鍵字放在開頭）

要是能攜帶一個數(shù)據(jù)區(qū)分發(fā)送日志類型就好了。遺憾的是，nxlog并沒有提供這種選項(xiàng)，怎么辦？
修改傳輸?shù)臄?shù)據(jù)。
nxlog每讀一行發(fā)送到logstash，在每行日志前加個特殊的字符串，然后logstash截取這個字符串，根據(jù)這個特殊的字符串創(chuàng)建不同的index。

原理是logstash字符引用，只要是input輸入的值，都可以引用

nxlog配置：

<Input in_donglilog>Module im_fileFile "D:\\jar\\dongli\\logs\\spring-boot.log"SavePos TRUE </Input> <Input in_kelianlog>Module im_fileFile "D:\\jar\\kelaien\\logs\\spring-boot.log"SavePos TRUE </Input><Processor proc_donglilog>Module pm_transformerExec $raw_event = "dongli " + $raw_event; </Processor> <Processor proc_kelianlog>Module pm_transformerExec $raw_event = "kelian " + $raw_event; </Processor><Output out_donglitcp>Module om_tcpHost 192.168.1.238Port 514 </Output><Output out_keliantcp>Module om_tcpHost 192.168.1.238Port 514 </Output><Route 1>Path in_donglilog => proc_donglilog => out_donglitcp </Route><Route 2>Path in_kelianlog => proc_kelianlog => out_keliantcp </Route>

通過Processor模塊，在每行日志行都添加了應(yīng)用名。
logstash配置：

input {tcp {port => 514type=>"plm"} } filter{if [type] == "plm" {grok{match=>{"message" => "%{WORD:key} %{WORD}"}}mutate{gsub=>["message","%{key}",""]}} } output{if [type] == "plm"{if [key] == "dongli" { elasticsearch {hosts => ["127.0.0.1:9200"]index => "dongli-%{+YYYY.MM.dd}"}}if [key] == "kelian" { elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelian-%{+YYYY.MM.dd}"}}} }

處理的關(guān)鍵在過濾器代碼中

grok{match=>{#拿到應(yīng)用名"message" => "%{WORD:key} %{WORD}"}}mutate{#將message里應(yīng)用名替換為空gsub=>["message","%{key}",""]}

在output里就可以使用字段引用功能做判斷了

if [type] == "plm"{if [key] == "dongli" { }if [key] == "kelian" { }}

缺點(diǎn)
有個缺點(diǎn)，只對單行日志有作用，如果有多行合并為一行的異常日志就不適合，因?yàn)閷㈥P(guān)鍵字添加到行頭，破壞了數(shù)據(jù)完。
當(dāng)使用multiline插件過濾輸入數(shù)據(jù)時，行頭沒辦法區(qū)分,當(dāng)使用multiline插件，合并多行時會出問題。

codec => multiline{#以[開頭視為一行pattern => "^["negate => truewhat => "previous"}

再優(yōu)化nxlog（關(guān)鍵字放在結(jié)尾）

如果把關(guān)鍵字放在每行的結(jié)尾呢
nxlog配置：

<Input in_donglilog>Module im_fileFile "D:\\jar\\dongli\\logs\\spring-boot.log"SavePos TRUE </Input> <Input in_kelianlog>Module im_fileFile "D:\\jar\\kelaien\\logs\\spring-boot.log"SavePos TRUE </Input><Processor proc_donglilog>Module pm_transformerExec $raw_event = $raw_event + "(dongli)"; </Processor> <Processor proc_kelianlog>Module pm_transformerExec $raw_event = $raw_event + "(kelian)"; </Processor><Output out_donglitcp>Module om_tcpHost 192.168.1.238Port 514 </Output><Output out_keliantcp>Module om_tcpHost 192.168.1.238Port 514 </Output><Route 1>Path in_donglilog => proc_donglilog => out_donglitcp </Route><Route 2>Path in_kelianlog => proc_kelianlog => out_keliantcp </Route>

注意代碼

<Processor proc_donglilog>Module pm_transformerExec $raw_event = $raw_event + "(dongli)"; </Processor> <Processor proc_kelianlog>Module pm_transformerExec $raw_event = $raw_event + "(kelian)"; </Processor>

關(guān)鍵放在行尾并用小括號括起來。
logstash配置：

input {tcp {port => 514codec => multiline{pattern => "^\d{4}(\-|\/|.)\d{1,2}\1\d{1,2}"negate => truewhat => "previous"}type=>"plm"} } filter{if [type] == "plm" {grok{match=>{"message" => "(?<ckey>[(]\w+[)\\r])"}}mutate{gsub=>["message","[(]%{ckey}[)]",""]#gsub=>["ckey","\r",""]}} } output{if [type] == "plm"{if [ckey] == "(dongli)" { elasticsearch {hosts => ["127.0.0.1:9200"]index => "dongli-%{+YYYY.MM.dd}"}}if [ckey] == "(kelian)" { elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelian-%{+YYYY.MM.dd}"}}} }

主要的處理在過濾器里

if [type] == "plm" {grok{match=>{"message" => "(?<ckey>[(]\w+[)\\r])"}}mutate{gsub=>["message","[(]%{ckey}[)]",""]}}

拿到關(guān)鍵字，然后把message字段里關(guān)鍵刪除。
這種方式也不失為一種解決方案,但不絕不是優(yōu)雅的方案

使用filebeat

它天生可多攜帶關(guān)鍵字，而且在windows運(yùn)行也很穩(wěn)定，所以我推薦使用filebeat替代nxlog。
filebeat.yml配置：

filebeat.inputs: - type: logenabled: truepaths:- D:\jar\dongli\logs\spring-boot.loggfields:appname: dongli - type: logenabled: truepaths:- D:\jar\kelaien\logs\spring-boot.logfields:appname: kelaien

logstash配置

input{beats {port => 515type=>"beatss"} }output{if [fields][appname] == "dongli"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "dongli-%{+YYYY.MM.dd}"}}if [fields][appname] == "kelaien"{elasticsearch {hosts => ["127.0.0.1:9200"]index => "kelaien-%{+YYYY.MM.dd}"}} }

上面對應(yīng)的是單行日志，如果是多行日志，它的配置放在filebeats而不是logstash
filebeat.yml配置：

filebeat.inputs: - type: logenabled: truepaths:- D:\jar\dongli\logs\spring-boot.loggmultiline:pattern: '^\d{4}-\d{1,2}-\d{1,2}'negate: truematch: afterfields:appname: dongli - type: logenabled: truepaths:- D:\jar\kelaien\logs\spring-boot.logmultiline:pattern: '^\d{4}-\d{1,2}-\d{1,2}'negate: truematch: afterfields:appname: kelaien

多行的關(guān)鍵代碼是

multiline:pattern: '^\d{4}-\d{1,2}-\d{1,2}'negate: truematch: after

總結(jié)

以上是生活随笔為你收集整理的ELK-使用nxlog+filebeat采集不同类型的日志输出到logstash的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。