Эластичная опускается

У меня есть стек ELK с Redis. Схема: logstash -> redis -> logstash (индексатор) -> elasticsearch -> kibana

Индексатор Logstash получает данные из Redis и помещает их в эластичный:

    input { 
                redis { 
                        host=>"redis"
                        type=>"redis-input"
                        data_type=>"list"
                        key=>"logstash"
                        } 
                }
        filter { 
                geoip { 
                        source=>"ipaddr"
                        target=>"geoip"
                        database=>"/GeoLiteCity.dat"
                        add_field=>["[geoip][coordinates]","%{[geoip][longitude]}"]
                        add_field=>["[geoip][coordinates]","%{[geoip][latitude]}"]
                        } 
                mutate {
                        remove_field=>["message","@version","timestamp"]
                        convert=>{"[geoip][coordinates]"=>"float"}
                        }
                }
        output { 
                    elasticsearch { 
                                    template=>"/typing-template.json"
                                    template_overwrite=>true
                                    hosts=>["elasticsearch:9200"] 
                                    }
                }

Есть 4 сервера, логи которых я хочу собрать. Вот их конфиг logstash:

input {
    file {
        path => [ "C:/Program Files (x86)/*/logs/*.log", "C:/Program Files (x86)/**/logs/*.log", "C:/Program Files/***/logs/*.log", "C:/Program Files/****/logs/*.log" ]
        start_position => "beginning"
        type => "mtdclog"
        ignore_older => 0
        sincedb_path => "NUL"
            }
}   

filter {
        grok {      match => { "path" => "%{GREEDYDATA}/(?<logdate>[0-9]{8})\.log" }}
        grok {      match => [  "message", "%{NONNEGINT:log_stream}\t%{TIME:logtime}\s{1,2}%{IPV4:ipaddr}\t'%{NUMBER:account}': (?<event>login) \[ver: (?<client_build>[0-9\.]+)",
                                "message", "%{NONNEGINT:log_stream}\t%{TIME:logtime}\s{1,2}%{IPV4:ipaddr}\t'%{NUMBER:account}': (?<event>liveupdate) '%{GREEDYDATA:data}'",
                                "message", "%{NONNEGINT:log_stream}\t%{TIME:logtime}\s{1,2}%{IPV4:ipaddr}\t'%{NUMBER:account}': (?<event>check version)%{GREEDYDATA:data}",
                                "message", "%{NONNEGINT:log_stream}\t%{TIME:logtime}\s{1,2}%{IPV4:ipaddr}\t'%{NUMBER:account}': %{GREEDYDATA:data}",
                                "message", "%{NONNEGINT:log_stream}\t%{TIME:logtime}\s{1,2}(?<event>News):%{GREEDYDATA:data}",
                                "message", "%{NONNEGINT:log_stream}\t%{TIME:logtime}\s{1,2}%{IPV4:ipaddr}\t(?<event>unknown command) (?<command_code>[A-Z0-9]+)",
                                "message", "%{NONNEGINT:log_stream}\t%{TIME:logtime}\s{1,2}(?<event>History):%{GREEDYDATA:data}",
                                "message", "%{NONNEGINT:log_stream}\t%{TIME:logtime}\s{1,2}%{GREEDYDATA:log_line}",
                                "message", "%{GREEDYDATA:log_line}"
                                ]   
                    }

        mutate {
            add_field => { "ts" => "%{logdate} %{logtime}"} 
            remove_field => [ "logdate", "logtime" ]
                }

        date {
                match => [ "ts", "YYYYMMdd HH:mm:ss.SSS" ]
                target => "@timestamp"
                }   


        if [path] =~ "Pattern1" { mutate {  add_field => { "dc_type" => "Pattern1" }    }}
        if [path] =~ "Pattern2" { mutate {  add_field => { "dc_type" => "Pattern2" }    }}

        mutate {    remove_field => [ "message", "@version", "ts", "path", "host"  ]    
                    add_field => { "location" => "somecity" }
                    convert => {    "log_stream" => "integer"
                                    "client_build"   => "integer"
                                    "account" => "integer"
                                    }
                    }                               

}

output {

            redis  {
            host => "xxx.yyy.zzz.aaa"
            port => "6381"
            data_type => "list"
            key => "logstash" }

Задача: я хочу обработать старые журналы за 1 месяц. Это около 35 МБ файлов ежедневного журнала. Таким образом, общий объем с 4 серверов составляет около 140 МБ, не так много.

Проблема: Потом запускаю службы logstash - все хорошо, 4-5 часов работает нормально. Я вижу проанализированные данные в кибане и могу с ними работать. Но потом резинка идет вниз. Сообщение: «Тайм-аут запроса через 30000 мс».

Тот же стек ELK, который я использую для других серверов и конфигураций logstash - он отлично работает и обрабатывает больше строк журнала. Но я не могу понять, в чем проблема в этом случае.