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"]
    }
}

我原来的答案是：

您需要使用自定义过滤器（不能从ruby代码过滤器调用yield，这是生成新事件所需的）。

像这样的东西（放入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 {} }放入配置文件中。

Answer 2

为了将来参考并基于@alcanzar答案，now possible做这样的事情：

  ruby {
    code => "
      # somefield is an array
      array = event.get('somefield')

      # drop the current event (this was my use case, I didn't need the feeding event)
      event.cancel

      # iterate over to construct new events
      array.each { |a|

        # creates a new logstash event
        generated = LogStash::Event.new({ 'foo' => 'something' })

        # puts the event in the pipeline queue
        new_event_block.call(generated)
      }
    "
  }