Logstash/elasticsearch 停止接受新数据

我已经建立了一个新的概念证明 logstash 系统

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

从 RPM 安装的 Elasticsearch 和 Logstash……

# rpm -q elasticsearch logstash
elasticsearch-1.7.1-1.noarch
logstash-1.5.3-1.noarch

JVM: 1.8.0_51

我输入的数据是表单的简单记录……

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

（字段是机器名称、用户 ID、日期、时间、登录时间 - 一切都是简单的 US-ASCII）

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

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

这种情况的症状是

• elasticsearch 仍然响应 - 搜索和访问配置仍然很好
• 索引中的文档数量停止上升
• 系统变得基本空闲 - 只有后台磁盘活动和最少的 CPU 使用率
• 如果我尝试停止 logstash 进程（仍在运行），它只会以kill -9.

这似乎发生在大约 200K 文档中。它不受索引数量的影响——我从每日索引开始，然后改为每周索引——它仍然停止在 200K 文档左右。

因为这是在单台机器上运行的概念证明，所以我将副本计数降低到 0，分片降低到 1——我认为这不会对这个问题产生任何影响。

尽管在两者上都增加了详细程度，但我在 logstash 或 elasticsearch 日志中看不到任何错误。

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

我不确定还能看什么。这感觉像是一个微不足道的问题（对于 ELK），我在处理我们的邮件日志的现有 ELK 设置上看不到这个问题（尽管它运行的是早期版本并且有多个弹性搜索存储节点）

尽管我确信输入文件中没有奇数字节序列，但我已charset => "US-ASCII"在file输入插件节中明确将输入声明为 US_ASCII 。我不希望这会产生任何影响（测试仍在运行）。

更新：虽然当导入停止时日志中没有什么有趣的东西，但当logstash被要求关闭时记录的行很有趣……

{: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. 这显示了相同的行为 - 导入在一段时间后停止。

放回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] =~ "[a-z][0-9]{7}" {
                mutate {
                        add_field => [ "userClass", "student" ]
                }
        } else if [userid] =~ "n[a-z].*" {
                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}"
        }
}