linux - 从 bash 脚本启动时 Logstash 关闭停止

Question

我编写了一个 bash 脚本，它在指定的文件夹中查找 CSV 文件，并使用正确的配置文件将它们通过管道传输到 logstash 中。然而，当运行这个脚本时，我遇到了以下错误，说关闭过程被停滞，导致无限循环，直到我用 ctrl+c 手动停止它:

[2018-03-22T08:59:53,833][INFO ][logstash.runner          ] Starting Logstash {"logstash.version"=>"6.2.3"}
[2018-03-22T08:59:54,211][INFO ][logstash.agent           ] Successfully started Logstash API endpoint {:port=>9600}
[2018-03-22T08:59:57,970][INFO ][logstash.pipeline        ] Starting pipeline {:pipeline_id=>"main", "pipeline.workers"=>2, "pipeline.batch.size"=>125, "pipeline.batch.delay"=>50}
[2018-03-22T08:59:58,116][INFO ][logstash.pipeline        ] Pipeline started succesfully {:pipeline_id=>"main", :thread=>"#<Thread:0xf6851b3 run>"}
[2018-03-22T08:59:58,246][INFO ][logstash.agent           ] Pipelines running {:count=>1, :pipelines=>["main"]}
[2018-03-22T08:59:58,976][INFO ][logstash.outputs.file    ] Opening file {:path=>"/home/kevin/otrs_customer_user"}
[2018-03-22T09:00:06,471][WARN ][logstash.shutdownwatcher ] {"inflight_count"=>0, "stalling_thread_info"=>{["LogStash::Filters::CSV", {"separator"=>";", "columns"=>["IOT", "OID", "SUM", "XID", "change_by", "change_time", "city", "company", "company2", "create_by", "create_time", "customer_id", "email", "fax", "first_name", "id", "inst_city", "inst_first_name", "inst_last_name", "inst_street", "inst_zip", "last_name", "login", "mobile", "phone", "phone2", "street", "title", "valid_id", "varioCustomerId", "zip"], "id"=>"f1c74146d6672ca71f489aac1b4c2a332ae515996657981e1ef44b441a7420c8"}]=>[{"thread_id"=>23, "name"=>nil, "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:90:in `read_batch'"}]}}
[2018-03-22T09:00:06,484][ERROR][logstash.shutdownwatcher ] The shutdown process appears to be stalled due to busy or blocked plugins. Check the logs for more information.
[2018-03-22T09:00:11,438][WARN ][logstash.shutdownwatcher ] {"inflight_count"=>0, "stalling_thread_info"=>{["LogStash::Filters::CSV", {"separator"=>";", "columns"=>["IOT", "OID", "SUM", "XID", "change_by", "change_time", "city", "company", "company2", "create_by", "create_time", "customer_id", "email", "fax", "first_name", "id", "inst_city", "inst_first_name", "inst_last_name", "inst_street", "inst_zip", "last_name", "login", "mobile", "phone", "phone2", "street", "title", "valid_id", "varioCustomerId", "zip"], "id"=>"f1c74146d6672ca71f489aac1b4c2a332ae515996657981e1ef44b441a7420c8"}]=>[{"thread_id"=>23, "name"=>nil, "current_call"=>"[...]/logstash-core/lib/logstash/util/wrapped_synchronous_queue.rb:90:in `read_batch'"}]}}

当我使用 bash logstash -f xyz.config < myfile.config 手动运行相同的文件和相同的配置时它按预期工作并且该过程被正确终止。在 bash 脚本中，我基本上使用了确切的命令，但遇到了上面的错误。

我还注意到问题似乎是随机的，并不是每次都出现在同一个文件和配置上。

我的配置包括一个 stdin 输入、一个 csv 过滤器和用于测试以 json 格式输出到一个文件(也删除了 stdout{} )。

有人知道为什么我的进程在脚本执行期间停滞吗？或者，如果没有，是否有办法告诉 logstash 在停滞时关闭？

示例配置:

input {
  stdin {
    id => "${LS_FILE}"
  }

}
filter {
    mutate {
        add_field => { "foo_type" => "${FOO_TYPE}" }
        add_field => { "[@metadata][LS_FILE]" => "${LS_FILE}"}
    }
    if [@metadata][LS_FILE] == "contacts.csv" {
      csv {
          separator => ";"
          columns =>
          [
            "IOT",
            "OID",
            "SUM",
            "XID",
            "kundenid"              
          ]
      }
      if [kundenid]{
        mutate {
            update => { "kundenid" => "n-%{kundenid}" }
        }
        }
    }
}
output {   
       if [@metadata][LS_FILE] == "contacts.csv" {
           file{
                path => "~/contacts_file"
                codec => json_lines
            }
       }
}

示例脚本:

LOGSTASH="/customer/app/logstash-6.2.3/bin/logstash"

    for file in $(find $TARGETPATH -name *.csv) # Loop each file in given path
    do
        if [[ $file = *"foo"* ]]; then
            echo "Importing $file"
            export LS_FILE=$(basename $file)
            bash $LOGSTASH -f $CFG_FILE < $file  # Starting logstash
            echo "file $file imported."
        fi
    done

我在 bash 脚本中导出环境变量，并将它们设置为 logstash 配置中的元数据，以便为不同的输入文件执行一些条件。文件中的 JSON 输出仅用于测试目的。

Answer 1

当您尝试关闭时，Logstash 会尝试执行各种步骤，例如，

• 它停止所有输入、过滤器和输出插件
• 处理所有飞行中的事件
• 终止 Logstash 进程

并且有多种因素使得关机过程非常不可预测，例如，

• 一个以缓慢的速度接收数据的输入插件。
• 慢速过滤器，例如执行 sleep(10000) 的 Ruby 过滤器或执行非常繁重查询的 Elasticsearch 过滤器。
• 一个断开连接的输出插件，正在等待重新连接以刷新运行中的事件。

来自 Logstash documentation ,

Logstash has a stall detection mechanism that analyzes the behavior of the pipeline and plugins during shutdown. This mechanism produces periodic information about the count of inflight events in internal queues and a list of busy worker threads.

您可以在启动 logstash 时使用 --pipeline.unsafe_shutdown 标志以在停滞关闭的情况下强制终止进程。当 --pipeline.unsafe_shutdown 未启用时，Logstash 会继续运行并定期生成这些报告，这就是为什么问题在您的案例中看起来是随机的。

请记住，不安全的关闭、强制终止 Logstash 进程或崩溃Logstash 进程由于任何其他原因可能导致数据丢失(除非您已启用 Logstash 以使用 persistent queues )。