前言
在LogStash中對日志進行解析後再将日志傳輸到ElasticSearch中，最後通過Kibana檢視日志。
上文
EFK實戰一 - 基礎環境搭建已經搭建好了EFK的基礎環境，本文我們通過真實案例打通三者之間的資料傳輸以及解決EFK在使用過程中的一些常見問題。
首先看一下實際的業務日志
2020-01-09 10:03:26,719 INFO ========GetCostCenter Start===============
2020-01-09 10:03:44,267 WARN 成本中心編碼少于10位！{"deptId":"D000004345","companyCode":"01"}
2020-01-09 10:22:37,193 ERROR java.lang.IllegalStateException: SessionImpl[abcpI7fK-WYnW4nzXrv7w,]: can't call getAttribute() when session is no longer valid.
at com.caucho.server.session.SessionImpl.getAttribute(SessionImpl.java:283)
at weaver.filter.PFixFilter.doFilter(PFixFilter.java:73)
at com.caucho.server.dispatch.FilterFilterChain.doFilter(FilterFilterChain.java:87)
at weaver.filter.MonitorXFixIPFilter.doFilter(MonitorXFixIPFilter.java:30)
at weaver.filter.MonitorForbiddenUrlFilter.doFilter(MonitorForbiddenUrlFilter.java:133) 日志組成格式為：
時間 日志級别 日志詳情
主要任務就是将這段日志正常寫入EFK中。
filebeat安裝配置
- 下載下傳 filebeat7.5.1
- 将下載下傳後的檔案上傳至伺服器并解壓
tar -zxvf filebeat-7.5.1-linux-x86_64.tar.gz
- 修改filebeat.yml,
filebeat.inputs:
- type: log
enabled: true
paths:
- /app/weaver/Resin/log/xxx.log 此段配置日志輸入，指定日志存儲路徑
output.logstash:
# The Logstash hosts
hosts: ["172.31.0.207:5044"] 此段配置日志輸出，指定Logstash存儲路徑
- 啟動filebeat
./filebeat -e -c filebeat.yml
nohup ./filebeat -e -c filebeat.yml &
logstash配置
logstash的配置主要分為三段
input
，
filter
output
。
input
用于指定輸入，主要是開放端口給Filebeat用于接收日志
filter
用于指定過濾，對日志内容進行解析過濾。
output
用于指定輸出，直接配置ES的位址即可
input {
beats {
port => 5044
}
}
output {
elasticsearch {
hosts => ["http://172.31.0.127:9200"]
index => "myindex-%{+YYYY.MM.dd}"
user => "elastic"
password => "xxxxxx"
}
} 我們配置好logstash後通過指令重新開機logstash
docker-compose -f elk.yml restart logstash
經過上述兩步配置後應用程式往日志檔案寫入日志，filebeat會将日志寫入logstash。在kibana檢視寫入的日志結果如下：
日志顯示有2個問題：
- 由于錯誤日志堆棧資訊有多行，在kibana中展示成了多行，資料檢視很亂。需要将堆棧異常整理成一行顯示。
- 需要對日志進行解析，拆成“時間 日志級别 日志詳情”的顯示格式。
優化更新
-
在filebeat中設定合并行
filebeat預設是行傳輸的，但是我們的日志肯定是多行一個日志，我們要把多行合并到一起就要找到日志的規律。比如我們的日志格式全都是以時間格式開頭，是以我們在filebeat中
filebeat.inputs
# 以日期作為字首
multiline.pattern: ^\d{4}-\d{1,2}-\d{1,2}
# 開啟多行合并
multiline.negate: true
# 合并到上一行之後
multiline.match: after -
在logstash中設定對日志的解析
将日志解析成“時間 日志級别 日志詳情”的展示格式，是以我們需要在logstash配置檔案中添加filter段
filter {
grok{
match => {
"message" => "(?<date>\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}),\d{3} %{LOGLEVEL:loglevel} (?<des>.*)"
}
}
} 這裡主要是使用grok文法對日志進行解析，通過正規表達式對日志進行過濾。大家可以通過kibana裡的grok調試工具進行調試
配置完成後我們重新打開kibana Discover界面檢視日志，符合預期，完美！
常見問題
kibana 亂碼
這個主要原因還是用戶端日志檔案格式有問題，大家可以通過
file xxx.log
檢視日志檔案的編碼格式，如果是ISO8859的編碼基本都會亂碼，我們可以在filebeat配置檔案中通過encoding指定日志編碼進行傳輸。
filebeat.inputs:
- type: log
enabled: true
paths:
- /app/weaver/Resin/log/xxx.log
encoding: GB2312 kibana 提取字段出錯
如上所示，打開kibana Discover面闆時出現此異常，大家隻要删除ES中的
.kibana_1
索引然後重新通路Kibana即可。
檢視周圍檔案
我們在終端檢視日志某關鍵字時一般會查上下文資訊便于排查問題，如經常用到的指令
cat xxx.log | grep -C50 keyword
，那麼在Kibana中如何實作這功能呢。
在Kibana中搜尋關鍵字，然後找到具體日志記錄，點選左邊向下箭頭，然後再點選“檢視周圍文檔”即可實作。
動态索引
我們日志平台可能需要對接多個業務系統，需要根據業務系統建立不同的索引。
- 在filebeat中給日志打上标記
- type: log
......
fields:
logType: oabusiness - 在logstash中根據标記生成索引
input {
beats {
port => 5044
}
}
filter {
if [fields][logType] == "oabusiness" {
grok{
match => {
"message" => "(?<date>\d{4}-\d{2}-\d{2}\s\d{2}:\d{2}:\d{2}),\d{3} %{LOGLEVEL:loglevel} (?<des>.*)"
}
}
}
}
output {
elasticsearch {
hosts => ["http://172.31.0.207:9200"]
index => "%{[fields][logType]}-%{+YYYY.MM.dd}"
user => "elastic"
password => "elastic"
}
} 好了，各位朋友們，本期的内容到此就全部結束啦，能看到這裡的同學都是優秀的同學，下一個升職加薪的就是你了！
如果覺得這篇文章對你有所幫助的話請掃描下面二維碼加個關注。“轉發” 加 “在看”，養成好習慣！咱們下期再見！