實戰ELK(5) Logstash 入門
Logstash 是一個開源的資料收集引擎,它具有備實時資料傳輸能力。它可以統一過濾來自不同源的資料,並按照開發者的制定的規範輸出到目的地。
一、原理
Logstash 通過管道進行運作,管道有兩個必需的元素,輸入和輸出,還有一個可選的元素,過濾器。
輸入外掛從資料來源獲取資料,過濾器外掛根據使用者指定的資料格式修改資料,輸出外掛則將資料寫入到目的地。如下圖:
輸入:採集各種樣式、大小和來源的資料
資料往往以各種各樣的形式,或分散或集中地存在於很多系統中。Logstash 支援各種輸入選擇 ,可以在同一時間從眾多常用來源捕捉事件。能夠以連續的流式傳輸方式,輕鬆地從您的日誌、指標、Web 應用、資料儲存以及各種 AWS 服務採集資料。
過濾器:實時解析和轉換資料
資料從源傳輸到儲存庫的過程中,Logstash 過濾器能夠解析各個事件,識別已命名的欄位以構建結構,並將它們轉換成通用格式,以便更輕鬆、更快速地分析和實現商業價值。
Logstash 能夠動態地轉換和解析資料,不受格式或複雜度的影響:
- 利用 Grok 從非結構化資料中派生出結構
- 從 IP 地址破譯出地理座標
- 將 PII 資料匿名化,完全排除敏感欄位
- 整體處理不受資料來源、格式或架構的影響
輸出:選擇你的儲存,匯出你的資料
儘管 Elasticsearch 是我們的首選輸出方向,能夠為我們的搜尋和分析帶來無限可能,但它並非唯一選擇。
Logstash 提供眾多輸出選擇,您可以將資料傳送到您要指定的地方,並且能夠靈活地解鎖眾多下游用例。
二、安裝
2.1、下載
Logstash 依賴 JDK1.8 ,因此在安裝之前請確保機器已經安裝和配置好 JDK1.8。
Logstash 下載:https://artifacts.elastic.co/downloads/logstash/logstash-6.5.1.rpm
2.2、安裝
yum install logstash-6.5.1.rpm
檢視啟動檔案來了解下檔案目錄
vim /etc/systemd/system/logstash.service
常用的目錄:
/etc/default/logstash /etc/sysconfig/logstash /usr/share/logstash/bin/logstash
/etc/logstash
2.3、設定執行許可權
- 方法1,修改
logstash.service
vim /etc/systemd/system/logstash.service User=root #直接以root賬戶允許 Group=root
- 方法2
chown -R logstash:logstash /var/lib/logstash chown -R logstash:logstash /var/log/logstash
2.4 、開機啟動
systemctl start logstash.service
systemctl enable logstash.service
systemctl status logstash.service
2.5、我們先來一個簡單的案例:
cd /usr/share/logstash bin/logstash -e 'input { stdin { } } output { stdout {} }'
可能會有點慢
這時候輸入個Hello World
在生產環境中,Logstash 的管道要複雜很多,可能需要配置多個輸入、過濾器和輸出外掛。
因此,需要一個配置檔案管理輸入、過濾器和輸出相關的配置。配置檔案內容格式如下:
# 輸入
input {
...
}
# 過濾器
filter {
...
}
# 輸出
output {
...
}
根據自己的需求在對應的位置配置 輸入外掛、過濾器外掛、輸出外掛 和 編碼解碼外掛 即可。
三、外掛用法
Logstash 每讀取一次資料的行為叫做事件。
在 Logstach 目錄中建立一個配置檔案,名為 logstash.conf(名字任意)。
cd /usr/share/logstash
3.1 輸入外掛
輸入外掛允許一個特定的事件源可以讀取到 Logstash 管道中,配置在 input {} 中,且可以設定多個。
修改配置檔案:
input { # 從檔案讀取日誌資訊 file { path => "/var/log/elasticsearch/" type => "elasticsearch" start_position => "beginning" } } # filter { # # } output { # 標準輸出 stdout { codec => rubydebug } }
其中,messages 為系統日誌。
儲存檔案。鍵入:
bin/logstash -f logstash.conf
3.2 輸出外掛
輸出外掛將事件資料傳送到特定的目的地,配置在 output {} 中,且可以設定多個。
修改配置檔案:
input { # 從檔案讀取日誌資訊 file { path => "/var/log/error.log" type => "error" start_position => "beginning" } } # filter { # # } output { # 輸出到 elasticsearch elasticsearch { hosts => ["192.168.50.70:9200"] index => "error-%{+YYYY.MM.dd}" } }
儲存檔案。鍵入:
bin/logstash -f logstash.conf
3.3 編碼解碼外掛
編碼解碼外掛本質是一種流過濾器,配合輸入外掛或輸出外掛使用。
從上圖中,我們發現一個問題:Java 異常日誌被拆分成單行事件記錄到 Elasticsearch 中,這不符合開發者或運維人員的檢視習慣。因此,我們需要對日誌資訊進行編碼將多行事件轉成單行事件記錄起來。
我們需要配置 Multiline codec 外掛,這個外掛可以將多行日誌資訊合併成一行,作為一個事件處理。
Logstash 預設沒有安裝該外掛,需要開發者自行安裝。鍵入:
bin/logstash-plugin install logstash-codec-multiline
修改配置檔案:
input { # 從檔案讀取日誌資訊 file { path => "/var/log/error.log" type => "error" start_position => "beginning" # 使用 multiline 外掛 codec => multiline { # 通過正則表示式匹配,具體配置根據自身實際情況而定 pattern => "^\d" negate => true what => "previous" } } } # filter { # # } output { # 輸出到 elasticsearch elasticsearch { hosts => ["192.168.2.41:9200"] index => "error-%{+YYYY.MM.dd}" } }
儲存檔案。鍵入:
bin/logstash -f logstash.conf
4.4 過濾器外掛
過濾器外掛位於 Logstash 管道的中間位置,對事件執行過濾處理,配置在 filter {},且可以配置多個。
本次測試使用 grok 外掛演示,grok 外掛用於過濾雜亂的內容,將其結構化,增加可讀性。
安裝:
bin/logstash-plugin install logstash-filter-grok
修改配置檔案:
input { stdin {} } filter { grok { match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER :duration}" } } } output { stdout { codec => "rubydebug" } }
儲存檔案。鍵入:
bin/logstash -f logstash.conf
啟動成功後,我們輸入:
55.3.244.1 GET /index.html 15824 0.043
控制檯返回:
[[email protected] logstash-5.6.3]# bin/logstash -f logstash.conf Sending Logstash's logs to /root/logstash-5.6.3/logs which is now configured via log4j2.properties [2017-10-30T08:23:20,456][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"fb_apache", :directory=>"/root/logstash-5.6.3/modules/fb_apache/configuration"} [2017-10-30T08:23:20,459][INFO ][logstash.modules.scaffold] Initializing module {:module_name=>"netflow", :directory=>"/root/logstash-5.6.3/modules/netflow/configuration"} [2017-10-30T08:23:21,447][INFO ][logstash.pipeline ] Starting pipeline {"id"=>"main", "pipeline.workers"=>1, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>5, "pipeline.max_inflight"=>125} The stdin plugin is now waiting for input: [2017-10-30T08:23:21,516][INFO ][logstash.pipeline ] Pipeline main started [2017-10-30T08:23:21,573][INFO ][logstash.agent ] Successfully started Logstash API endpoint {:port=>9600} 55.3.244.1 GET /index.html 15824 0.043 { "duration" => "0.043", "request" => "/index.html", "@timestamp" => 2017-10-30T15:23:23.912Z, "method" => "GET", "bytes" => "15824", "@version" => "1", "host" => "localhost.localdomain", "client" => "55.3.244.1", "message" => "55.3.244.1 GET /index.html 15824 0.043" }
輸入的內容被匹配到相應的名字中。