豐富的過濾器插件是 logstash威力如此強大的重要因素，過濾器插件主要處理流經當前Logstash的事件信息，可以添加字段、移除字段、轉換字段類型，通過正則表達式切分數據等，也可以根據條件判斷來進行不同的數據處理方式。 [TOC] ## A grok正則捕獲插件 [Logstash簡單介紹](https://blog.csdn.net/chenleiking/article/details/73563930) [grokdebug在線調試grok](http://grokdebug.herokuapp.com/) [grok預裝正則表達式](https://www.cnblogs.com/stozen/p/5638369.html) [github上的預裝正則表達式](https://github.com/logstash-plugins/logstash-patterns-core/tree/master/patterns) grok插件是一個十分耗費資源的插件,是Logstash中將非結構化數據解析成結構化數據以便于查詢的最好工具，非常適合解析syslog logs，apache log， mysql log，以及一些其他的web log ### **預定義表達式調用** Logstash提供120個常用正則表達式可供安裝使用，安裝之后你可以通過名稱調用它們，語法如下： ```sh %{SYNTAX:SEMANTIC} SYNTAX：表示已經安裝的正則表達式的名稱 SEMANTIC：表示給Event中匹配到的內容指定什么名稱 ``` >例如：Event的內容為`“[debug] 127.0.0.1 - test log content”`，匹配`%{IP:client}`將獲得`“client: 127.0.0.1”`的結果； 如果想對捕獲的數據進行數據類型轉換,可以使用`%{NUMBER:num:int}`這種語法 默認返回結果都是string類型，且當前Logstash所支持的轉換類型僅有“int”和“float”； ### **grok匹配事例：** ```sh #日志內容：55.3.244.1 GET /index.html 15824 0.043 filter { grok { match => {"message" => "%{IP:client} %{WORD:method} %{URIPATHPARAM:request} %{NUMBER:bytes} %{NUMBER:duration}"} } } ``` **?輸出結果：** ~~~sh client: 55.3.244.1 method: GET request: /index.html bytes: 15824 duration: 0.043 ~~~ ### 自定義表達式 * 自定義表達式語法 語法：`(?<field_name>the_pattern)` 舉例：捕獲10或11和長度的十六進制`queue_id` 可以使用表達式`(?<queue_id>[0-9A-F]{10,11})` * 安裝自定義表達式 可以將自定義的表達式配置到Logstash中，就可以像于定義的表達式一樣使用； 1. 在Logstash根目錄下創建文件夾`patterns` 2. 在`patterns`文件夾中創建文件`extra`(文件名自定義) 3. 在文件“extra”中添加表達式，格式：`patternName regexp` 4. 如:`POSTFIX_QUEUEID [0-9A-F]{10,11}` 5. 使用自定義的表達式時需要指定“patterns_dir”變量舉例如下： ``` filter { grok { patterns_dir => "/path/to/your/own/patterns" match => { "message" => ........" } } } ``` ## B date時間處理插件 這里需要合前面的grok插件剝離出來的值logdate配合使用。可以格式化為需要的樣子。 **為什什么要格式化？** 日志產生的時間肯定早于日志在logstash中處理的時間,如果不修改，保存進ES的時間(@timestamp)就是logstash當前處理數據的時間,時間就會不一致 格式化以后，可以通過target屬性來指定到@timestamp，這樣數據的時間就會是準確的，對以后圖表的建設來說萬分重要。 最后，順手刪除logdate這個字段已無用的字段。 ~~~ filter{ date{ match=>["logdate","dd/MMM/yyyy:HH:mm:ss Z"] target=>"@timestamp" remove_field => 'logdate' } } ~~~ >需要強調的是，@timestamp字段的值，不可以隨便修改，最好就按照數據的某一個時間點來使用 如果是日志，就使用grok把時間摳出來，如果是數據庫，就指定一個字段的值來格式化 如果沒有這個字段的話，千萬不要試著去修改它。 ## C mutate插件 mutate 插件是 Logstash另一個重要插件。提供了豐富的基礎類型數據處理能力。可以重命名，刪除，替換和修改事件中的字段。 ~~~ filter { mutate { #接收一個數組，其形式為value，type convert => [ #把request_time的值轉換為浮點型，把costTime的值轉換為整型 "request_time", "float"， "costTime", "integer" ] } } ~~~ * convert 字段轉換 將指定字段轉換為指定類型，字段內容是數組，則轉換所有數組元素，如果字段內容是hash，則不做任何處理 目前支持的轉換類型包括：integer,float, string, and boolean. 例如： convert=> { “fieldname” => “integer” } * rename 字段重命名 修改一個或者多個字段的名稱。 例如： ` rename=> { “HOSTORIP” => “client_ip” }` * replace 值替換 使用新值完整的替換掉指定字段的原內容，支持變量引用。 例如： 使用字段“source_host”的內容拼接上字符串“: My new message”之后的結果替換“message”的值： `replace=> { “message” => “%{source_host}: My new message” }` * gsub 字符串替換 類似replace方法，但是只針對字符串類型有效 例如：`[ “fieldname”, “/”, “:“, “fieldname2”, “[\\?#-]”, “.”]`， 解釋：使用`:`替換掉`fieldname`中的所有`/`，使用“.”替換掉“fieldname2”中的所有`\ ? # 和-` * update 更新字段 更新現有字段的內容， 例如： 將“sample”字段的內容更新為“Mynew message”： `update=> { “sample” => “My new message” }` * join 連接數組字段 使用指定的符號將array字段的每個元素連接起來，對非array字段無效。 例如： 使用`,`將array字段`fieldname`的每一個元素連接成一個字符串： `join=> { “fieldname” => “,” } ` * merge 合并數組 合并兩個array或者hash，將一個array和一個hash合并。 例如： 將”added_field”合并到”dest_field”： `merge=> { “dest_field” => “added_field” }` * split 字段分割 按照自定的分隔符將字符串字段拆分成array字段，只能作用于string類型的字段。 例如： 將“fieldname”的內容按照`,`拆分成數組： `split=> { “fieldname” => “,” } ` * strip 去掉空格 去掉字段內容兩頭的空白字符。例如： `strip=> [“field1”, “field2”]` * uppercase 大寫轉換 將指定的字段值轉換為大寫 * lowercase 小寫轉換 將指定的字段值轉換為小寫 ## D ruby插件 ruby插件可以使用任何的ruby語法，無論是邏輯判斷，條件語句，循環語句，對字符串的操作，或是對EVENT對象的操作 **ruby插件有兩個屬性，一個init 還有一個code** init屬性是用來初始化字段的，這個字段只是在ruby{}作用域里面生效。 code屬性使用兩個冒號進行標識，所有ruby語法都可以在里面進行。 ~~~ filter { ruby { init => [field={}] code => " array=event.get('message').split('|') array.each do |value| if value.include? 'MD5_VALUE' then require 'digest/md5' md5=Digest::MD5.hexdigest(value) event.set('md5',md5) end if value.include? 'DEFAULT_VALUE' then event.set('value',value) end end remove_field=>"message" " } } #首先，把message字段里面的值拿到，并按照“|”分割為數組 #第二步，循環數組判斷其值是否是我需要的數據（ruby條件語法、循環結構） #第三步，把需要的字段添加進入EVEVT對象。 ##event就是Logstash對象，可以在ruby插件的code屬性里面操作，可以添加屬性字段，刪除，修改，數值運算。 #第四步，選取一個值，進行MD5加密或其他操作,最后把冗余的message字段去除。 ~~~ ## E json插件 JSON插件用于解碼JSON格式的字符串，一般是一堆日志信息中，部分是JSON格式，部分不是的情況下,會將json格式部分的數據解析出來放到單獨的字段中 和input中的codec插件的json區別在于,input中的插件,解析json格式后,會替代message中的數據,而filter中的json插件,解析出的字段是單獨的,不會覆蓋message的數據 ~~~ filter{ #source指定你的哪個值是json數據。 json { source => "value" } } ~~~ >如果你的json數據是多層的，那么解析出來的數據在多層結里是一個數組，你可以使用ruby語法對他進行操作，最終把所有數據都轉換為平級的。 json插件還是需要注意一下使用的方法的，下圖就是多層結構的弊端：  對應的解決方案為： ~~~ruby ruby{ code=>" kv=event.get('content')[0] kv.each do |k,v| event.set(k,v) end" remove_field => ['content','value','receiptNo','channelId','status'] } ~~~