如何编写logstash-filter插件

gem "logstash-filter-test", :path => "vendor/bundle/jruby/1.9/gems/logstash-filter-test-1.0.0"

Gem::Specification.new do |s|

  s.name            = 'logstash-filter-geoip'
  s.version         = '4.0.4'
  s.licenses        = ['Apache License (2.0)']
  s.summary         = "$summary"
  s.description     = "This gem is a Logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/logstash-plugin install gemname. This gem is not a stand-alone program"
  s.authors         = ["Elastic"]
  s.email           = 'info@elastic.co'
  s.homepage        = "http://www.elastic.co/guide/en/logstash/current/index.html"
  s.require_paths = ["lib"]
  s.platform      = "java"

  # Files
  s.files = Dir['lib/**/*','spec/**/*','vendor/**/*','*.gemspec','*.md','CONTRIBUTORS','Gemfile','LICENSE','NOTICE.TXT', 'maxmind-db-NOTICE.txt']

  # Tests
  s.test_files = s.files.grep(%r{^(test|spec|features)/})

  # Special flag to let us know this is actually a logstash plugin
  s.metadata = { "logstash_plugin" => "true", "logstash_group" => "filter" }

  # Gem dependencies
  s.add_runtime_dependency "logstash-core-plugin-api", ">= 1.60", "<= 2.99"

  s.requirements << "jar com.maxmind.geoip2:geoip2, 2.5.0, :exclusions=> [com.google.http-client:google-http-client]"

	s.name		=	'logstash-filter-test'
	s.version	=	'1.0.0'

#encoding:utf-8

require "logstash/filters/base"

require "logstash/namespace"

require 'net/https'
require '/logstash/vendor/bundle/jruby/1.9/gems/json-1.8.3-java/lib/json.rb'
require 'digest/md5'
require 'uri'

class LogStash::Filters::Test < LogStash::Filters::Base

  config_name "test"

  config :remote_ip, :validate => :string
  config :tag_ip_failure, :validate => :array, :default => ["_test_parseip_failure"]


  public
  def register
	begin
		uri = '获取IP配置表接口的URL地址'
		$strIpInfor = JSON.parse httpPostMethod(uri)
	rescue
		puts "get ininfor error..."
	end

  end

  public

  def filter(event)
	begin
		if nil != event[@remote_ip] && event[@remote_ip].scan(".").length == 3 
			ipRange = getIpRange(event[@remote_ip])
			event.set("source",ipRange[0])
			event.set("sourceType",ipRange[1])
		else
			event.set("source",'other')
			event.set("sourceType",'other')
		end
	rescue
		event.set("source",'other')
		event.set("sourceType",'other')
		@tag_ip_failure.each{|tag| event.tag(tag)}
	end	
  end # def filter

#get ip range
public
def getIpRange(ip)
    for strIp in $strIpInfor
	puts strIp['ipinfor']
	ipinfor = strIp['ipinfor'].split('/')	#get ipinfor
	ipdot = ipinfor.at(0).split('.')	#split with .
	@ipmin = ipdot.at(ipdot.size-1)
	if ipinfor.size == 1
		@ipmax = @ipmin
	else
		@ipmax = ipinfor.at(1)
	end
	client_ip = ip.split('.')	#split with .
	if ipdot.at(0).to_i == client_ip.at(0).to_i && ipdot.at(1).to_i == client_ip.at(1).to_i
		if client_ip.at(2).to_i >= @ipmin.to_i && client_ip.at(2).to_i <= @ipmax.to_i
			return strIp['enname'],strIp['sourcetype']
		end
	end
    end
    return 'other','other'
end

#get http json
public
def httpPostMethod(uri)
    params = {}
    uri = URI.parse(uri)
    res = Net::HTTP.post_form(uri, params)
    return res.body
end

end # class LogStash::Filters::Test

input {
    file {
      type => "nginx_access_log"
      path => "/var/log/nginx/access.log"
    }
}
 
filter {

    grok {
      match => [ "message", "%{IPV4:remote_ip}, (%{IPV4:http_x_forwarded_for}|-) (%{IPV4:remote_user}|-) \[%{MONTHDAY:day}\/%{MONTH:month}\/%{YEAR}:%{HOUR:hour}:%{MINUTE:min}:%{SECOND:sec} %{ISO8601_TIMEZONE:tz}\] %{BASE16FLOAT:request_time} %{HOSTNAME:domain} \"%{WORD:httpmethod} %{URIPATH:uripath} %{URIPROTO:uriproto1}\/%{BASE16FLOAT:httpversion} %{URIPROTO:uriproto2}\" %{INT:http_status} - (%{INT:body_bytes_sent}|\-) (%{INT:bytes_sent}|\-) (%{INT:sent_http_content_length}|\-)  \"(%{INT:sent_http_content_Range}|\-)\"   \"(%{URI:http_refer}|\-)\" \"%{DATA:user_agent}\" (%{DATA:sent_http_x_cacche}|\-) (%{DATA:sent_http_content_type}|\-) (up_addr:%{URIHOST:upstream_urihost}|\-) (up_resp:%{DATA:request_time2}s|\-) (up_status:%{INT:upstream_status}|\-)" ]
    }
    date {
      match => ["timestamp","dd/MMM/YYYY:HH:mm:ss Z","dd/MMM/YYYY:HH:mm:ss ZZ","dd/MMM/YYYY:HH:mm:ss ZZZ","YYYY-MM-dd HH:mm:ss"]
      locale => "en"
    }
    if "_grokparsefailure" in [tags] { drop { } }
 
    test {
	     remote_ip => "remote_ip"
    } 
}
 
output {
  redis {
    host => '192.168.201.132'
    data_type => 'list'
    key => 'logstash'
  }
}