Logstash解析Nginx访问日志

Nginx日志格式

    log_format  main  '$remote_addr - $remote_user [$time_local] "$request" '                      '$status $body_bytes_sent "$http_referer" '                      '"$http_user_agent" "$http_x_forwarded_for" '                        '"$http_host" "$request_time" "$upstream_response_time" $http_device $http_appversion $upstream_addr '                      '$http_openudid $http_code $http_networkType "$http_deviceModel" "$http_osVersion"';

解析格式为

%{IPORHOST:Client_IP} (%{WORD:ident}|-) (%{USERNAME:auth}|-) \[%{HTTPDATE:timestamp}\] "%{WORD:Http_Method} %{URIPATHPARAM:Http_Request} HTTP/%{NUMBER:Http_Version}" %{NUMBER:Http_Status_Code} (?:%{NUMBER:Http_Bytes}|-) (?:"(?:%{URI:Http_Referrer}|-)"|%{QS:Http_Referrer}) %{QS:User_Agent} "(%{QS:X_Forwarded_For}|-)" "(%{IPORHOST:Site}|-)" "(%{NUMBER:Request_Time}|-)" "(%{NUMBER:Upstream_Response_Time}|-)" (%{WORD:Device}|-) (%{USERNAME:App_Version}|-) (%{HOSTNAME:Upstream_Host}:%{POSINT:Upstram_Port}|-) (%{WORD:Openudid}|-) (%{WORD:Usercode}|-) (%{WORD:NetType}|-) "(%{GREEDYDATA:Device_Name}|-)" "(%{GREEDYDATA:System_Verion}|-)"

logstash配置文件如下:

input {    redis {        data_type => "list"        key => "filebeat:nginx-access"        host => "redis"        port => 6379        db => 0        # password => "123456"        # codec => "json"    }}filter {        grok {                match => {                        "message" => '%{IPORHOST:Client_IP} (%{WORD:ident}|-) (%{USERNAME:auth}|-) \[%{HTTPDATE:timestamp}\] "%{WORD:Http_Method} %{URIPATHPARAM:Http_Request} HTTP/%{NUMBER:Http_Version}" %{NUMBER:Http_Status_Code} (?:%{NUMBER:Http_Bytes}|-) (?:"(?:%{URI:Http_Referrer}|-)"|%{QS:Http_Referrer}) %{QS:User_Agent} "(%{QS:X_Forwarded_For}|-)" "(%{IPORHOST:Site}|-)" "(%{NUMBER:Request_Time}|-)" "(%{NUMBER:Upstream_Response_Time}|-)" (%{WORD:Device}|-) (%{USERNAME:App_Version}|-) (%{HOSTNAME:Upstream_Host}:%{POSINT:Upstram_Port}|-) (%{WORD:Openudid}|-) (%{WORD:Usercode}|-) (%{WORD:NetType}|-) "(%{GREEDYDATA:Device_Name}|-)" "(%{GREEDYDATA:System_Verion}|-)"'                }        }        date {                match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]        }        geoip {                database => "/usr/share/logstash/geodb/GeoLite2-City.mmdb"                source => "Client_IP"                target => "geoip"                fields => ["country_name","region_name", "city_name", "ip", "longitude", "latitude", "location"]                add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]                add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}"  ]        }        useragent {                source => "User_Agent"                target => "ua"        }        useragent {                source => "User_Agent"                target => "ua"        }        mutate {                convert => [ "[geoip][coordinates]", "float" ]                convert => [ "Http_Status_Code", "integer" ]                convert => [ "Http_Bytes", "integer" ]                convert => [ "Request_Time", "float" ]                convert => [ "Upstream_Response_Time", "float" ]                # split => ["Http_Request", "?"]                # add_field => { "Http_URI" => "%{Http_Request[0]}" }                remove_field => [ "message", "beat", "@version", "auth", "prospector", "source", "offset"]        }}output {        #stdout {codec => rubydebug}        elasticsearch {                hosts => ["elasticsearch:9200"]                index => "nginx-access-%{+YYYY.MM.dd}"        }}