Logstash动态拆分事件

Question

有没有办法将logstash(1.4.2)事件拆分成多个其他事件？

我的输入如下:

{ "parts" => ["one", "two"],
  "timestamp" => "2014-09-27T12:29:17.601Z"
  "one.key=> "1", "one.value"=>"foo",
  "two.key" => "2", "two.value"=>"bar"
}

我想创建两个包含以下内容的事件:

{ "key" => "1", "value" => "foo", "timestamp" => "2014-09-27T12:29:17.601Z" }
{ "key" => "2", "value" => "bar", "timestamp" => "2014-09-27T12:29:17.601Z" }

问题是我无法知道实际的"部件"......

谢谢你的帮助 :)

Answer 1

更新一个非常旧的答案,因为有更好的方法可以在较新版本的logstash中执行此操作,而无需使用自定义筛选器.

您可以使用ruby过滤器和拆分过滤器来执行此操作:

filter {
    ruby {
        code => '
            arrayOfEvents = Array.new()
            parts = event.get("parts")
            timestamp = event.get("timestamp")
            parts.each { |part|
                arrayOfEvents.push({ 
                    "key" => event.get("#{part}.key"), 
                    "value" => event.get("#{part}.value"),
                    "timestamp" => timestamp
                })
                event.remove("#{part}.key")
                event.remove("#{part}.value")
            }
            puts arrayOfEvents

            event.remove("parts")
            event.set("event",arrayOfEvents)
        '
    }
    split {
        field => 'event'
    }
    mutate {
        rename => {
            "[event][key]" => "key"
            "[event][value]" => "value"
            "[event][timestamp]" => "timestamp"
        }
        remove_field => ["event"]
    }
}

我原来的答案是:

您需要使用自定义过滤器(您无法yield从ruby代码过滤器调用,这是生成新事件所需的).

像这样的东西(放入lib/logstash/filters/custom_split.rb):

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

# custom code to break up an event into multiple
class LogStash::Filters::CustomSplit < LogStash::Filters::Base
  config_name "custom_split"
  milestone 1

  public
  def register
    # Nothing
  end # def register

  public
  def filter(event)
    return unless filter?(event)
    if event["parts"].is_a?(Array)
      event["parts"].each do |key|
        e =  LogStash::Event.new("timestamp" => event["timestamp"],
          "key" => event["#{key}.key"],
          "value" => event["#{key}.value"])
        yield e
      end
      event.cancel
    end
  end
end

然后只需filter { custom_split {} }输入您的配置文件.