多个 Grok 过滤器不存储第一个过滤器匹配记录

我正在使用 Logstash 来解析 postfix 日志。我主要关注从后缀日志获取退回的电子邮件日志，并将其存储在数据库中。

为了获取日志，首先我需要找到 postfix 生成的与我的消息 ID 相对应的 ID，然后使用该 ID，我需要查找电子邮件的状态。对于以下配置，我能够获取日志。

grok {
       patterns_dir => "patterns"
       match => [
            "message", "%{SYSLOGBASE} %{POSTFIXCLEANUP}",
            "message", "%{SYSLOGBASE} %{POSTFIXBOUNCE}"
        ]
        named_captures_only => true
    }

我使用以下 if 条件来存储与模式匹配的日志：

if "_grokparsefailure" not in [tags] {
   #database call
}

正如您所看到的，我使用两种模式从一个日志文件中查找相应的两个不同日志。

现在，我想根据标签区分这两种模式。所以我修改了我的配置如下：

  grok {
       patterns_dir => "patterns"
       match => [
            "message", "%{SYSLOGBASE} %{POSTFIXBOUNCE}"
        ]
        add_tag => ["BOUNCED"]
        remove_tag => ["_grokparsefailure"]
        named_captures_only => true
    }

    grok {
       patterns_dir => "patterns"
       match => [
            "message", "%{SYSLOGBASE} %{POSTFIXCLEANUP}"            
        ]
        add_tag => ["INTIALIZATION"]
        remove_tag => ["_grokparsefailure"]
        named_captures_only => true
    }

现在，它仅存储 %{POSTFIXCLEANUP} 模式日志。如果我颠倒顺序，它只存储 %{POSTFIXBOUNCE} 模式。

因此，在删除该 if 条件后，我发现从第一个过滤器解析的消息具有“_grokparsefailure”标签和第一个过滤器标签，因此它不存储该记录。

谁能告诉我需要做什么来纠正这个问题？我有什么错误吗？

您需要保护第二个 grok 块——即，如果第一个块成功，则不要执行它。

if ("BOUNCED" not in [tags]) {
  grok {
    patterns_dir => "patterns"
    match => [
        "message", "%{SYSLOGBASE} %{POSTFIXCLEANUP}"            
    ]
    add_tag => ["INTIALIZATION"]
    remove_tag => ["_grokparsefailure"]
    named_captures_only => true
  }
}