Logstash 解析包含多个日志条目的 xml 文档

Question

我目前正在评估 logstash 和 elasticsearch 是否对我们的用例有用。我拥有的是一个包含多个条目的日志文件，其形式为

<root>
    <entry>
        <fieldx>...</fieldx>
        <fieldy>...</fieldy>
        <fieldz>...</fieldz>
        ...
        <fieldarray>
            <fielda>...</fielda>
            <fielda>...</fielda>
            ...
        </fieldarray>
    </entry>
    <entry>
    ...
    </entry>
    ...
<root>

每个entry元素将包含一个日志事件。（如果您有兴趣，该文件实际上是一个 Tempo Timesheets（一个 Atlassian JIRA 插件）工作日志导出。）

是否可以在不编写自己的编解码器的情况下将此类文件转换为多个日志事件？

Answer 1

好的，我找到了一个对我有用的解决方案。该解决方案的最大问题是 XML 插件......不是很不稳定，但要么记录不充分且有缺陷，要么记录不充分且不正确。

TLDR

bash命令行：

gzcat -d file.xml.gz | tr -d "\n\r" | xmllint --format - | logstash -f logstash-csv.conf

Logstash 配置：

input {
    stdin {}
}

filter {
    # add all lines that have more indentation than double-space to the previous line
    multiline {
        pattern => "^\s\s(\s\s|\<\/entry\>)"
        what => previous
    }
    # multiline filter adds the tag "multiline" only to lines spanning multiple lines
    # We _only_ want those here.
    if "multiline" in [tags] {
        # Add the encoding line here. Could in theory extract this from the
        # first line with a clever filter. Not worth the effort at the moment.
        mutate {
            replace => ["message",'<?xml version="1.0" encoding="UTF-8" ?>%{message}']
        }
        # This filter exports the hierarchy into the field "entry". This will
        # create a very deep structure that elasticsearch does not really like.
        # Which is why I used add_field to flatten it.
        xml {
            target => entry
            source => message
            add_field => {
                fieldx         => "%{[entry][fieldx]}"
                fieldy         => "%{[entry][fieldy]}"
                fieldz         => "%{[entry][fieldz]}"
                # With deeper nested fields, the xml converter actually creates
                # an array containing hashes, which is why you need the [0]
                # -- took me ages to find out.
                fielda         => "%{[entry][fieldarray][0][fielda]}"
                fieldb         => "%{[entry][fieldarray][0][fieldb]}"
                fieldc         => "%{[entry][fieldarray][0][fieldc]}"
            }
        }
        # Remove the intermediate fields before output. "message" contains the
        # original message (XML). You may or may-not want to keep that.
        mutate {
            remove_field => ["message"]
            remove_field => ["entry"]
        }
    }
}

output {
    ...
}

详细的

我的解决方案有效，因为至少在entry级别之前，我的 XML 输入非常统一，因此可以通过某种模式匹配来处理。

由于导出基本上是一行非常长的 XML，并且 logstash xml 插件基本上只适用于包含 XML 数据的字段（读取：行中的列），我不得不将数据更改为更有用的格式。

外壳：准备文件

• gzcat -d file.xml.gz |: 数据太多了——显然你可以跳过它

• tr -d "\n\r" |：删除 XML 元素内的换行符：某些元素可以包含换行符作为字符数据。下一步要求将这些删除或以某种方式编码。即使它假设此时您将所有 XML 代码都放在一个大行中，但此命令是否删除元素之间的任何空格都没有关系

• xmllint --format - |: 用 xmllint 格式化 XML（libxml 自带）

  这里单个巨大的 XML ( <root><entry><fieldx>...</fieldx></entry></root>)意大利面条行格式正确：

  <root>
    <entry>
      <fieldx>...</fieldx>
      <fieldy>...</fieldy>
      <fieldz>...</fieldz>
      <fieldarray>
        <fielda>...</fielda>
        <fieldb>...</fieldb>
        ...
      </fieldarray>
    </entry>
    <entry>
      ...
    </entry>
    ...
  </root>
  

日志存储

logstash -f logstash-csv.conf

（.conf在 TL;DR 部分查看文件的完整内容。）

在这里，multiline过滤器起作用了。它可以将多行合并为一条日志消息。这就是为什么需要格式化的原因xmllint：

filter {
    # add all lines that have more indentation than double-space to the previous line
    multiline {
        pattern => "^\s\s(\s\s|\<\/entry\>)"
        what => previous
    }
}

这基本上是说每行的缩进超过两个空格（或者 is </entry>/ xmllint 默认情况下使用两个空格进行缩进）都属于前一行。这也意味着字符数据不能包含换行符（tr在 shell 中被剥离）并且 xml 必须被规范化 (xmllint)