如何配置Logstash Elasticsearch输出，避免COMPLETE状态被覆盖？

要实现「仅当文档状态不为COMPLETE时允许更新，状态为COMPLETE时禁止覆盖」的需求，需要修改Logstash的Elasticsearch输出配置，使用update动作+条件脚本来控制更新逻辑，具体配置如下：

elasticsearch {
    hosts => [ "localhost:9200" ]
    index => "import-export-logger-%{index-name}"
    document_id => "%{jobID}"
    # 指定执行更新动作，而非默认的覆盖索引
    action => "update"
    # 当文档不存在时，插入当前事件的内容（处理首次出现的日志）
    upsert => {
        "jobID" => "%{jobID}"
        "status" => "%{status}"
        "data" => "%{data}"
    }
    # 用Painless脚本实现条件更新逻辑
    script => {
        source => "if (ctx._source?.status != 'COMPLETE') { ctx._source = params.event; }"
        # 将当前Logstash事件的字段传递给脚本
        params => { "event" => {
            "jobID" => "%{jobID}"
            "status" => "%{status}"
            "data" => "%{data}"
        }}
    }
    script_lang => "painless"
}

配置说明 #

• action => "update"：放弃默认的index覆盖动作，改用update模式，允许基于现有文档的状态做判断。
• upsert：处理文档不存在的场景（比如某jobID的日志首次被采集），直接插入当前事件的内容。
• 条件脚本：
  • 使用Elasticsearch的Painless脚本语言，通过ctx._source?.status安全获取现有文档的状态（?.避免文档字段为空时的报错）。
  • 如果现有文档的状态不是COMPLETE，则用传入的event参数（当前Logstash事件的字段）覆盖原文档内容；如果状态已经是COMPLETE，则不执行任何更新操作，保留原文档的最终状态。

注意事项 #

确保Logstash已经正确解析了日志中的jobID、status、data字段（比如通过grok过滤器），否则脚本无法获取到对应参数，会导致更新失败。

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