logstash - 複数の Grok フィルターが最初のフィルター一致レコードを保存しない

Question

Postfix ログを解析するために Logstash を使用しています。私は主に、後置ログからバウンスされた電子メール ログを取得し、データベースに保存することに重点を置いています。

ログを取得するには、まずメッセージ ID に対応する postfix によって生成された 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
}

ご覧のとおり、1 つのログ ファイルから対応する 2 つの異なるログを見つけるために 2 つのパターンを使用しています。

ここで、タグに基づいて両方のパターンを区別したいと思います。そのため、次のように構成を変更しました。

  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」タグと最初のフィルタータグがあり、そのためそのレコードを保存していないことがわかりました。

これを修正するために何をする必要があるか誰か教えてもらえますか? 私は間違いを犯していますか？