logstash：想要在同一个过滤器中匹配“单个 grok 模式”和“多个 grok 模式”

Question

我的日志文件行如下所示。

[2020-07-10T10:00:04.979+00:00] [app_server1] [NOTIFICATION] [report-thread] [tid: 1346887] [userId: user10.id2] start-getchunk: Report=/Report Folder Path/Report name, TemplateName=Template1004, OutFormat=excel, Locale=en_US
[2020-07-10T10:00:25.085+00:00] [app_server2] [NOTIFICATION] [report-thread] [tid: 1346887] [userId: user1.id1] end-getchunk: Report=/Report Folder Path/Report name, TemplateName=Template2007, OutFormat=html, Locale=en_US
[2020-07-10T10:00:25.080+00:00] [app_server2] [NOTIFICATION] [report-thread] [tid: 1346887] [userId: user2.id1][
    Start report processing details
    -----------------------------------------------------
    Report path: /Report Folder Path/Report name
    Datamodel name: /Report Folder Path/Datamodel name
    User name: user2
    Output format: 1
    chunk size limit: 524288000
    -----------------------------------------------------
    End report processing details
]
[Log lines with some other patterns]

我想在我的 logstash 配置文件中使用以下类型的匹配模式来匹配不同的行。

(a) match => {  "message" => "\[%{TIMESTAMP_ISO8601:process_timstamp}\] \[%{WORD:app_server}] \[%{WORD:log_level}] \[] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\] (?<event_type>\sstart-getchunk): Report=(?<report_name>[^,]*), TemplateName=(?<template_name>[^,]*), OutFormat=(?<output_format>[^,]*), Locale=(?<locale>[^,]*)" }
(b) match => {  "message" => "\[%{TIMESTAMP_ISO8601:process_timstamp}\] \[%{WORD:app_server}] \[%{WORD:log_level}] \[] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\] (?<event_type>\send-getchunk): Report=(?<report_name>[^,]*), TemplateName=(?<template_name>[^,]*), OutFormat=(?<output_format>[^,]*), Locale=(?<locale>[^,]*)" }
(c) to match a multiline having this match
    break_on_match => false
    match => {
        "message => [
            "Report path: (?<report_path>[^,\r\n]*)",
            "Datamodel name: (?<datamodel_name>[^,\r\n]*)",
            "User name: (?<user_name>[^,\r\n]*)",
            (?<event_type>Start report processing details)
        ]
    }
(d) other n numbers of matches like the one mentioned in points (a) & (b) above

现在，假设我的配置中只有 (a)、(b) 和 (c)。

我的配置如下所示（“match =>”行的顺序发生了变化）。

filter{
    grok{
        break_on_match => false
        ## single grok pattern match
        match => {  "message" => "\[%{TIMESTAMP_ISO8601:process_timstamp}\] \[%{WORD:app_server}] \[%{WORD:log_level}] \[] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\] (?<event_type>\sstart-getchunk): Report=(?<report_name>[^,]*), TemplateName=(?<template_name>[^,]*), OutFormat=(?<output_format>[^,]*), Locale=(?<locale>[^,]*)" }
        match => {  "message" => "\[%{TIMESTAMP_ISO8601:process_timstamp}\] \[%{WORD:app_server}] \[%{WORD:log_level}] \[] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\] (?<event_type>\send-getchunk): Report=(?<report_name>[^,]*), TemplateName=(?<template_name>[^,]*), OutFormat=(?<output_format>[^,]*), Locale=(?<locale>[^,]*)" }
        ## multple grok pattern match
        match => {
            "message => [
                "Report path: (?<report_path>[^,\r\n]*)",
                "Datamodel name: (?<datamodel_name>[^,\r\n]*)",
                "User name: (?<user_name>[^,\r\n]*)",
                (?<event_type>Start report processing details)
            ]
        }
    }
}

现在问题陈述：- 配置仅匹配配置文件中提到的三个“match =>”中的最后一个的模式。 所以，如果在配置文件中

