Logstash三类消息解析后写入Elasticsearch不同索引配置失败问题

• 核心配置错误：错误使用in操作符判断字符串包含子串。Logstash语法规则中，in操作符仅适用于判断某个元素是否存在于数组类型字段中，若要判断字符串字段包含特定子串，必须使用正则匹配符=~。
• 错误导致的连锁问题：三类日志的前两个条件分支全部判断失败，所有消息都进入最后的else分支的grok规则，其中前两类日志无法匹配该分支的grok pattern，导致msgType字段添加失败，最终在output分支判断时全部命中兜底的default索引。

你可以直接使用修改后的完整配置，同时做了冗余代码优化：

input {
    beats {
        port => 5044
    }
}

filter {
    # 改用正则匹配判断子串存在
    if [message] =~ /\[S=/ and [message] =~ /\[ID=/ {
        grok {
            match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}%{SPACE}%{IP:IP}%{SPACE}\[%{DATA:SeqNo}\]%{SPACE}\[%{DATA:ID}\]%{SPACE}%{GREEDYDATA:event_detail}" }
            add_field => {
                "msgType" => "message1"
            }
            tag_on_failure => ["_grok_failure_msg1"]
        }
    }
    else if [message] =~ /\[S=/ {
        grok {
            match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}%{SPACE}%{IP:IP}%{SPACE}\[%{DATA:SeqNo}\]%{SPACE}%{GREEDYDATA:event_detail}" }
            add_field => {
                "msgType" => "message2"
            }
            tag_on_failure => ["_grok_failure_msg2"]
        }
    } else {
        grok {
            match => { "message" => "%{TIMESTAMP_ISO8601:timestamp}%{SPACE}%{IP:IP}%{SPACE}\[%{DATA:anotherDate}\]%{SPACE}%{GREEDYDATA:event_detail}" }
            add_field => {
                "msgType" => "message3"
            }
            tag_on_failure => ["_grok_failure_msg3"]
        }
    }
}

output {
    elasticsearch {
        hosts => "10.133.11.23:9200"
        user => "elastic"
        password => "changeme"
        ecs_compatibility => disabled
        index => "%{msgType}-%{+YYYY.MM.dd}"
    }
}

主要修改说明 #

1. 修正条件判断逻辑：用正则匹配符=~替代错误的in操作符，正则中[是特殊字符需要加反斜杠转义。
2. 保留原始日志：将grok中原覆盖message字段的逻辑改为赋值到新字段event_detail，避免丢失原始日志影响后续排查。
3. 新增排障标识：为每个grok添加tag_on_failure配置，匹配失败的日志会打上对应标签，方便快速定位异常日志。
4. 简化output配置：直接通过%{msgType}变量引用字段值生成索引名，删除了冗余的重复ES配置块。

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