Logstash收集nginx访问日志和错误日志

1、收集访问日志

1）、首先是要在nginx里面配置日志格式化输出

    log_format  main  "$http_x_forwarded_for | $time_local | $request | $status | $body_bytes_sent | $request_body | $content_length | $http_referer | $http_user_agent |"
                      "$http_cookie | $remote_addr | $hostname | $upstream_addr | $upstream_response_time | $request_time" ;

    access_log  /var/log/nginx/access.log  main;

2）、接下来开始在logstash创建处理nginx的配置文件

input {
        file {
                path => ["/var/log/nginx/access.log"]
        }
}

filter {
        ruby {
                init => "@kname =['http_x_forwarded_for','time_local','request','status','body_bytes_sent','request_body','content_length','http_referer','http_user_agent','http_cookie','remote_addr','hostname','upstream_addr','upstream_response_time','request_time']"
                code => "new_event = LogStash::Event.new(Hash[@kname.zip(event.get('message').split('|'))])
                new_event.remove('@timestamp')
                event.append(new_event)
                "
        }

if [request] {
        ruby {
                init => "@kname = ['method','uri','verb']"
                code => "
                        new_event = LogStash::Event.new(Hash[@kname.zip(event.get('request').split(' '))])
                        new_event.remove('@timestamp')
                        event.append(new_event)
                "
        }
 } 
if [uri] {
        ruby{
                init => "@kname = ['url_path','url_args']"
                code => "
                        new_event = LogStash::Event.new(Hash[@kname.zip(event.get('uri').split('?'))])
                        new_event.remove('@timestamp')
                        event.append(new_event)
                "
        }
 }
kv {
        prefix =>"url_"
        source =>"url_args"
        field_split =>"&"
        include_keys => ["uid","cip"]
        remove_field => ["url_args","uri","request"]
}
mutate {
        convert => [
                "body_bytes_sent","integer",
                "content_length","integer",
                "upstream_response_time","float",
                "request_time","float"
        ]
 }
date {
        match => [ "time_local","dd/MMM/yyyy:hh:mm:ss Z" ]
        locale => "en"
 }
}
output{stdout{}}

此处的例子借鉴ELKstack权威指南里面的例子，不过书中的例子有错，我这里修改好了，可以参考书籍39页和66页

github:https://github.com/weixinqing/Logstash-example/blob/master/initnginx.conf

3）、最后允许一下看一下效果所示：

{
                  "url_path" => "/",
           "body_bytes_sent" => 0,
                  "@version" => "1",
                   "message" => "- | 05/Mar/2019:16:21:40 +0800 | GET / HTTP/1.1 | 304 | 0 | - | - | - | Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 SE 2.X MetaSr 1.0 |- | 172.16.0.10 | elk-chaofeng07 | - | - | 0.000",
                      "host" => "ELK-chaofeng07",
               "http_cookie" => "- ",
             "upstream_addr" => " - ",
    "upstream_response_time" => 0.0,
                "@timestamp" => 2019-03-05T08:21:41.352Z,
                       "uri" => "/",
                   "request" => " GET / HTTP/1.1 ",
                      "path" => "/var/log/nginx/access.log",
                  "url_args" => nil,
                  "hostname" => " elk-chaofeng07 ",
                      "verb" => "HTTP/1.1",
           "http_user_agent" => " Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.36 SE 2.X MetaSr 1.0 ",
                "time_local" => " 05/Mar/2019:16:21:40 +0800 ",
              "request_body" => " - ",
               "remote_addr" => " 172.16.0.10 ",
                    "status" => " 304 ",
              "request_time" => 0.0,
                    "method" => "GET",
              "http_referer" => " - ",
                      "tags" => [
        [0] "_dateparsefailure"
    ],
            "content_length" => 0,
      "http_x_forwarded_for" => "- "
}

唯一不足的就是中间报了个错误，可以自行解决一下。

2、收集错误日志

定义logstash处理的配置文件

input{
        file {
                path => ["/var/log/nginx/error.log"]
        }
}
filter{
        grok {
                match => {"message" => "(?<datetime>dddd/dd/dd dd:dd:dd) [(?<errortype>w+)] S+: *d+ (?<errormsg>[^,]+), w+: %{IP:remotehost}, w+: w+, w+: (?<request>[^,]+), w+: "%{IP:localhost}""}
        }
        mutate {
                remove_field => ["message"]
        }
if [request] {
        ruby {
                init => "@kname = ['method','uri','verb']"
                code => "
                        new_event = LogStash::Event.new(Hash[@kname.zip(event.get('request').split(' '))])
                        new_event.remove('@timestamp')
                        event.append(new_event)
                "
        }
}

}
output{stdout{}}

查看一下效果：

{
      "@version" => "1",
          "path" => "/var/log/nginx/error.log",
    "remotehost" => "172.16.0.10",
       "request" => ""GET /8 HTTP/1.1"",
          "verb" => "HTTP/1.1"",
           "uri" => "/8",
          "host" => "ELK-chaofeng07",
     "localhost" => "172.16.0.57",
        "method" => ""GET",
    "@timestamp" => 2019-03-05T10:43:54.377Z,
      "datetime" => "2019/03/05 18:43:53",
      "errormsg" => "open() "/usr/share/nginx/html/8" failed (2: No such file or directory)",
     "errortype" => "error"
}