[A] the sequence of "match =>" lines are (a) -> (b) -> (c) (as shown in above config): it matches patterns for only (c).
[B] the sequence of "match =>" lines are (c) -> (b) -> (a): it matches patterns for only (a).
[C] the sequence of "match =>" lines are (c) -> (a) -> (b): it matches patterns for only (b).

但是，我希望在输入中匹配所有 (a)、(b)、(c)。

另外，我想保持 (a)、(b)、(c) 的匹配模式不变。 实际上，对于 (c)，我有一个像 (a) 和 (b) 这样的 grok 模式。那时配置试图匹配所有 (a)、(b) 和 (c)。但由于 (c) 出现“_groktimeout”错误，将模式更改为 (c) 中提到的模式 Ref。

为了解决这个问题，我尝试了

• 同一过滤器下的多个 grok{} 块{} 块 AND
• 多个过滤器{} 块在同一个配置文件中 通过将“（a），（b）”和“（c）”拆分为单独的块来解决问题。但在上述两种情况下，它并没有产生任何输出。

我尝试了另一个选项，如下所示。这可行，但这不是一个性能有效的选择。 因此在不使用“if”条件中的 grok 的情况下寻找更好、性能更高效的选项。

filter{
    grok{
        ## single grok pattern match
        match => {  "message" => "\[%{TIMESTAMP_ISO8601:process_timstamp}\] \[%{WORD:app_server}] \[%{WORD:log_level}] \[] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\] (?<event_type>\sstart-getchunk): Report=(?<report_name>[^,]*), TemplateName=(?<template_name>[^,]*), OutFormat=(?<output_format>[^,]*), Locale=(?<locale>[^,]*)" }
        match => {  "message" => "\[%{TIMESTAMP_ISO8601:process_timstamp}\] \[%{WORD:app_server}] \[%{WORD:log_level}] \[] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\] (?<event_type>\send-getchunk): Report=(?<report_name>[^,]*), TemplateName=(?<template_name>[^,]*), OutFormat=(?<output_format>[^,]*), Locale=(?<locale>[^,]*)" }
        match => {  "message" => "\[%{TIMESTAMP_ISO8601:process_timstamp}\] \[%{WORD:app_server}] \[%{WORD:log_level}] %{GREEDYDATA}(?<event_type>Start report processing details)"}
        if " Start report processing details" in [event_type]{
            grok {
                break_on_match => false
                ## multple grok pattern match
                match => {
                    "message => [
                        "Report path: (?<report_path>[^,\r\n]*)",
                        "Datamodel name: (?<datamodel_name>[^,\r\n]*)",
                        "User name: (?<user_name>[^,\r\n]*)",
                        (?<event_type>Start report processing details)
                    ]
                }
            }
        }
    }
}

我的预期输出如下所示，为此我删除了配置文件中的几个字段。

{"tags":["reportlog"],"userId":"user10.id2","process_timestamp":"2020-07-10T10:00:25.085+00:00","thread_type":"report-thread","template_name":"Template2007","log_level":"NOTIFICATION","event_type":"end-getchunk","output_format":"html","tid":"199163","app_server":"app_server2","report_name":"/Report Folder Path/Report name"}
{"tags":["reportlog"],"userId":"user1.id1","process_timestamp":"2020-07-10T10:00:04.979+00:00","thread_type":"report-thread","template_name":"Template1004","log_level":"NOTIFICATION","event_type":"start-getchunk","output_format":"excel","tid":"800760","app_server":"app_server1","report_name":"/Report Folder Path/Report name"}
{"tags":["multiline","reportlog"],"userId":["user2.id1"],"report_path":"/Report Folder Path/Report name", "process_timestamp":["2020-07-10T10:00:25.080+00:00"],"datamodel_name":"/Report Folder Path/Datamodel name","thread_type":"report-thread","log_level":"NOTIFICATION","event_type":"Start report processing details","tid":["812409"],"app_server":"app_server2"}

