Logstash，从多个文档中的 xml 文件拆分事件，保留来自根标签的信息

Question

我的问题： 我有 XML 文件，其中包含我想使用 Logstash 解析的事件，以便在之后使用 Kibana 请求它。我想在每个事件中保留 ROOT 标签中的所有信息。

输入看起来像：

<?xml version="1.0" encoding="UTF-8"?>
<ROOT number="34">
  <EVENTLIST>
    <EVENT name="hey"/>
    <EVENT name="you"/>
  </EVENTLIST>
</ROOT>

我想要的，两个这样的文件：

{
  "number":"34"
  "name": "Hey"
}
{
  "number":"34"
  "name": "you"
}

Logstash 配置：

input {
  stdin { }
}
filter {
  xml {
    store_xml => "false"
    source => "message"
    target => "EVENT"
    xpath => [
      "/ROOT/@number","number",
      "/ROOT/EVENTLIST/EVENT/@name","name"
    ]
  }
}
output { elasticsearch { host => localhost } stdout { codec => rubydebug } }

没用。我得到了什么：

{
  "number" : ["34"]
  "name":["hey,"you""]
}

我按照这个帖子的解决方案：https://serverfault.com/questions/615196/logstash-parsing-xml-document-containing-multiple-log-entries

但我的问题仍然存在，我丢失了根标签中的信息。

解决方案之一可能是使用一些 ruby​​ 过滤器来处理它，但我不知道 ruby​​。 另一种是在发送到elasticsearch之前使用一些java编程将XML转换为JSON...

有什么想法可以解决这个问题，还是我必须学习 ruby​​？

Answer 1

如果你的结构和你展示的一样简单，你可以使用我写的memorize插件。

您的配置将如下所示：

filter {
  if ([message] =~ /<ROOT/) {
    grok {
      match => [ "message", 
        'number="(?<number>\d+)" number2="(?<number1>\d+)"'
      ] 
    }
  } else if ([message] =~ /<EVENT /) {
    grok { 
      match => [ "message", 'name="(?<name>[^"]+)"']
    }
  }
  memorize {
    fields => ["number","number1"]
  }
  if ([message] !~ /<EVENT /) {
    drop {}
  } else {
    mutate { remove_field => ["message"] }
  }
}

我的示例显示了根据下面的 cmets 在 ROOT 元素中查找多个内容。这是支持记忆多个字段的插件版本：

# encoding: utf-8
require "logstash/filters/base"
require "logstash/namespace"
require "set"
#
# This filter will look for fields from an event and record the last value
# of them.  If any are not present, their last value will be added to the
# event
#
# The config looks like this:
#
#     filter {
#       memorize {
#         fields => ["time"]
#         default => { "time" => "00:00:00.000" }
#       }
#     }
#
# The `fields` is an array of the field NAMES that you want to memorize
# The `default` is a map of field names to field values that you want
# to use if the field isn't present and has no memorized value (optional)

class LogStash::Filters::Memorize < LogStash::Filters::Base

  config_name "memorize"
  milestone 2

  # An array of the field names to to memorize
  config :fields, :validate => :array, :required => true
  # a map for default values to use if its not seen before we need it
  config :default, :validate => :hash, :required => false

  # The stream identity is how the filter determines which stream an
  # event belongs to. See the multiline plugin if you want more details on how
  # this might work
  config :stream_identity , :validate => :string, :default => "%{host}.%{path}.%{type}"

  public
  def initialize(config = {})
    super

    @threadsafe = false

    # This filter needs to keep state.
    @memorized = Hash.new
  end # def initialize

  public
  def register
    # nothing needed
  end # def register

  public
  def filter(event)
    return unless filter?(event)

    any = false
    @fields.each do |field|
      if event[field].nil?
    map = @memorized[@stream_identity]
        val = map.nil? ? nil : map[field]
        if val.nil?
          val = @default.nil? ? nil : @default[field]
        end
    if !val.nil?
          event[field] = val
          any = true
    end
      else
        map = @memorized[@stream_identity]
    if map.nil?
          map = @memorized[@stream_identity] = Hash.new
    end
    val = event[field]
    map[field] = event[field]
      end #if
      if any
        filter_matched(event)
      end
    end #field.each
  end
end

对于logstash 1.5 及更高版本，此插件可通过以下方式安装

bin/plugin install logstash-filter-memorize

Answer 2

试试这个过滤器：

filter {
  xml {
    source => "message"
    target => "xml_content"
  }
  split {
    field => "xml_content[EVENTLIST]"
  }
  split {
    field => "xml_content[EVENTLIST][EVENT]"
  }
  mutate {
    add_field => { "number" => "%{xml_content[number]}" }
    add_field => { "name" => "%{xml_content[EVENTLIST][EVENT][name]}" }
    remove_field => ['xml_content', 'message', 'path']
  }
}
output {
  stdout {
    codec => rubydebug
  }
}

它返回这个事件：

{
        "number" => "34",
    "@timestamp" => 2016-12-23T12:01:17.888Z,
      "@version" => "1",
          "host" => "xubuntu",
          "name" => "hey"
    ]
}
{
        "number" => "34",
    "@timestamp" => 2016-12-23T12:01:17.888Z,
      "@version" => "1",
          "host" => "xubuntu",
          "name" => "you"
    ]
}