Logstash使用grok過濾nginx日誌（二）

　　在生產環境中，nginx日誌格式每每使用的是自定義的格式，咱們須要把logstash中的message結構化後再存儲，方便kibana的搜索和統計，所以須要對message進行解析。

　　本文采用grok過濾器，使用match正則表達式解析，根據本身的log_format定製。

一、nginx日誌格式

　　log_format配置以下：

log_format  main  '$remote_addr - $remote_user [$time_local] $http_host $request_method "$uri" "$query_string" '
                  '$status $body_bytes_sent "$http_referer" $upstream_status $upstream_addr $request_time $upstream_response_time '
                  '"$http_user_agent" "$http_x_forwarded_for"' ;

　　對應的日誌以下：

1.1.1.1 - - [06/Jun/2016:00:00:01 +0800] www.test.com GET "/api/index" "?cms=0&rnd=1692442321" 200 4 "http://www.test.com/?cp=sfwefsc" 200 192.168.0.122:80 0.004 0.004 "Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/45.0.2454.101 Safari/537.36" "-"

二、編寫正則表達式

　　logstash中默認存在一部分正則讓咱們來使用，能夠訪問Grok Debugger來查看，能夠在 $logstash/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-4.0.0/patterns/ 目錄裏面查看。

　　基本定義在grok-patterns中，咱們可使用其中的正則，固然並非全部的都適合nginx字段，這時就須要咱們自定義正則，而後經過指定patterns_dir來調用。

　　同時在寫正則的時候可使用Grok Debugger或者Grok Comstructor工具來幫助咱們更快的調試。在不知道如何使用logstash中的正則的時候也可以使用Grok Debugger的Descover來自動匹配。

　　1）nginx標準日誌格式

　　　　logstash自帶的grok正則中有Apache的標準日誌格式：

COMMONAPACHELOG %{IPORHOST:clientip} %{HTTPDUSER:ident} %{USER:auth} \[%{HTTPDATE:timestamp}\] "(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})" %{NUMBER:response} (?:%{NUMBER:bytes}|-)
COMBINEDAPACHELOG %{COMMONAPACHELOG} %{QS:referrer} %{QS:agent}

　　　　對於nginx標準日誌格式，能夠發現只是最後多了一個 $http_x_forwarded_for 變量。則nginx標準日誌的grok正則定義爲：

MAINNGINXLOG %{COMBINEDAPACHELOG} %{QS:x_forwarded_for}

　　2）自定義格式

　　　　經過log_format來匹配對應的正則以下：

%{IPV4:remote_addr} - (%{USERNAME:user}|-) \[%{HTTPDATE:log_timestamp}\] (%{HOSTNAME1:http_host}|-) (%{WORD:request_method}|-) \"(%{URIPATH1:uri}|-|)\" \"(%{URIPARM1:param}|-)\" %{STATUS:http_status} (?:%{BASE10NUM:body_bytes_sent}|-) \"(?:%{GREEDYDATA:http_referrer}|-)\" (%{STATUS:upstream_status}|-) (?:%{HOSTPORT1:upstream_addr}|-) (%{BASE16FLOAT:upstream_response_time}|-) (%{STATUS:request_time}|-) \"(%{GREEDYDATA:user_agent}|-)\" \"(%{FORWORD:x_forword_for}|-)\"

　　　　這裏面有幾個是我自定義的正則：

