使用Logstash filter grok過濾日誌檔案
阿新 • • 發佈:2019-02-01
Logstash Filter Plugin Grok
Logstash提供了一系列filter過濾plugin來處理收集到的log event,根據log event的特徵去切分所需要的欄位,方便kibana做visualize和dashboard的data analysis。所有logstash支援的event切分外掛檢視這裡。下面我們主要講grok切分。
Grok基本介紹
- Grok 使用文字片段切分的方式來切分日誌事件,語法如下:
%{SYNTAX:SEMANTIC}
* `SYNTAX`代表匹配值的型別,例如,`0.11`可以`NUMBER`型別所匹配,`10.222.22.25`可以使用`IP` 匹配。
* `SEMANTIC`表示儲存該值的一個變數宣告,它會儲存在`elasticsearch`當中方便`kibana`做欄位搜尋和統計,你可以將一個`IP`定義為客戶端IP地址`client_ip_address`,eg:`%{IP:client_ip_address}`,所匹配到的值就會儲存到`client_ip_address`這個欄位裡邊,類似資料庫的列名,也可以把event log中的數字當成數字型別儲存在一個指定的變數當中,比如響應時間`http_response_time`,假設event log record如下:
55.3.244.1 GET /index.html 15824 0.043
可以使用如下grok pattern來匹配這種記錄
%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}
在logstash conf.d資料夾下面建立filter conf檔案,內容如下
# /etc/logstash/conf.d/01-filter.conf
filter {
grok {
match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time}" }
}
}
以下是filter結果
client_id_address: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
http_response_time: 0.043
grok內建的預設型別有很多種,檢視所有預設型別。
- 使用自定義型別
更多時候logstash grok沒辦法提供你所需要的匹配型別,這個時候我們可以使用自定義- 第一種,直接使用oniguruma語法去匹配文字片段,語法如下
假設你需要匹配的文字片段為一個長度為10或11的十六進位制的值,使用下列語法可以獲取該片段,並把值賦予(?<field_name>the pattern here)
queue_id
(?<queue_id>[0-9A-F]{10,11})
- 第二種,建立自定義pattern檔案
- 建立資料夾
patterns
,在此資料夾下面建立一個檔案,檔名隨意,eg:postfix
# contents of ./patterns/postfix: POSTFIX_QUEUEID [0-9A-F]{10,11}
- 然後將patterns file引入,告訴logstash你的自定義型別檔案,以下面的event log record為例子:
在logstash conf.d資料夾下面建立filter conf檔案,內容如下55.3.244.1 GET /index.html 15824 0.043 ABC24C98567
filter { grok { patterns_dir => ["./patterns"] match => { "message" => "%{IP:client_id_address} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:http_response_time} %{POSTFIX_QUEUEID:queue_id}" } } }
- 建立資料夾
- 第一種,直接使用oniguruma語法去匹配文字片段,語法如下
匹配結果如下:
client_id_address: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
http_response_time: 0.043
queue_id: ABC24C98567
推薦使用grokdebugger來寫匹配模式,輸入event log record,再逐步使用pattern微調切分,下方會根據你所寫的模式將輸入切分欄位。
- 其他常用內建方法
add_field:
當pattern匹配切分成功之後,可以動態的對某些欄位進行特定的修改或者新增新的欄位,使用%{fieldName}
來獲取欄位的值
Exmaple:
filter {
grok{
add_field => { "foo_%{somefield}" => "Hello world, %{somefield}" }
}
}
# You can also add multiple fields at once:
filter {
grok {
add_field => {
"foo_%{somefield}" => "Hello world, %{somefield}"
"new_field" => "new_static_value"
}
}
}
如果somefield=dad
,logstash會將foo_dad
新欄位加入elasticsearch
,並將值Hello world, dad
賦予該欄位
add_tag:
為經過filter或者匹配成功的event新增標籤
Example:
filter {
grok {
add_tag => [ "foo_%{somefield}" ]
}
}
# You can also add multiple tags at once:
filter {
grok {
add_tag => [ "foo_%{somefield}", "taggedy_tag"]
}
}
連結:https://www.jianshu.com/p/d46b911fb83e