logstash收集ngx日志

if [type] =~ "ngx-" {
		#去掉重复的行
        mutate { 
            add_field => {"line_message" => "%{message} %{offset}"}	
        }
        ruby {
	    code => "
		require 'digest/md5';
		event.set('computed_id', Digest::MD5.hexdigest(event.get('line_message')))
            "
	    }
		#匹配nginx日志
    	grok {
             match => { "message" => "%{IPORHOST:clientip} - %{NOTSPACE:remote_user} [%{HTTPDATE:timestamp}] "(?:%{WORD:method} %{NOTSPACE:request}(?: %{URIPROTO:proto}/%{NUMBER:httpversion})?|%{DATA:rawrequest})" %{NUMBER:status} (?:%{NUMBER:size}|-) "(?:%{DATA:referrer}|-)" "(?:%{DATA:agent}|-)" "(%{DATA:xforwardedfor}|-)" "(?:%{DATA:domain}|-)" "%{NUMBER:server_port}" %{NUMBER:reqtime} %{DATA:forward_ip}" }
             remove_field => ["source","host","message","forward_ip","domain","beat.name","remote_user"]
        }
		#设置日期格式
        date {
            match => ["timestamp", "dd/MMM/YYYY:HH:mm:ss +0800"]
            target => "@timestamp"
            "locale" => "en"
            timezone => "UTC"
            remove_field => ["timestamp"]
        }
		#设置IP地址
        geoip {
            source => "clientip"
            target => "geoip"
        }
		#删除[geoip][ip]以192.100.10.的IP地址
	    if [geoip][ip] =~ "192.100.10." {
            drop {}
	    }
		#删除行信息
        mutate {
            remove_field => ["line_message"]
        }
        useragent {
            source => "agent"
        }
    }