EFK實戰二 - 日誌集成

前言

在EFK基礎架構中，咱們須要在客戶端部署Filebeat，經過Filebeat將日誌收集並傳到LogStash中。在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"
    }
}複製代碼

好了，各位朋友們，本期的內容到此就所有結束啦，能看到這裏的同窗都是優秀的同窗，下一個升職加薪的就是你了！若是以爲這篇文章對你有所幫助的話請掃描下面二維碼加個關注。"轉發" 加 "在看"，養成好習慣！我們下期再見！

歡迎掃碼關注微信公衆號或 我的博客