1 背景

ELK是三個開源軟體的縮寫，分別表示：Elasticsearch , Logstash, Kibana , 它們都是開源軟體。新增了一個FileBeat，它是一個輕量級的日誌收集處理工具(Agent)，Filebeat佔用資源少，適合於在各個伺服器上搜集日誌後傳輸給Logstash。

這是一個最簡單的架構圖：

filebeat ==> logstash ==> elasticsearch ==> kibana

• 資源準備

  下載部署包地址：https://www.elastic.co/cn/downloads/past-releases#elasticsearch

  注：elk全家桶各個應用版本需一致

• java版本

  java版本要求jdk1.8+，檢視版本：
  java -version

• 建立使用者

  elk需要用非root使用者部署，因此需要建立一個賬號用於部署
  useradd elk

  修改目錄的許可權，給elk賦權
  chown elk:elk /data/software/elk -R

2 logstash

logstash是一個開源的資料處理工具，可以同時從多個數據源收集資料，並對其進行特定格式轉換，再輸出到elasticsearch或其它資料儲存工具

logstash預設埠為9600

2.1 下載安裝

• 解壓安裝包到指定目錄/data/software/elk
  tar -vxf logstash-7.15.2-linux-x86_64.tar.gz -C /data/software/elk

2.2 配置檔案

• 在config目錄下建立配置檔案
  vim itcast-pipeline.conf

input {
    beats {
        port => 5044 # 輸入埠號
    }
}

filter {
  grok { # 可配置多個grok，按照順序進行匹配
        match =>{
          "message" => "%{MONTH:MONTH} %{MONTHDAY:MONTHDAY} %{TIME:TIME} %{MONTH:MONTH_log} %{MONTHDAY:MONTHDAY_log} %{YEAR:YEAR_log} %{TIME:TIME_log}%{ISO8601_TIMEZONE:ISO8601_TIMEZONE} %{WORD:system_name} %%%{INT:sys_version}%{EMAILLOCALPART:module}/%{INT:log_level}/%{NOTSPACE:summary_1}:\s*(?<summary_2>([\d|\w|\s]*)). \(Task=%{WORD:task}, Ip=%{NOTSPACE:Ip}, VpnName=%{NOTSPACE:VpnName}, User=%{NOTSPACE:User}, AuthenticationMethod=%{QUOTEDSTRING:AuthenticationMethod}, Command=%{QUOTEDSTRING:Command}"
          }

        match =>{
          "message" => "%{MONTH:MONTH} %{MONTHDAY:MONTHDAY} %{TIME:TIME} %{MONTH:MONTH_log} %{MONTHDAY:MONTHDAY_log} %{YEAR:YEAR_log} %{TIME:TIME_log}%{ISO8601_TIMEZONE:ISO8601_TIMEZONE} %{WORD:system_name} %%%{INT:sys_version}%{EMAILLOCALPART:module}/%{INT:log_level}/%{NOTSPACE:summary_1}:\s*(?<summary_2>([\d|\w|\s]*)). \(Task=%{WORD:task}, Ip=%{NOTSPACE:Ip}, VpnName=, User=%{NOTSPACE:User}, AuthenticationMethod=%{QUOTEDSTRING:AuthenticationMethod}, Command=%{QUOTEDSTRING:Command}"
          }

        match =>{
          "message" => "%{MONTH:MONTH} %{MONTHDAY:MONTHDAY} %{TIME:TIME} %{MONTH:MONTH_log} %{MONTHDAY:MONTHDAY_log} %{YEAR:YEAR_log} %{TIME:TIME_log}%{ISO8601_TIMEZONE:ISO8601_TIMEZONE} %{WORD:system_name} %%%{INT:sys_version}%{EMAILLOCALPART:module}/%{INT:log_level}/%{NOTSPACE:summary_1}:\s*(?<summary_2>([\d|\w|\s]*))."
          }
    }
}

output {
        elasticsearch {
                hosts => ["http://176.16.9.3:9200"]
                index => "xc-syslog-%{+YYYY-MM-dd}"
                user => "elastic"
                password => "ap20pOPS20"
    }
}

# output { # 除錯時可輸出到終端方便檢視處理後資料
#     stdout { codec => rubydebug }
#     }
 

2.3 grok

來自不同資料來源的資料，在logstash中讀取都是一行一行的文字資料，輸出的時候它們都是作為一行放到輸出的message欄位中，因此需要對message進行切割放入不同的欄位中，以便後續進行分析。在logstash中可以使用grok對資料進行處理

