Logstash中查询keyword类型message字段未达预期结果问题

你当前的核心误区是误用了Logstash的elasticsearch filter插件——这个插件的作用是从Elasticsearch中查询关联数据来增强当前事件，而非过滤正在处理的事件流。你要实现的是"过滤当前事件中message字段包含'My Alert'的条目"，根本不需要调用ES查询，直接在Logstash中处理即可。

正确配置方案 #

方案1：用条件判断过滤并提取字段 #

如果只是需要筛选出包含"My Alert"的事件，同时把message存入[@metadata][message]，可以直接用if条件：

filter {
    mutate {
        add_field => {"secret" => "<my_secret>"}
    }

    # 筛选message包含"My Alert"的事件
    if "My Alert" in [message] {
        mutate {
            add_field => {"[@metadata][message]" => "%{message}"}
        }
        # 这里添加你的grok处理逻辑
        grok {
            match => {"[@metadata][message]" => "你的grok模式"}
        }
    } else {
        # 丢弃不匹配的事件，如果需要的话
        drop {}
    }
}

方案2：用grok直接匹配提取 #

如果你的grok模式本身需要匹配包含"My Alert"的内容，可以直接用grok同时完成匹配和提取：

filter {
    mutate {
        add_field => {"secret" => "<my_secret>"}
    }

    grok {
        # 匹配包含"My Alert"的message，同时提取所需字段
        match => {"message" => "My Alert %{DATA:extract_field}"}
        # 匹配成功后将原message存入metadata
        add_field => {"[@metadata][message]" => "%{message}"}
        # 不匹配的事件直接丢弃（可选）
        tag_on_failure => []
        drop_if_match => false # 如果需要保留不匹配的，设为false；否则设为true
    }
}

为什么之前的配置无效 #

你用elasticsearch filter插件时，它会拿着配置的query去指定ES索引中查询其他数据，然后把查询结果关联到当前处理的事件上——这和你当前要过滤自己的事件流完全是两回事。哪怕你写的query能在ES中查到数据，这些数据也只是被附加到当前事件，不会筛选掉当前不符合条件的事件。

内容的提问来源于stack exchange，提问作者manuel