一个简单定制的Logstash filter

摘要: 自己写一个logstash filter，实现对document增加一个sequence。

官方参考：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

内容分别如下：

[code=plain]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

[code=language-ruby]# 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文件，增加一行：

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

使用方法：

[code=plain]filter{
seq{
seqname=> "testseq"
}
}

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

当初使用这个filter的原因是因为以下的filter达不到目的。因为这是以时间值作为属性值，如果logstash处理速度过快，就会出现相邻两条记录的daytag值一样。

[code=plain]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。