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

我有一个 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 and ipsum分别。

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

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

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

output {
    stdout { codec => rubydebug }
}

我已经测试了用于查找匹配项的正则表达式（名为product）在源字段的几个地方（都格罗克构造函数 http://grokconstructor.appspot.com/do/match, 格罗克调试 https://grokdebug.herokuapp.com/ and rubular https://rubular.com/），它们似乎都产生了所需的结果：我得到了产品的命名匹配与路径中最后一个文件夹的执行值。

但是，当我使用上述管道配置运行 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 tag.

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

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

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

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

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

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

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

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

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

output {
    stdout { codec => rubydebug }
}

现在我成功获取了提取到的路径中最后一个文件夹的名称product场，没有_grokparsefailure tag.