Logstash / elasticsearch停止接受新的数据

我已经build立了logstash系统的新概念

CentOS 6.6 (on Vmware 5.5) - single CPU VM with 12G RAM allocated 

Elasticsearch和Logstash从RPM安装…

 # rpm -q elasticsearch logstash elasticsearch-1.7.1-1.noarch logstash-1.5.3-1.noarch JVM: 1.8.0_51 

我正在喂的数据是forms的简单logging…

 M1234 z123 2015-01-31 23:28:09.417 8.55373 

（字段是机器名称，用户名，date，时间，login时间 – 一切都很简单US-ASCII）

下面的logstashconfiguration（这个数据来自MSSQL数据库，目前我正在导出到文本文件并将文件传输到logstash服务器）。

这对于一天的日志（11Klogging）来说工作得很好，但是当我尝试处理这个日历年的积压时，它会“挂起”。

这是症状

• elasticsearch仍然响应 – search和访问configuration仍然很好
• 指数中的文件数量不再增加
• 系统变得非常重要 – 只有后台磁盘活动和最less的CPU使用率
• 如果我试图停止logstash进程（它仍在运行），它只会死亡kill -9 。

这似乎发生在20万左右的文件。 它不受指数数量的影响 – 我以每日指数开始，然后变成每周 – 它仍然停止在20万左右的文档。

因为这是在一台机器上运行的概念certificate，我已经把副本计数降低到0，碎片降到1 – 我不认为这应该对这个问题有任何的区别。

尽pipe在两个版本上都有变化，但是在logstash或elasticsearch日志中我没有看到任何错误。

我不认为系统内存不足，磁盘空间不足，文件描述符。

我不知道还有什么要看。 这感觉就像一个普通的问题（对于ELK），我不知道现有的ELK设置，它处理我们的邮件日志（虽然这是运行早期版本，并有多个elasticsearch存储节点）

虽然我确信input文件中没有奇数字节序列，但是我已经在fileinput插件节中显式地将input声明为US_ASCII，其charset => "US-ASCII" 。 我不期望这会有什么不同（testing仍在运行）。

更新：虽然在日志中没有任何有趣的东西，当导入停滞logstash被要求closureslogging的行是有趣的…

 {:timestamp=>"2015-08-03T10:17:39.104000+0100", :message=>["INFLIGHT_EVENTS_REPORT", "2015-08-03T10:17:39+01:00", {"input_to_filter"=>20, "filter_to_output"=>0, "outputs"=>[]}], :level=>:warn} 

这意味着问题出在过滤阶段，而不是elasticsearch的输出。 我已经证实，首先摆脱elasticsearch输出，只是有stdout 。 这表明了同样的行为 – 一段时间后import停顿。

把elasticsearch输出放回去，但清除filter部分中的所有内容，给了我一个成功的，完整的数据导入。

我现在已经有了一个解决这个问题的答案。

 input { file { path => "/var/lib/clusters/*" type => "clusterF" start_position => "beginning" } } filter { mutate { remove_field => [ "path", "host" ] } # 13COMP014 nabcteam 2015-07-29 11:09:21.353 153.493 if [type] == "clusterF" { grok { match => { "message" => "%{NOTSPACE:client} +%{WORD:userid} +%{TIMESTAMP_ISO8601:datestamp} +%{BASE10NUM:elapsed:float}" } } } if [elapsed] < 0 { drop {} } if [elapsed] > 1000.0 { drop {} } if [userid] =~ "[az][0-9]{7}" { mutate { add_field => [ "userClass", "student" ] } } else if [userid] =~ "n[az].*" { mutate { add_field => [ "userClass", "staff" ] } } else { mutate { add_field => [ "userClass", "other" ] } } date { match => [ "datestamp", "ISO8601" ] } mutate { remove_field => [ "message" ] } } output { elasticsearch { bind_host => "clog01.ncl.ac.uk" protocol => "http" cluster => "elasticsearch" flush_size => 10 index => "clusters-%{+xxxx.ww}" } }