针对ELK Stack的Grok条件解析

如何解决针对ELK Stack的Grok条件解析

我有这种日志：

2020-09-02 14:29:22,854 [http-something] [ERROR] JavaClass(JavaLine) - [6652942]: Error message with no stack trace
2020-09-02 14:29:08,976 [http-something] [INFO] JavaClass(JavaLine) - [6791732]: Some message
2020-09-02 14:29:09,116 [http-something] [ERROR] JavaClass(JavaLine) - [6791732]: Error message with stack trace
JavaException: This is not going well
    at JavaClass
    at JavaClass
    at JavaClass
    at JavaClass
    at JavaClass
Caused by: JavaClass: This is a problem
    at JavaClass
    at JavaClass
    at JavaClass
    at JavaClass
    ... 48 more

并且我使用此过滤器在Kibana上添加了更具可读性的日志：

filter {

    # INFO and ERROR
    grok {
        tag_on_failure => ["_stackTraceFailure"]
        match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}%{SPACE}(\[%{DATA:thread}\])?%{SPACE}\[%{LOGLEVEL:log_level}\]%{SPACE}%{GREEDYDATA}%{SPACE}\-%{SPACE}%{GREEDYDATA:action}" }
        overwrite => [ "message" ]
    }

    # JAVA ERROR
    if ("_stackTraceFailure" in [tags]) {
        grok { 
            tag_on_failure => ["_grokParseFailure"]
            match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}%{SPACE}(\[%{DATA:thread}\])?%{SPACE}\[%{LOGLEVEL:log_level}\]%{SPACE}%{GREEDYDATA}%{SPACE}\-%{SPACE}%{DATA:issue}(\r|\n)+(?m)%{GREEDYDATA:stack-trace}" }
            overwrite => [ "message" ]
            remove_tag => "_stackTraceFailure"
        }
    }
}

问题在于第一个模式匹配所有内容，将所有堆栈跟踪（如果有）都放在 action 标记中，导致第二个模式永远无法使用。我知道这个问题是由GREEDYDATA引起的，但是我对regex并不真正了解，也没有找到解决方案来完成我想要的事情。

我不想交换模式的位置，因为INFO和ERROR（无堆栈跟踪）更为常见，因此我需要一种在多行日志或其他任何情况下使第一个失败的方法如果存在某种堆栈跟踪，则使第一个失败。从目前为止我能做得到吗？

解决方法

您需要在使用怪胎之前使用条件句。您可以使用条件过滤器来过滤整个邮件，并使用两个不同的grok过滤器，也可以将第一个grok过滤器保持不变，并使用条件过滤器来仅解析action字段，我建议第二种选择。

在两种情况下，您都需要根据多行消息中仅存在的内容（例如"at JavaClass"字符串）进行条件过滤。

所以您需要这样的东西：

if "at JavaClass" not in [message] {
  grok { your first grok }
} else {
    grok { your second grok }
}

如果您想保留第一个小技巧，而使用第二个小技巧仅解析action字段，则将是这样。

if "at JavaClass" in [action] {
    grok {
        tag_on_failure => ["_grokParseFailure"]
        match => { "action" => "%{DATA:issue}(\r|\n)+(?m)%{GREEDYDATA:stack-trace}" }
    }
}

您没有说如何收集日志，如果您使用文件输入或日志输入中带有multiline编码的logstash，则还可以基于标签进行过滤，因为您将有一个名为multiline记录您的日志。