Logstash如何根据kv插件解析的字段值过滤丢弃指定CEF日志

API 探路者

2026-10-1

问题根因与解决方案

核心问题有两点：

过滤规则的字段名配置错误
你当前drop规则判断的字段是[FTNTFGTsrcintfrole_s]，但你提供的解析后日志样例中，对应实际生成的字段名为FTNTFGTsrcintfrole，没有末尾的_s后缀，导致条件永远无法匹配，drop规则自然不生效。
原有配置存在执行顺序问题，你提供的日志中已经出现_mutate_error标签，且message字段被错误赋值为%{tmp_message}，需要同步优化mutate操作的逻辑避免异常。

调整后的完整filter配置

filter {
    if [type] == "cef" {
        # 先单独保存原始message
        mutate {
            add_field => { "tmp_message" => "%{message}" }
        }
        mutate {
            split => ["message", "|"]
            add_field => { 
                "version" => "%{[message][0]}"
                "device_vendor" => "%{[message][1]}"
                "device_product" => "%{[message][2]}"
                "device_version" => "%{[message][3]}"
                "sig_id" => "%{[message][4]}"
                "sig_name" => "%{[message][5]}"
                "sig_severity" => "%{[message][6]}"
            }
        }
        kv {
            field_split => " "
            trim_value => "<>\[\],"
        }
        # 恢复原始message并清理临时字段
        mutate {
            replace => { "message" => "%{tmp_message}" }
            remove_field => [ "tmp_message" ]
        }
    }
    # 替换为实际存在的字段名
    if [FTNTFGTsrcintfrole] == "wan" {
        drop {}
    }
}