URIPARM1 [A-Za-z0-9$.+!*'|(){},~@#%&/=:;^\\_<>`?\-\[\]]*
URIPATH1 (?:/[\\A-Za-z0-9$.+!*'(){},~:;=@#% \[\]_<>^\-&?]*)+
HOSTNAME1 \b(?:[0-9A-Za-z_\-][0-9A-Za-z-_\-]{0,62})(?:\.(?:[0-9A-Za-z_\-][0-9A-Za-z-:\-_]{0,62}))*(\.?|\b)
STATUS ([0-9.]{0,3}[, ]{0,2})+
HOSTPORT1 (%{IPV4}:%{POSINT}[, ]{0,2})+
FORWORD (?:%{IPV4}[,]?[ ]?)+|%{WORD}

　　message是每段讀進來的日誌，IPORHOST、USERNAME、HTTPDATE等都是patterns/grok-patterns中定義好的正則格式名稱，對照日誌進行編寫。

　　grok pattren的語法爲：%{SYNTAX:semantic}，":" 前面是grok-pattrens中定義的變量，後面能夠自定義變量的名稱。(?:%{SYNTAX:semantic}|-)這種形式是條件判斷。

　　若是有雙引號""或者中括號[]，須要加 \ 進行轉義。

　　詳解自定義正則：

 URIPARAM \?[A-Za-z0-9$.+!*'|(){},~@#%&/=:;_?\-\[\]<>]* 

 URIPARM1 [A-Za-z0-9$.+!*'|(){},~@#%&/=:;^\\_<>`?\-\[\]]* grok-patterns中正則表達式，能夠看到grok-patterns中是以「?」開始的參數，在nginx的 $query_string 中已經把「?」去掉了，因此咱們這裏再也不須要「?」。另外單獨加入日誌中出現的  ^ \ _ < > ` 特殊符號 

 URIPATH (?:/[A-Za-z0-9$.+!*'(){},~:;=@#%&_\-]*)+ 

 URIPATH1 (?:/[\\A-Za-z0-9$.+!*'(){},~:;=@#% \[\]_<>^\-&?]*)+ grok-patterns中正則表達式，grok-patterns中的URIPATH不能匹配帶空格的URI，因而在中間加一個空格。另外還有 \ [ ] < > ^ 特殊符號。

 HOSTNAME \b(?:[0-9A-Za-z][0-9A-Za-z-]{0,62})(?:\.(?:[0-9A-Za-z][0-9A-Za-z-]{0,62}))*(\.?|\b) 

 HOSTNAME1 \b(?:[0-9A-Za-z_\-][0-9A-Za-z-_\-]{0,62})(?:\.(?:[0-9A-Za-z_\-][0-9A-Za-z-:\-_]{0,62}))*(\.?|\b) 添加匹配 http_host 中帶有 "-" 的字符。

 HOSTPORT %{IPORHOST}:%{POSINT} 

 HOSTPORT1 (%{IPV4}:%{POSINT}[, ]{0,2})+ 在匹配 upstream_addr 字段時發現，會出現多個IP地址的狀況出現，匹配多個IP地址。

 STATUS ([0-9.]{0,3}[, ]{0,2})+ 該字段是當出現多個 upstream_addr 字段時匹配多個 http_status 。

 FORWORD (?:%{IPV4}[,]?[ ]?)+|%{WORD} 當 x_forword_for 字段出現多個IP地址時匹配。

　　nginx左右字段都定義完成，可使用Grok Debugger或者Grok Comstructor工具來測試。添加自定義正則的時候，在Grok Debugger中能夠勾選「Add custom patterns」。

　　以上日誌匹配結果爲：

{
  "remote_addr": [
    "1.1.1.1"
  ],
  "user": [
    "-"
  ],
  "log_timestamp": [
    "06/Jun/2016:00:00:01 +0800"
  ],
  "http_host": [
    "www.test.com"
  ],
  "request_method": [
    "GET"
  ],
  "uri": [
    "/api/index"
  ],
  "param": [
    "?cms=0&rnd=1692442321"
  ],
  "http_status": [
    "200"
  ],
  "body_bytes_sent": [
    "4"
  ],
  "http_referrer": [
    "http://www.test.com/?cp=sfwefsc"
  ],
  "port": [
    null
  ],
  "upstream_status": [
    "200"
  ],
  "upstream_addr": [
    "192.168.0.122:80"
  ],
  "upstream_response_time": [
    "0.004"
  ],
  "request_time": [
    "0.004"
  ],
  "user_agent": [
    ""Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/45.0.2454.101 Safari/537.36""
  ],
  "client_ip": [
    "2.2.2.2"
  ],
  "x_forword_for": [
    null
  ]
}

三、logstash的配置文件

　　建立自定義正則目錄

# mkdir -p /usr/local/logstash/patterns
# vi /usr/local/logstash/patterns/nginx

　　而後寫入上面自定義的正則

URIPARM1 [A-Za-z0-9$.+!*'|(){},~@#%&/=:;_?\-\[\]]*
URIPATH1 (?:/[A-Za-z0-9$.+!*'(){},~:;=@#%&_\- ]*)+
URI1 (%{URIPROTO}://)?(?:%{USER}(?::[^@]*)?@)?(?:%{URIHOST})?(?:%{URIPATHPARAM})?
NGINXACCESS %{IPORHOST:remote_addr} - (%{USERNAME:user}|-) \[%{HTTPDATE:log_timestamp}\] %{HOSTNAME:http_host} %{WORD:request_method} \"%{URIPATH1:uri}\" \"%{URIPARM1:param}\" %{BASE10NUM:http_status} (?:%{BASE10NUM:body_bytes_sent}|-) \"(?:%{URI1:http_referrer}|-)\" (%{BASE10NUM:upstream_status}|-) (?:%{HOSTPORT:upstream_addr}|-) (%{BASE16FLOAT:upstream_response_time}|-) (%{BASE16FLOAT:request_time}|-) (?:%{QUOTEDSTRING:user_agent}|-) \"(%{IPV4:client_ip}|-)\" \"(%{WORD:x_forword_for}|-)\"

　　logstash.conf配置文件內容

input {
        file {
                path => "/data/nginx/logs/access.log"
                type => "nginx-access"
                start_position => "beginning"
                sincedb_path => "/usr/local/logstash/sincedb"
        }
}
filter {
        if [type] == "nginx-access" {
                grok {
                        patterns_dir => "/usr/local/logstash/patterns"        //設置自定義正則路徑
                        match => {
                                "message" => "%{NGINXACCESS}"
                        }
                }
                date {
                        match => [ "log_timestamp" , "dd/MMM/YYYY:HH:mm:ss Z" ]
                }
　　　　　　　　　urldecode {
        　　　　　　　　　all_fields => true
    　　　　　　　}
　　　　　　　　　//把全部字段進行urldecode（顯示中文）
        }
}
output {
        if [type] == "nginx-access" {
                elasticsearch {
                        hosts => ["10.10.10.26:9200"]
                        manage_template => true
                        index => "logstash-nginx-access-%{+YYYY-MM}"
                }
        }

}

 

 

 四、啓動logstash，而後就能夠查看日誌是否寫入elasticsearch中。