EFK實戰之如何實現日志集成

本篇內容主要講解“EFK實戰之如何實現日志集成”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“EFK實戰之如何實現日志集成”吧!

前言

在EFK基礎架構中，我們需要在客戶端部署Filebeat，通過Filebeat將日志收集并傳到LogStash中。在LogStash中對日志進行解析后再將日志傳輸到ElasticSearch中，最后通過Kibana查看日志。

上文已經搭建好了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">

這里主要是使用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">

到此，相信大家對“EFK實戰之如何實現日志集成”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！