一个简单定制的Logstash filter

官方参考：https://www.elastic.co/guide/en/logstash/current/how_to_write_a_logstashfilter_plugin.html

官方实例：https://github.com/logstash-plugins/logstash-filter-example/

网上guide：http://www.cnblogs.com/xing901022/p/5259750.html

插件plugin命名：logstash-filter-seq

通过测试，实际只须要最少两个文件：

1 logstash-filter-seq.gemspec

2 lib\logstash\filters\seq.rb

内容分别以下：

Gem::Specification.new do |s|
  s.name = 'logstash-filter-seq'
  s.version         = '1.0.0'
  s.licenses = ['Apache License (2.0)']
  s.summary = "This seq filter adds sequence to each document during filtering."
  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 = ["zhaoxp"]
  s.email = 'zhaoxp2@lenovo.com'
  s.homepage = "http://www.elastic.co/guide/en/logstash/current/index.html"
  s.require_paths = ["lib"]

  # Files
  s.files = Dir['lib/**/*','spec/**/*','vendor/**/*','*.gemspec','*.md']
   # 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", ">= 2.0.0", "< 3.0.0"
  #s.add_development_dependency 'logstash-devutils'
end

 seq.rb

# encoding: utf-8
require "logstash/filters/base"
require "logstash/namespace"

# This example filter will replace the contents of the default 
# message field with whatever you specify in the configuration.
#
# It is only intended to be used as an example.
class LogStash::Filters::Seq < LogStash::Filters::Base

  # Setting the config_name here is required. This is how you
  # configure this filter from your Logstash config.
  #
  # filter {
  #   seq {
  #     message => "My message..."
  #   }
  # }
  #
  config_name "seq"
  
  # Replace the message with this value.
  #config :message, :validate => :string, :default => "Hello World!"
  config :seqname, :validate => :string, :default => "seq"
  

  public
  def register
    # Add instance variables 
    @lineindex=0
  end # def register

  public
  def filter(event)

#    if @message
      # Replace the event message with our message as configured in the
      # config file.
#      event["message"] = @message
#    end

    # filter_matched should go in the last line of our successful code
    #filter_matched(event)
    @lineindex=@lineindex+1
	event[seqname]=@lineindex
  end # def filter
end # class LogStash::Filters::Seq

部署方法：

1 将logstash-filter-seq目录放入logstash下vendor\bundle\jruby\1.9\gems中。

2 修改logstash下的Gemfile文件，增长一行：

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

 

使用方法：

filter{
  seq{
    seqname=> "testseq"
  }
}

这样的话，使用这个filter最终会产生一个属性名为testseq，值为数值而且逐个加一的属性值。

当初使用这个filter的缘由是由于如下的filter达不到目的。由于这是以时间值做为属性值，若是logstash处理速度过快，就会出现相邻两条记录的daytag值同样。

ruby {
  code => "event['daytag'] = event.timestamp.time.localtime.strftime('%Y-%m-%d');event['seq'] = Time.now.strftime('%Y%m%d%H%M%S%L').to_i"
#  code => "event['daytag'] = event.timestamp.time.localtime.strftime('%Y-%m-%d')"
}

 

这个例子太多简单，实际要考虑到的问题还有：

若是logstash重启，那么sequence会从1开始从新计算，因此有两种考虑方案：

1 再增长一个属性，来表示sequence开始计时时的时间。

2 在plugin中增长一个属性，表示一个存放文件的路径。这个文件中记录了上次sequence最后的值，这样涉及到一个问题是sequence刷新时间。这参考一下logstash-input-file这个plugin。