Logstash 过滤和解析 Dies 输出

Question

环境

• Ubuntu 16.04
• Lo​​gstash 5.2.1
• 弹性搜索 5.1

我已经配置了我们的 Deis 平台，可以毫无问题地将日志发送到我们的 Logstack 节点。但是，我对 Ruby 还是很陌生，而正则表达式并不是我的强项。

日志示例：

2017-02-15T14:55:24UTC deis-logspout[1]: 2017/02/15 14:55:24 routing all to udp://x.x.x.x:xxxx\n

Logstash 配置：

input {
    tcp {
        port => 5000
        type => syslog
        codec => plain
    }
    udp {
        port => 5000
        type => syslog
        codec => plain
    }
}

filter {
    json {
        source => "syslog_message"
    }
}

output {
    elasticsearch { hosts => ["foo.somehost"] }
}

Elasticsearch 输出：

"@timestamp" => 2017-02-15T14:55:24.408Z,
"@version" => "1",
"host" => "x.x.x.x",
"message" => "2017-02-15T14:55:24UTC deis-logspout[1]: 2017/02/15 14:55:24 routing all to udp://x.x.x.x:xxxx\n",
"type" => "json"

期望的结果：

"@timestamp" => 2017-02-15T14:55:24.408Z,
"@version" => "1",
"host" => "x.x.x.x",
"type" => "json"
"container" => "deis-logspout"
"severity level" => "Info"
"message" => "routing all to udp://x.x.x.x:xxxx\n"

如何将消息中的信息提取到各自的字段中？

Answer 1

很遗憾，您对自己尝试做的事情的假设略有偏差，但我们可以解决这个问题！

您为 JSON 创建了一个正则表达式，但您没有解析 JSON。您只是在解析一个混杂的 syslog 日志（请参阅source 中的 syslogStreamer），但实际上不是 syslog 格式（RFC 5424 或 3164）。 Logstash 之后提供 JSON 输出。

让我们分解消息，它成为您解析的源。关键是你必须从头到尾解析消息。

消息：

2017-02-15T14:55:24UTC deis-logspout[1]: 2017/02/15 14:55:24 routing all to udp://x.x.x.x:xxxx\n

• 2017-02-15T14:55:24UTC：时间戳是一种常见的 grok 模式。这主要遵循 TIMESTAMP_ISO8601，但不完全遵循。
• deis-logspout[1]：这将是您的日志源，您可以将其命名为容器。您可以使用 grok 模式 URIHOST。
• routing all to udp://x.x.x.x:xxxx\n：由于大多数日志的消息都包含在消息的末尾，因此您可以使用 grok 模式 GREEDYDATA，它相当于正则表达式中的 .*。李>
• 2017/02/15 14:55:24：另一个与常见的 grok 模式不匹配的时间戳（为什么？）。

使用 grok 过滤器，您可以将语法（正则表达式的抽象）映射到语义（您提取的值的名称）。例如%{URIHOST:container}

您会看到我对 grok 过滤器进行了一些修改，以使格式化工作。即使您不打算捕获结果，您也有匹配的文本部分。如果您无法更改时间戳的格式以匹配标准，请创建自定义模式。

配置：

input {
    tcp {
        port => 5000
        type => deis
    }
    udp {
        port => 5000
        type => deis
    }
}

filter {
    grok {
        match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}(UTC|CST|EST|PST) %{URIHOST:container}\[%{NUMBER}\]: %{YEAR}/%{MONTHNUM}/%{MONTHDAY} %{TIME} %{GREEDYDATA:msg}" }
    }
}

output {
    elasticsearch { hosts => ["foo.somehost"] }
}

输出：

{
    "container" => "deis-logspout",
    "msg" => "routing all to udp://x.x.x.x:xxxx",
    "@timestamp" => 2017-02-22T23:55:28.319Z,
    "port" => 62886,
    "@version" => "1",
    "host" => "10.0.2.2",
    "message" => "2017-02-15T14:55:24UTC deis-logspout[1]: 2017/02/15 14:55:24 routing all to udp://x.x.x.x:xxxx",
    "timestamp" => "2017-02-15T14:55:24"
    "type" => "deis"
}

您还可以更改项目以删除 @timestamp、@host 等，因为这些是由 Logstash 默认提供的。另一个建议是使用date filter 将找到的任何时间戳转换为可用格式（更适合搜索）。

根据日志格式，您可能需要稍微改变模式。我只有一个例子。这也保留了原始的完整消息，因为在 Logstash 中完成的任何字段操作都是破坏性的（它们会覆盖具有相同名称的字段的值）。

资源：

• Grok
• Grok Patterns
• Grok Debugger