Logstash 6.5.4写入Elasticsearch时日志最终状态异常问题

问题描述 #

日志格式为jobID status data，例如某job的日志按以下顺序生成：

5hgsxyt3838 RUNNING data
5hgsxyt3838 RUNNING data
5hgsxyt3838 COMPLETE data

期望Elasticsearch中该jobID的最终状态为最后一行的COMPLETE，但在多jobID场景下，最终状态却停留在之前的状态（如RUNNING），而非最后一行的状态。

当前使用的Logstash输出配置如下：

output {
    if [fields][target_index] == "import-export-logger" {
        if [status] == "SCHEDULED" {
            elasticsearch {
                hosts => [ "localhost:9200" ]
                index => "import-export-logger-%{index-name}"
                document_id => "%{jobID}"
                action => "create"
            }
        }
        else {
            elasticsearch {
                hosts => [ "localhost:9200" ]
                index => "import-export-logger-%{index-name}"
                document_id => "%{jobID}"
                action => "update"
                doc_as_upsert => true
            }
        }
    }
}

可能的原因 #

• 日志乱序处理：Logstash并行处理多job日志时，某job的COMPLETE日志先被写入ES，但后续该job延迟到达的RUNNING日志会通过update操作直接覆盖文档，导致状态回退。
• 无更新顺序校验：当前配置未引入版本控制或时间戳校验，只要是update操作就会直接覆盖文档，完全不考虑日志的实际生成顺序。

解决方案 #

方式一：用Elasticsearch Pipeline实现状态保护 #

创建Pipeline，当文档当前状态为COMPLETE时，直接丢弃后续的更新请求；也可结合时间戳确保只有更新时间更晚的日志才能生效。

1. 创建状态保护Pipeline：

PUT _ingest/pipeline/protect_complete_status
{
  "processors": [
    {
      "script": {
        "source": """
          // 若现有文档状态已是COMPLETE，终止本次更新
          if (ctx._source?.status == 'COMPLETE') {
            ctx._ingest._drop = true;
          }
          // 可选：如果日志包含生成时间戳，只允许更新时间晚于现有文档的请求
          /*
          if (ctx._source?.log_timestamp && ctx.log_timestamp <= ctx._source.log_timestamp) {
            ctx._ingest._drop = true;
          }
          */
        """
      }
    }
  ]
}

2. 修改Logstash输出配置，指定使用该Pipeline：

output {
    if [fields][target_index] == "import-export-logger" {
        if [status] == "SCHEDULED" {
            elasticsearch {
                hosts => [ "localhost:9200" ]
                index => "import-export-logger-%{index-name}"
                document_id => "%{jobID}"
                action => "create"
                pipeline => "protect_complete_status"
            }
        }
        else {
            elasticsearch {
                hosts => [ "localhost:9200" ]
                index => "import-export-logger-%{index-name}"
                document_id => "%{jobID}"
                action => "update"
                doc_as_upsert => true
                pipeline => "protect_complete_status"
            }
        }
    }
}

方式二：引入版本控制或时间戳校验 #

• 添加日志时间戳：确保每条日志携带精确的生成时间戳（如log_timestamp字段），在ES更新脚本中判断，只有当新日志的时间戳晚于文档现有时间戳时，才执行更新操作。
• 使用ES版本号：创建文档时记录版本号，后续更新请求携带递增的版本号，只有版本号匹配当前文档版本的请求才能成功更新，避免旧日志覆盖新状态。

方式三：调整Logstash处理顺序 #

对单个job的日志做串行处理，确保同一job的日志按生成顺序进入Logstash，这样COMPLETE日志会最后执行更新。但该方式会降低处理吞吐量，仅适合低负载场景。

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