• 預設匹配規則

  • grok的語法格式

    %{SYNTAX:SEMANTIC}
    # SYNTAX是資料生成的欄位名稱，SEMANTIC是匹配出內容的規則（用正則表示式匹配出資料）
    

    logstash中資料提取需要使用正則表示式匹配，官方已預設一些匹配規則，詳細定義參見github 中的說明，以下為常用規則：

  表示式	名稱	例子
  USERNAME	使用者名稱（由數字、大小寫及特殊字元(._-)組成的字串）	1234、Bob、Alex.Wong
  EMAILLOCALPART	使用者名稱（首位由大小寫字母組成，其他位由數字、大小寫及特殊字元(_.+-=:)組成的字串）	windcoder、windcoder_com
  EMAILADDRESS	電子郵箱地址	[email protected]
  INT	整數（包括0和正負整數）	0、-123
  NUMBER	整數或者小數
  WORD	字串（包括數字和大小寫字母）	String、3529345
  NOTSPACE	不帶任何空格的字串
  SPACE	空格字串
  QUOTEDSTRING	帶引號的字串	"This is an apple"
  IP	IP地址（IPv4或IPv6地址）	127.0.0.1
  HOSTPORT	主機名(IP)+埠	127.0.0.1:3306
  PATH	路徑
  URIPROTO	URI協議	http、ftp
  URI	完整的URI
  MONTH	月份名稱（英文）	Jan、January
  MONTHNUM	月份數字	03、9、12
  MONTHDAY	日期數字	03、9、31
  DAY	星期幾名稱	Mon、Monday
  YEAR	年份數字	1986
  HOUR	小時數字
  MINUTE	分鐘數字
  SECOND	秒數字
  TIME	時間	00:01:23
  DATE_US	美國時間	10-01-1892、10/01/1892/
  DATE_EU	歐洲日期格式	01-10-1892、01/10/1882、01.10.1892
  ISO8601_TIMEZONE	ISO8601時間格式	+10:23、-1023
  TIMESTAMP_ISO8601	ISO8601時間戳格式	2016-07-03T00:34:06+08:00
  DATESTAMP	完整日期+時間	07-03-2016 00:34:06
  HTTPDATE	http預設日期格式	03/Jul/2016:00:36:53 +0800
  GREEDYDATA	匹配所有剩餘的資料

  • 示例

  日誌資訊為：

  Dec 21 02:31:10+08:00 MM_104_F15_XC_MGMT_CE5735_4
  

  匹配規則為：

  "%{MONTH:MONTH} %{MONTHDAY:MONTHDAY} %{TIME:TIME}{ISO8601_TIMEZONE:ISO8601_TIMEZONE} %{WORD:system_name}"
  

• 自定義正則表示式

  除了可以用上述的預設匹配規則之外，還可以自定義正則表示式

  • 語法格式

    (?<class_info>([\S+]*))
    # <class_info>資料生成的欄位名稱，[]裡面為自定義正則表示式
    
    

  • 示例

    日誌資訊為：

    %%01SHELL/5/CMDRECORD(s)[703]:Recorded command information. (Task=VT0, Ip=192.168.144.161, VpnName=, User=huaweiqyw3, AuthenticationMethod="Local-user", Command="system-view")
    

    匹配規則為：

    "%%%{INT:sys_version}%{EMAILLOCALPART:module}/%{INT:log_level}/%{NOTSPACE:summary_1}:\s*(?<summary_2>([\d|\w|\s]*)). \(Task=%{WORD:task}, Ip=%{NOTSPACE:Ip}, VpnName=%{NOTSPACE:VpnName}, User=%{NOTSPACE:User}, AuthenticationMethod=%{QUOTEDSTRING:AuthenticationMethod}, Command=%{QUOTEDSTRING:Command}"
    

2.4 服務部署啟用

在logstash目錄下啟用：
./bin/logstash -f ./config/itcast-pipeline.conf -l ./logstash.log --config.reload.automatic

-f 為指定配置檔案
-l為指定輸出日誌路徑
--config.reload.automatic為啟用自動配置載入，每次修改完配置檔案以後無需重啟Logstash

檢查服務：
ps -ef | grep logstash

2.5 報錯處理

2.5.1. Java路徑

報錯資訊為：could not find java; set JAVA_HOME or ensure java is in PATH

解決方案：在bin目錄下logstash.lib.sh檔案裡面新增 JAVACMD='java的啟用路徑'

儲存後重新啟動

本文來自部落格園，作者：liu_kx，轉載請註明原文連結：https://www.cnblogs.com/liukx/p/15715630.html