EFK實戰二 - 日誌集成

前言

在EFK基礎架構中，咱們須要在客戶端部署Filebeat，經過Filebeat將日誌收集並傳到LogStash中。在LogStash中對日誌進行解析後再將日誌傳輸到ElasticSearch中，最後經過Kibana查看日誌。java

上文已經搭建好了EFK的基礎環境，本文咱們經過真實案例打通三者之間的數據傳輸以及解決EFK在使用過程當中的一些常見問題。linux

首先看一下實際的業務日誌正則表達式

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中。docker

filebeat安裝配置

下載filebeat7.5.1服務器
將下載後的文件上傳至服務器並解壓tar -zxvf filebeat-7.5.1-linux-x86_64.tar.gz微信
修改filebeat.yml,session

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存儲路徑app

啓動filebeat
./filebeat -e -c filebeat.yml
若是須要靜默啓動，則使用 nohup ./filebeat -e -c filebeat.yml & 命令啓動便可。elasticsearch

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"
	}
}

若是本文對你有幫助，

別忘記來個三連：

點贊，轉發，評論

本文分享自微信公衆號 - JAVA日知錄（javadaily）。
若有侵權，請聯繫 support@oschina.cn 刪除。
本文參與「OSC源創計劃」，歡迎正在閱讀的你也加入，一塊兒分享。