Logstash 常用 filter 插件介紹(二)
date：日誌解析grok：正則匹配解析dissect：分割符解析mutate：對字段做處理，比如重命名、刪除、替換等json：按照 json 解析字段內容到指定字段中geoip：增加地理位置數據ruby： 利用 ruby 代碼來動態修改 Logstash Event
data
從字段解析日期以用作事件的Logstash時間戳，以下配置解析名為logdate的字段以設置Logstash時間戳：
filter { date { match => [ "logdate", "MMM dd yyyy HH:mm:ss" ] } }
返回結果：
{"logdate":"Jan 01 2018 12:02:03"}
{
"@version" => "1",
"host" => "Node2",
"@timestamp" => 2018-01-01T04:02:03.000Z,
"logdate" => "Jan 01 2018 12:02:03"
}說明：
match => [ "logdate", "MMM dd yyyy HH:mm:ss" ,"MMM d yyyy HH:mm:ss","ISO8601"]match：類型為數組，用於指定日期匹配的格式，可以一次指定多種日誌格式
target：類型為字符串，用於指定賦值的字段名，默認是@timestamp timezone：類型為字符串，用於指定時區
grok插件
將非結構化事件數據分析到字段中。 這個工具非常適用於系統日誌，Apache和其他網絡服務器日誌，MySQL日誌，以及通常為人類而不是計算機消耗的任何日誌格式。
55.3.244.1 GET /index.html 15824 0.043以下配置將消息解析為字段：
filter { grok { match => { "message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}" } } }
應用過濾器後，示例中的事件將具有以下字段：
client: 55.3.244.1
method: GET
request: /index.html
bytes: 15824
duration: 0.043Grok語法：
%{SYNTAX:SEMANTIC}：SYNTAX為grok pattern的名稱，SEMANTIC為賦值字段名稱%{NUMBER:duration}：可以匹配數值類型，但是grok匹配出的內容都是字符串類型，可以通過在最後指定為int或者float來強制轉換類型。
%{NUMBER:duration:float}
常見pattern可以查看：GitHub或者logstash家目錄下的：
vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns/grok-patterns我們也可以自定義規則，比如我們下面添加的nginx日誌的規則，放到上面的grok-patterns裏面：
# Nginx logs
NGUSERNAME [a-zA-Z\.\@\-\+_%]+
NGUSER %{NGUSERNAME}
NGINXACCESS %{IPORHOST:clientip} %{NGUSER:ident} %{NGUSER:auth} \[%{HTTPDATE:timestamp}\] "%{WORD:verb} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:response} (?:%{NUMBER:bytes}|-) (?:"(?:%{URI:referrer}|-)"|%{QS:referrer}) %{QS:agent}我們也可以在安裝的Kibana進行調試。
dissect 插件
基於分隔符原理解析數據，解決grok解析時消耗過多cpu資源的問題
使用分隔符將非結構化事件數據提取到字段中。 解剖過濾器不使用正則表達式，速度非常快。 但是，如果數據的結構因行而異，grok過濾器更合適。
dissect的應用有一定的局限性：主要適用於每行格式相似且分隔符明確簡單的場景
dissect語法比較簡單，有一系列字段(field)和分隔符(delimiter)組成
%{}字段
%{}之間是分隔符例如，假設日誌中包含以下消息：
Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool...以下配置解析消息：
filter {
dissect {
mapping => { "message" => "%{ts} %{+ts} %{+ts} %{src} %{prog}[%{pid}]: %{msg}" }
}
}解剖過濾器應用後，事件將被解剖到以下領域：
{
"msg" => "Starting system activity accounting tool...",
"@timestamp" => 2017-04-26T19:33:39.257Z,
"src" => "localhost",
"@version" => "1",
"host" => "localhost.localdomain",
"pid" => "1",
"message" => "Apr 26 12:20:02 localhost systemd[1]: Starting system activity accounting tool...",
"type" => "stdin",
"prog" => "systemd",
"ts" => "Apr 26 12:20:02"
}說明：
Apr 26 12:20:02
%{ts} %{+ts} %{+ts} #+代表該匹配值追加到ts字段下
{
"ts":"Apr 26 12:20:02"
}
two three one go
%{+order/2} %{+order/3} %{+order/1} %{+order/4} #/後面的數字代表拼接的次序
{
"order": "one two three go"
}
a=1&b=2
%{?key1}=%{&key1}&%{?key2}=%{&key2} #%{?}代表忽略匹配值，但是富裕字段名，用於後續匹配用；%{&}代表將匹配值賦予key1的匹配值
{
"a":"1",
"b":"2"
}
#dissect可以自動處理空的匹配值
John Smith,Big Oaks,Wood Lane,Hambledown,Canterbury,CB34RY
%{name},%{addr1},%{addr2},%{addr3},%{city},%{zip}
Jane Doe,4321 Fifth Avenue,,,New York,87432
{
"name":"Jane Doe",
"addr1":"4321 Fifth Avenue",
"addr2":"",
"addr3":"",
"city":"New York",
"zip":"87432"
}
#dissect分割後的字段值都是字符串，可以使用convert_datatype屬性進行類型轉換
filter{
dissect{
convert_datatype => {age => "int"}
}
}mutate 插件
使用最頻繁的操作，可以對字段進行各種操作，比如重命名、刪除、替換、更新等，主要操作如下：
convert #類型轉換
gsub #字符串替換
split/join/merge #字符串切割、數組合並為字符串、數組合並為數組
rename #字段重命名
update/replace #字段內容更新或替換
remove_field #刪除字段convert：實現字段類型的轉換，類型為hash,僅支持轉換為integer、float、string和Boolean
filter{
mutate{
convert => {"age" => "integer"}
}
}gsub：對字段內容進行替換，類型為數組，每3項為一個替換配置
filter {
mutate {
gsub => [
# replace all forward slashes with underscore
"fieldname", "/", "_",
# replace backslashes, question marks, hashes, and minuses
# with a dot "."
"fieldname2", "[\\?#-]", "."
]
}
} split：將字符串切割為數組
filter {
mutate {
split => { "fieldname" => "," }
}
} join：將數組拼接為字符串 merge：將兩個數組合並為1個數組，字符串會被轉為1個元素的數組進行操作 rename：字段重命名 update/replace：更新字段內容，區別在於update只在字段存在時生效，而replace在字段不存在時會執行新增字段的操作
filter {
mutate {
update => { "sample" => "My new message" }
update => { "message" => "source from c:%{source_host}" } #%{source_host}可以引用logstash Event中的字段值
}
} input {
stdin{type=>stdin}
}
filter{
dissect{ mapping => {"message" => "%{a}-%{b}-%{c}"} }
mutate{ replace => {"d" =>"source from c:%{c}"} }
}
output{
stdout{codec=>rubydebug}
}
hi-hello-123
{
"a" => "hi",
"b" => "hello",
"@timestamp" => 2018-06-29T02:01:24.473Z,
"c" => "123",
"d" => "source from c:123",
"@version" => "1",
"host" => "Node2",
"message" => "hi-hello-123",
"type" => "stdin"
}json 插件
將字段內容為json格式的數據進行解析
filter {
json {
source => "message" #要解析的字段名
target => "msg_json" #解析後的存儲字段，默認和message同級別
}
} geoip 插件
常用的插件，根據ip地址提供對應的地域信息，比如經緯度、城市名等，方便進行地理數據分析
filter {
geoip {
source => "clientip"
}
}ruby 插件
最靈活的插件，可以以ruby語言來隨心所欲的修改Logstash Event對象
filter{
ruby{
code => ‘size = event.get("message").size;
event.set("message_size",size)‘
}
}
ruby {
code => "event.set(‘@read_timestamp‘,event.get(‘@timestamp‘))"
}詳細介紹請查看官方文檔。
Logstash 常用 filter 插件介紹(二)