如何在 Logstash 中获取部分 Filebeat 源文件名

Question

我有一个 Filebeat 实例（版本 7.5.0，在 Windows Server 上运行）监控本地文件夹中的日志文件，并将这些数据发送到 Logstash（版本 7.5.0，在 Docker 容器中运行）。在 Logstash 中，我想提取其中一个文件夹名称（最后一个）并将其添加为字段。

一个具体的例子是，从两个日志条目中，一个来自文件d:\\Logs\\Foo\\Bar\\lorem\\currentlog.txt，一个来自文件d:\\Logs\\Foo\\Bar\\ipsum\\currentlog.txt，我想分别提取值lorem和ipsum。

为此，我设置了以下（简化示例）：

input {
    pipeline { address => "test" }
}

filter {
    grok {
        match => { "source" => ".*\\\\.*\\\\(?<product>.*)\\\\.*" }
    }
}

output {
    stdout { codec => rubydebug }
}

我已经在几个地方（grockconstructor、grockdebug 和 rubular）测试了用于在源字段上查找匹配项（名为 product）的正则表达式，它们似乎都产生了所需的结果：我使用路径中最后一个文件夹的执行值获得了产品的命名匹配。

但是，当我使用上述管道配置运行 Logstash 时，它无法提取文件夹名称并将其值放在产品字段中。相反，我看到在logstash 输出中添加了一个标签，其值为grokparsefailure，表明我的grok 表达式有问题。但是我在上面引用的工具中的所有测试都表明我的表达没有问题......

完整的logstash输出如下所示：

{
    "@version" => "1",
    "tags" => [
        [0]"beats_input_codec_plain_applied",
        [1]"_grokparsefailure"
    ],
    "host" => {
        "name" => "test"
    },
    "message" => "Another line in the log",
    "agent" => {
        "id" => "e00d2f50-b10c-406a-a4fa-be381d15b869",
        "ephemeral_id" => "28dfe105-b936-40de-bc97-16c4a9196e30",
        "hostname" => "my-host",
        "name" => "test",
        "type" => "filebeat",
        "version" => "7.5.0"
    },
    "@timestamp" => 2019 - 12 - 16T14: 04: 09.064Z,
    "ecs" => {
        "version" => "1.1.0"
    },
    "log" => {
        "file" => {
            "path" => "d:\\Logs\\Foo\\Bar\\ipsum\\currentlog.txt"
        },
        "offset" => 21
    },
    "input" => {
        "type" => "log"
    }
}

我尝试将匹配更改为在 log.file.path 属性上，但这给了我相同的 _grokparsefailure 标记。

我也很确定这适用于 Filebeat/Logstash 的早期安装（可能是一两个主要版本），但我记不清了。

所以问题是：为什么 Logstash 不能从 Filebeat 源中提取文件夹名称？有没有办法进一步调试这个 grok 问题？

Answer 1

上述配置不起作用的原因是复合的，但我最终设法弄清楚了：

首先，没有来自 Filebeat 的 source 字段（我很确定以前有一些版本，但那是另一回事），这显然会导致 grok 过滤器不成功。

接下来，当我尝试深入了解log.file.path 字段时，我使用了错误的语法。访问嵌套字段的正确方法如下：[log][file][path]

最后，即使输出显示log.file.path 的值是"d:\\Logs\\Foo\\Bar\\ipsum\\currentlog.txt"，双反斜杠显然是在输出管道的某处添加的。因此，当我将正则表达式更改为匹配单反斜杠而不是双反斜杠时，它正确地从 "d:\Logs\Foo\Bar\ipsum\currentlog.txt" 中提取了 ipsum

因此，我的最终管道配置如下所示：

input {
    pipeline { address => "test" }
}

filter {
    grok {
        match => { "[log][file][path]" => ".*(\\|\/).*(\\|\/)(?<product>.*)(\\|\/).*"}
    }
}

output {
    stdout { codec => rubydebug }
}

现在我成功获取了提取到product 字段的路径中最后一个文件夹的名称，没有_grokparsefailure 标记。