Logstash語法經常使用案例解析（二）
摘要html
此篇主要講Filter插件，已經對nginx 日誌的各類處理實例nginx
接着上篇繼續說插件git
1，Filter插件json
Grok：正則捕獲數組
Date：時間處理ruby
Mutate：數據修改bash
Geoip：查詢歸類網絡
JSON：編解碼app
Grok：解析和結構化任何文本。dom
http://grokdebug.herokuapp.com/patterns# 匹配規則,注意空格，若是空格不匹配也會報錯
http://grokdebug.herokuapp.com/ 匹配檢查，並且有語法提示
Grok 目前是logstash最好的方式對非結構化日誌數據解析成結構化和可查詢化。logstash內置了120個匹配模式，知足大部分需求。
格式：
filter {
grok {
match => { "message" => "grok_pattern" }
}
}
注 ：
grok_pattern由零個或多個%{SYNTAX:SEMANTIC}組成，其中SYNTAX是表達式的名字，是由grok提供的，例如數字表達式的名字是NUMBER，IP地址表達式的名字是IP。SEMANTIC表示解析出來的這個字符的名字，由本身定義，例如IP字段的名字能夠是client。
簡單例子：
#cat conf.d/test.conf
input {stdin{}} #輸入方式爲標準輸入
filter {
grok { #grok插件匹配
#patterns_dir => "/path/to/patterns" #將匹配規則寫到指定文件方便管理
match => {
"message" => "%{WORD} %{NUMBER:request_time:float} %{WORD}"
} #WORD匹配字符串，NUMBER匹配數值，支持int,float格式。匹配的值賦給request_time變量
#remove_field => ["message"] #處理結果刪除掉message字段
}
}
output { #輸出方式爲標準輸出
stdout {codec=>rubydebug} #定義輸出格式爲rubydebug
}
結果：
# ./bin/logstash -f conf.d/test.conf
Logstash startup completed
begin 123.456 end
{
"message" => "begin 123.456 end", #remove以後就不顯示了。
"@version" => "1",
"@timestamp" => "2016-05-09T02:43:47.952Z",
"host" => "dev-online",
"request_time" => 123.456 #grok匹配中新加的變量
}
Nginx 日誌處理匹配：
由於nginx日誌已經被處理成json數據，傳過來就是key:value的方式，打印成rubydebug格式以下：
因此如今想要篩選不要的字段
input {
file {
path => "/var/log/nginx/access.log"
type => "json"
codec => "json"
start_position => "beginning"
}
}
filter {
grok {
match => {
"@timestamp" =>"%{WORD}" #先把不想要的字段匹配出來
"type" => "%{WORD}"
}
remove_field => ["@timestamp","type"] ＃再移除字段
}
}
output {
stdout {
codec=>rubydebug
}
}
運行結果：
nginx 日誌json格式：
log_format json '{"@timestamp":"$time_iso8601",'
'"@version":"1",'
'"host":"$server_addr",'
'"client":"$remote_addr",'
'"size":$body_bytes_sent,'
'"responsetime":$request_time,'
'"domain":"$host",'
'"url":"$request",'
'"refer":"$http_referer",'
'"agent":"$http_user_agent",'
'"status":"$status"}';
access_log /var/log/nginx/access.log json;
nginx配置文件經常使用正則匹配參數
nginx 日誌格式 匹配項目 備註
$remote_addr %{IPORHOST:clientip}
$remote_user %{NOTSPACE:remote_user}
[$time_local] \[%{HTTPDATE:timestamp}\] 「[]"須要屬於特殊字符須要轉義一下
"$request" "%{WORD:method} 訪問請求，通常都加"",匹配時也加一下。 WORD匹配GET,POST
method %{URIPATHPARAM:request} URIPATHPARAM匹配請求的uri
HTTP %{NUMBER:httpversion}" NUMBER匹配數字，並賦值給httpversion http協議版本
$status %{NUMBER:status} NUMBER匹配數字,並賦值給status,做爲返回狀態
$body_bytes_sent %{NUMBER:response} 內容大小
"$http_referer" "%{QS:referrer}" 匹配請求refer
"$http_user_agent" "%{QS:agent}" 匹配親切agent
"$http_x_forwarded_for" "%{QS:xforwardedfor}" 匹配xfw
$upstream_addr %{IPV4:upstream}:%{POSINT:port}
$scheme %{WORD:scheme} 匹配http or https
eg: nginx日誌格式：
log_format access '$remote_addr - $remote_user [$time_local] "$request" $status $body_bytes_sent "$http_referer" "$http_user_agent"';
日誌實例：
"192.168.1.22 - - [20/Apr/2016:16:28:14 +0800] "GET /ask/232323.html HTTP/1.1" 500 15534 "http://test.103.100xhs.com/" "Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/45.0.2454.101 Safari/537.36""
匹配規則
"%{IPORHOST:clientip} - %{NOTSPACE:remote_user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:status} %{NUMBER:response} %{QS:referrer} %{QS:agent}"
Geoip地址查詢：
GeoIP 是最多見的免費 IP 地址歸類查詢庫，GeoIP 庫能夠根據 IP地址提供對應的地域信息，包括國別，省市，經緯度等，對於可視化地圖和區域統計很是有用
input {
stdin{}
}
filter {
geoip {
source => "message" #source必須爲公網ip 不然geoip不會顯示數據 #
fields => ["city_name","country_code2","country_name","latitude","longitude"]
} #geoip輸出的內容比較多，能夠指定輸出的列
}
output {
stdout{
codec=>rubydebug
}
}
找到對應IP的key 就是geoip 中source所指定的值。
完整的例子：
結果：
注意：geoip 插件的 "source" 字段能夠是任一處理後的字段，好比 "client_ip"，可是字段內容卻須要
當心！geoip 庫內只存有公共網絡上的 IP 信息，查詢不到結果的，會直接返回 null，
JSON:
input {stdin{}}
filter {
json {
source => "message" #必選項
}
}
output {
stdout{
codec=>rubydebug
}
}
結果：
{"name":"wd","age":"15"}
{
"message" => "{\"name\":\"wd\",\"age\":\"15\"}",
"@version" => "1",
"@timestamp" => "2016-05-09T06:32:13.546Z", #加一個時間戳的好處是方便kibana導入
"host" => "dev-online",
"name" => "wd",
"age" => "15"
}
Date事件處理
注意：由於在稍後的 outputs/elasticsearch 中經常使用的 %{+YYYY.MM.dd} 這種寫法必須讀取 @timestamp 數據，因此必定不要直接刪掉這個字段保留本身的字段，而是應該用 filters/date 轉換後刪除本身的字段！
filter {
grok {
match => ["message", "%{HTTPDATE:logdate}"]
}
date {
match => ["logdate", "dd/MMM/yyyy:HH:mm:ss Z"]
}
}
注意：時區偏移量只須要用一個字母 Z 便可。
Mutate數據修改
1,類型轉換
能夠設置的轉換類型包括："integer"，"float" 和 "string"。示例以下
filter {
mutate {
convert => ["request_time", "float"]
}
}
注意：mutate 除了轉換簡單的字符值，還支持對數組類型的字段進行轉換，即將 ["1","2"] 轉換成 [1,2]。但不支持對哈希類型的字段作相似處理。有這方面需求的能夠採用稍後講述的 filters/ruby 插件完成。
2,字符串處理
gsub 僅對字符串類型字段有效
gsub => ["urlparams", "[\\?#]", "_"]
split
split => ["message", "|"]
隨意輸入一串以|分割的字符，好比 "123|321|adfd|dfjld*=123"，能夠看到以下輸出：
join 僅對數組類型字段有效
咱們在以前已經用 split 割切的基礎再 join 回去。配置改爲：
join => ["message", ","]
merge合併兩個數組或者哈希字段。依然在以前 split 的基礎上繼續：
merge => ["message", "message"]
rename 重命名某個字段，若是目的字段已經存在，會被覆蓋掉：
rename => ["syslog_host", "host"]
update 更新某個字段的內容。若是字段不存在，不會新建。
replace 做用和 update 相似，可是當字段不存在的時候，它會起到 add_field 參數同樣的效果，自動添加新的字段。
Codec編碼插件
json: 直接輸入預約義好的 JSON 數據，這樣就能夠省略掉 filter/grok 配置！
path => "/var/log/nginx/access.log_json""
codec => "json"
Multiline:合併多行數據
stdin {
codec => multiline {
pattern => "^\["
negate => true
what => "previous"
}
}
終端輸入：以 the end 結束，換行沒法結束。
- 1. Logstash語法經常使用案例解析（一）
- 2. logstash經常使用插件解析
- 3. Spark 經常使用案例
- 4. 經常使用 MySQL 案例!!
- 5. hive經常使用語法示例
- 6. markdown經常使用語法事例
- 7. es6經常使用語法（二）
- 8. sass經常使用語法以及sass換膚案例
- 9. 經常使用算法案例之動態規劃（C語言）
- 10. swoole使用 經常使用案例
- 更多相關文章...
- • XML DOM 解析器 - XML DOM 教程
- • Redis中使用Lua語言 - Redis教程
- • 常用的分佈式事務解決方案
- • Kotlin學習（一）基本語法
-
每一个你不满意的现在，都有一个你没有努力的曾经。
- 1. Logstash語法經常使用案例解析（一）
- 2. logstash經常使用插件解析
- 3. Spark 經常使用案例
- 4. 經常使用 MySQL 案例!!
- 5. hive經常使用語法示例
- 6. markdown經常使用語法事例
- 7. es6經常使用語法（二）
- 8. sass經常使用語法以及sass換膚案例
- 9. 經常使用算法案例之動態規劃（C語言）
- 10. swoole使用 經常使用案例