Logstash 为 elasticsearch 添加管道

分享于2022年12月29日 elasticsearch logstash 问答
【问题标题】:Logstash (6.5.4) Adding pipeline for elasticsearchLogstash (6.5.4) 为 elasticsearch 添加管道
【发布时间】:2022-12-06 20:49:36
【问题描述】:

日志格式 jobID status data

假设我的日志按顺序打印 5hgsxyt3838 RUNNING data 5hgsxyt3838 RUNNING data 5hgsxyt3838 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
            }
        }
    }
}


【解决方案1】:

您只需要一名工人来运行您的管道。如果您不这样做,那么将创建与您可以使用的可用 CPU 数量一样多的工作线程,在这种情况下,一些事件可能会被并发处理,并且前一行可能会在最后一行之后到达。

如果您在 pipeline.yml 中配置了管道,则需要在配置文件中进行设置:

pipeline.workers: 1

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

bin/logstash -f logstash.conf -w 1

或等价地

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

【讨论】: