基于message字段与appname创建索引的Logstash配置咨询

input {
    stdin{
        type => "stdin-type"
    }
    file{
        type => "json"
        path => [ "C:/prod/*.log", "C:/prod/*/**.log"]
        start_position => "beginning"
        tags => "prod"
    }
    file{
        type => "json"
        path => [ "C:/dev/*.log", "C:/dev/*/**.log"]
        start_position => "beginning"
        tags => "dev"
    }
}
filter {
    # 直接解析整行JSON格式的message字段，无需grok冗余处理
    json {
        source => "message"
        target => "payload"
        # 解析失败的日志加标签方便后续排查
        tag_on_failure => [ "_json_parse_failure" ]
    }

    # 处理appname字段，统一转小写适配ES索引命名规范
    if "_json_parse_failure" not in [tags] and [payload][appname] {
        mutate {
            add_field => { "appname" => "%{[payload][appname]}" }
            lowercase => [ "appname" ]
            # 不需要保留原始message字段可取消下方注释
            # remove_field => [ "message" ]
        }
    } else {
        # 解析失败或无appname时设置兜底值，避免生成非法索引
        mutate {
            add_field => { "appname" => "unknown" }
        }
    }

    date {
        match => [ "[payload][datetime]", "yyyy-MM-dd HH:mm:ss,SSS" ]
        target => "@timestamp"
    }
}
output {
    stdout {
        codec => rubydebug
    }
    elasticsearch {
        hosts => ["localhost:9200"]
        # 索引命名规则：环境标签-appname-logs，全小写符合ES规范
        index => "%{tags}-%{appname}-logs"
    }
}

配置说明 #

• 移除了原配置中冗余的grok规则：你的日志本身是标准整行JSON，直接通过json插件解析message字段即可，减少不必要的性能损耗
• 新增解析失败兜底逻辑：JSON解析失败或无appname字段时，自动将appname设为unknown，避免生成非法索引名称导致写入失败
• 统一appname字段为小写：适配Elasticsearch索引命名不能包含大写字母的规范，避免写入报错
• 索引按环境+应用拆分：后续可以针对不同应用单独设置索引生命周期、权限控制，查询效率也更高

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