Logstash使用grok插件解析Nginx日志

grok表達式的打印復制格式的完整語法是下面這樣的：

%{PATTERN_NAME:capture_name:data_type}
data_type 目前只支持兩個值：int 和 float。

在線gork正則的地址：http://grokdebug.herokuapp.com/
Logstash基礎正則地址：https://github.com/elastic/logstash/blob/v1.4.2/patterns/grok-patterns

也可以在你的安裝路徑下查找grok-patterns內置的正則表達式：

/usr/local/logstash-6.2.4/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns/grok-patterns

例如：

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

-----------------------------------------------------------------------------------------

logstash-filter-mutate 插件是Logstash 另一個重要插件,它提供了豐富的基礎類型數據處理能力,包括類型轉換,字符串處理和字段處理等

可以設置的轉換類型包括："integer"，"float" 和 "string"。示例如下：

filter {
    mutate {
        convert => ["request_time", "float"]
        gsub => [ "message", "aa", "" ]  字符串替換，此處為清除
    }
}

注意：mutate 除了轉換簡單的字符值，還支持對數組類型的字段進行轉換，即將 ["1","2"] 轉換成 [1,2]。但不支持對哈希類型的字段做類似處理。

對已有索引進行修改並且平滑過渡：

        mutate {
            convert => [ "request_time", "float" ]
            add_field => [ "response_time", "%{request_time}" ]
            remove_field => [ "request_time" ]         
        }

--------------------------------------------------------------------------------

在線grok表達式解析地址：http://grokdebug.herokuapp.com/  （NGINXACCESS為規則名稱，測試時不用填入）

 

根據已經生成的日志記錄逐個參數進行調試即可。

需要注意的是當日志或可能出現“-”時，必須在NGINXACCESS中指定，例如: (?:%{NUMBER:serverelapsed}|-)

需要修改的文件位置：/usr/local/logstash/vendor/bundle/jruby/2.3.0/gems/logstash-patterns-core-4.1.2/patterns/grok-patterns

如果找不到的話，grep -rw NGINXACCESS /usr/local/logstash

 

首先對nginx日志格式設定如下：

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

 

Logstash Grok默認的參數為：

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} %{QS:xforwardedfor} %{IPORHOST:host} %{BASE10NUM:request_duration}

 我這邊設置的是：

#nginx
WZ ([^ ]*)
NGINXACCESS %{IP:remote_ip} \- \- \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{WZ:request} HTTP/%{NUMBER:httpversion}" %{NUMBER:status} %{NUMBER:bytes} (?:%{QS:referer}|-) %{QS:agent} %{NUMBER:elapsed} (?:%{NUMBER:serverelapsed}|-) %{QS:xforward}

 或拆分請求的uri和請求參數：此處需要注意的是必須確保NGINXACCESS規則是一整行

WZ ([^ ]*)
URIPARAM [A-Za-z0-9$.+!*'|(){},~@#%&/=:;_?\-\[\]<>]*  #此處重新定義該匹配規則，不匹配參數開頭的？
NGINXACCESS %{IP:remote_ip} \- \- \[%{HTTPDATE:timestamp}\] "%{WORD:method} (%{URIPATH:request}|-|) (%{URIPARAM:requestParam}|-)" %{NUMBER:status}
%{NUMBER:bytes} %{QS:referer} %{QS:agent} %{NUMBER:elapsed} %{NUMBER:serverelapsed} %{QS:xforward}

  

自定義規則時也可以在/usr/local/logstash主目錄下新建一個目錄：patterns，並在該目錄下新建一個grok的表達式解析文件nginx，內容同上，也可以實現相同的目的。

 

Nginx logstash配置文件：

input{
  file{
    path => "/usr/local/nginx/logs/access.log"
    type => "nginx"
    start_position => "beginning"
  }
 
  file{
    path => "/usr/local/nginx/logs/admin.log"
    type => "admin-nginx"
    start_position => "beginning"
  }
}

filter {  
    grok {  
      match => { "message" => "%{NGINXACCESS}" }
    }
    mutate {
      convert => [ "elapsed", "float" ]
      convert => [ "serverelapsed", "float" ]
    }  
}
 
output{
  if [type] == "nginx" {
    elasticsearch {
    hosts=> ["172.17.102.202:9200"]
    index=> "nginx"
    }
  }
  else {
    elasticsearch {
    hosts=> ["172.17.102.202:9200"]
    index=> "admin-nginx"
    }
  }
}