Logstash / elasticsearch перестает принимать новые данные

Я установил новую систему проверки концепции 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

Данные, которые я ввожу, - это простые записи в форме…

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

(поля - это имя машины, идентификатор пользователя, дата, время, время, затраченное на вход в систему - все просто US-ASCII)

Конфигурация Logstash ниже (эти данные поступают из базы данных MSSQL, и на данный момент я экспортирую в текстовый файл и передаю файл на сервер logstash).

Это нормально работало для дневных журналов (11K записей), но когда я пытаюсь обработать отставание за этот календарный год, оно «зависает».

Симптомами этого являются

• elasticsearch по-прежнему реагирует - поиск и доступ к конфигурации все еще в порядке
• количество документов в индексах перестает расти
• система становится существенно простаивающей - только фоновая активность диска и минимальная загрузка ЦП
• если я попытаюсь остановить процесс logstash (который все еще работает), он умрет только с kill -9.

Похоже, это происходит примерно с 200 тыс. Документов. На него не влияет количество индексов - я начал с дневных индексов, а затем перешел на еженедельные - он все еще останавливает около 200 тыс. Документов.

Поскольку это доказательство концепции, работающей на одной машине, я уменьшил счетчик реплик до 0, а сегментов до 1 - я не думаю, что это должно иметь какое-либо значение для этой проблемы.

Я не вижу никаких ошибок в журналах logstash или elasticsearch, несмотря на то, что я включил многословие в обоих.

Я не думаю, что системе не хватает памяти, дискового пространства, файловых дескрипторов.

Я не знаю, на что еще смотреть. Это похоже на проблему незначительного размера (для ELK), и я не вижу этой проблемы в существующей настройке ELK, которая обрабатывает наши почтовые журналы (хотя она работает с более ранними версиями и имеет несколько узлов хранения elasticsearch)

Хотя я уверен, что во входных файлах нет нечетных байтовых последовательностей, я явно объявил вход как US_ASCII с charset => "US-ASCII" в file входной плагин строфа. Я не думаю, что это что-то изменит (тест все еще продолжается).

Обновление: хотя в журналах не было ничего интересного, когда импорт остановил строки, зарегистрированные при 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}"
        }
}