Logstash(6.5.4)为elasticsearch添加管道

Question

日志格式jobID status data.

假设我的日志按5hgsxyt3838 RUNNING data5hgsxyt3838 RUNNING data5hgsxyt3838 COMPLETE data的顺序打印。在这里，我希望我在elasticsearch的最终状态是日志的最后一行的状态。但是我观察到，对于多个jobID，我的最终状态不是最后一行中的状态，而是来自其他前一行。原因是什么?是否有一种方法来修复这个使用管道(如果我的最后一行状态是COMPLETE，任何其他之前的行不应该覆盖它)或其他东西?

输出块

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
}
}
}
}

Answer 1

您需要只使用一个worker来运行管道。如果不这样做，则创建的工作线程数量与可用cpu的数量一样多，在这种情况下，可能并发地处理某些事件，并且前一行可能在最后一行之后到达。

如果你在pipeline.yml中配置了你的管道，你需要在你的配置文件中设置它:

pipeline.workers: 1

否则，如果从命令行运行Logstash，只需要添加-w开关

bin/logstash -f logstash.conf -w 1

或者

bin/logstash -f logstash.conf --pipeline.workers 1