logstash采集nginx日志

logstash采集nginx日志

nginx 日志格式自定义

x
log_format json_log '{"time": "$time_local", '
                        '"remote_addr": "$remote_addr", '
                        '"reallyip": "$clientip", '
                        '"proxy_ip": "$http_x_forwarded_for", '
                        '"complete_url": "$host$request_uri", '
                        '"referer": "$http_referer", '
                        '"request_method": "$request_method", '
                        '"request_uri": "$request_uri", '
                        '"request_time": "$request_time", '
                        '"http_host": "$host", '
                        '"status": $status, '
                        '"sent_bytes": $body_bytes_sent, '
                        '"up_addr": "$upstream_addr", '
                        '"up_host": "$upstream_http_host", '
                        '"up_resp_time": "$upstream_response_time", '
                        '"agent": "$http_user_agent"}';

map $http_x_forwarded_for $clientip{
        "" $remote_addr;
        ~^(?P<firstAddr>[0-9\.]+),?.*$ $firstAddr;
    }
# server 使用
access_log  access.log  json_log;

采集nginx日志

x
input {
    file{
        path => "/var/log/nginx/access.log"      #需要采集的json格式nginx access日志文件路径
        codec => json                            #数据格式
        start_position => "beginning"            #默认为采集新增内容，begining为从文件开头开始采集
        type => "nginx_log"                      #类型
    }
}
output {
    elasticsearch {
        hosts => "eshost:9200"                  #es地址
        index => "nginx-%{+YYYY.MM.dd}"      #写入的索引名
    }
    # stdout { codec => rubydebug } 输出到控制台（调试）
}

过滤

x
filter {
        grok {
                match => { "message" => "%{IPORHOST:clientip} - %{NOTSPACE:remote_user} \[%{HTTPDATE:timestamp}\] \"(?:%{WORD:verb} %{NOTSPACE:request}(?: HTTP/%{NUMBER:httpversion})?|%{DATA:rawrequest})\" %{NUMBER:response_status} %{NUMBER:body_sent} %{QS:referrer} %{DATA:reallyip} \"%{HOSTNAME:domain}\" \"%{DATA:upstream}\" \"%{NUMBER:request_time}\"   \"%{NUMBER:response_time}\" %{QS:agent}" }
        }
        geoip {
                source => "clientip"
                #source => "http_x_forwarded_for"
                target => "geoip"
                database => "/zhiyoubao/elk_server/logstash-5.4.0/ipdb/GeoLite2-City.mmdb"
                add_field => [ "[geoip][coordinates]", "%{[geoip][longitude]}" ]
                add_field => [ "[geoip][coordinates]", "%{[geoip][latitude]}" ]
        }  # 使用IP定位,生成map热点图

        mutate {
                convert => [ "[geoip][coordinates]", "float" ]
                convert => [ "response","integer" ]
                convert => [ "bytes","integer" ]
                replace => { "type" => "%{type}" }
                #remove_field => "message"
        }
        date {
                match => [ "timestamp","dd/MMM/yyyy:HH:mm:ss Z"]
        }
}

出现中文\x \xx

x
filter {
        ruby {
                code => " if event.get('message')
                        event.set('message', event.get('message').gsub('\x','Xx'))
                        event.set('message', event.get('message').gsub('\\x','XXx'))
                        end
                        "
        }
        json {
                source => "message"
        }
        geoip {
                source => "clientip"
                target => "geoip"
                database => "/etc/logstash/ipdb/GeoLite2-City.mmdb"
        }
        urldecode {
            all_fields => true
        }
        mutate {
                remove_field => "message"
                remove_field => "@version"
                remove_field => "offset"
                remove_field => "input"
                remove_field => "source"
                remove_field => "beat"
                remove_field => "prospector"
                remove_field => "tags"
        } # 移除一些不必要信息
}