。 [1]：https://discuss.elastic.co/t/getting-groktimeout-error-for-a-particular-filter-sometimes-need-help-optimizing-the-filter/246342

Answer 1

前两条消息之间的唯一区别是一条具有字符串start-getchunck，另一条具有字符串end-getchunck，并且您将此字符串保存到两种消息的同一字段中，因此您的grok模式a 和 b 基本相同。

以下模式将匹配这两种类型的消息。

\[%{TIMESTAMP_ISO8601:process_timestamp}\] \[%{WORD:app_server}\] \[%{WORD:log_level}] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\] %{DATA:event_type}: Report=(?<report_name>[^,]*), TemplateName=(?<template_name>[^,]*), OutFormat=(?<output_format>[^,]*), Locale=(?<locale>[^,]*)

不要使用(?<event_type>\sstart-getchunk): 和(?<event_type>\send-getchunk):，而是使用%{DATA:event_type}:。

对于您的多行日志，由于您没有共享您的 input 配置，我使用以下多行代码配置来重现。

codec => multiline {
    pattern => "^\["
    negate => true
    what => "previous"
}

您用于此日志的grok 配置错误。

有这个：

grok {
    match => { 
        "message" => [
            "Report path: (?<report_path>[^,\r\n]*)",
            "Datamodel name: (?<datamodel_name>[^,\r\n]*)"
        ]
    }
} 

是否有这样的配置：

grok {
    match => { "message" => "Report path: (?<report_path>[^,\r\n]*)" }
    match => { "message" => "Datamodel name: (?<datamodel_name>[^,\r\n]*)" }
} 

你需要的是一个匹配你消息中所有字段的模式，下面的模式将匹配你的多行日志。

\[%{TIMESTAMP_ISO8601:process_timestamp}\] \[%{WORD:app_server}\] \[%{WORD:log_level}] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\]%{DATA}(?<event_type>Start report processing details)%{DATA}Report path: (?<report_path>[^,\r\n]*)%{DATA}Datamodel name: (?<datamodel_name>[^,\r\n]*)%{DATA}User name: (?<user_name>[^,\r\n]*)"

另外，您需要将break_on_match 设置为true，如果您的线路已经有模式匹配，则无需针对其他模式进行测试，它只会为您的管道添加更多处理。

以下filter 将匹配您的所有示例行。

filter{
    grok {
        break_on_match => true
        match => {  "message" => "^\[%{TIMESTAMP_ISO8601:process_timestamp}\] \[%{WORD:app_server}\] \[%{WORD:log_level}] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\] %{DATA:event_type}: Report=(?<report_name>[^,]*), TemplateName=(?<template_name>[^,]*), OutFormat=(?<output_format>[^,]*), Locale=(?<locale>[^,]*)" }
        match => {  "message" => "^\[%{TIMESTAMP_ISO8601:process_timestamp}\] \[%{WORD:app_server}\] \[%{WORD:log_level}] \[%{DATA:thread_type}\] \[tid: %{NUMBER:tid}\] \[userId: %{DATA:userId}\]%{DATA}(?<event_type>Start report processing details)%{DATA}Report path: (?<report_path>[^,\r\n]*)%{DATA}Datamodel name: (?<datamodel_name>[^,\r\n]*)%{DATA}User name: (?<user_name>[^,\r\n]*)" }
    }
}

模式中的^是为了锚定你的消息的位置，如果由于某种原因你的消息没有以[开头，grok甚至不会尝试解析它。

由于grok 依赖于正则表达式，它可能会出现性能问题，如果您仍然遇到超时错误，最好改变您的方法并开始使用条件来解析一些消息。

您可以尝试使用一个grok 解析所有公共字段，将消息的不同部分保存在另一个字段中并使用条件将该字段指向正确的grok，这比使用一个@987654346 执行得更好@ 试图匹配所有内容。