Отложенное индексирование ElasticSearch

В настоящее время у меня есть следующая настройка:

серверы syslog-ng -> Logstash -> ElasticSearch

Серверы syslog-ng сбалансированы по нагрузке и записывают в местоположение SAN, где Logstash просто отслеживает файлы и отправляет их в ES. В настоящее время я получаю около 1300 событий в секунду в кластер syslog для сетевых журналов. Проблема, с которой я сталкиваюсь, - это постепенная задержка, когда журналы становятся доступными для поиска в ES. Когда я запустил кластер (4 узла), он был мертв. Затем отставание на несколько минут, а теперь, через 4 дня, отставание на ~ 35 минут. Я могу подтвердить, что журналы записываются в режиме реального времени на серверах syslog-ng, и я также могу подтвердить, что мои 4 других индекса, которые используют ту же концепцию, но другой экземпляр Logstash, остаются в актуальном состоянии. Однако они значительно ниже (~ 500 событий в секунду).

Похоже, что экземпляр Logstash, который читает плоский файл, не успевает. Я уже однажды разделил эти файлы и создал 2 экземпляра Logstash, чтобы они помогли, но я все еще отстаю.

Любая помощь будет принята с благодарностью.

-

Обычно вводятся журналы ASA, в основном отказы и VPN-подключения.

Jan  7 00:00:00 firewall1.domain.com Jan 06 2016 23:00:00 firewall1 : %ASA-1-106023: Deny udp src outside:192.168.1.1/22245 dst DMZ_1:10.5.1.1/33434 by access-group "acl_out" [0x0, 0x0]
Jan  7 00:00:00 firewall2.domain.com %ASA-1-106023: Deny udp src console_1:10.1.1.2/28134 dst CUSTOMER_094:2.2.2.2/514 by access-group "acl_2569" [0x0, 0x0]

Вот моя конфигурация Logstash.

input {

file {
    type => "network-syslog"
    exclude => ["*.gz"]
    start_position => "end"
    path => [ "/location1/*.log","/location2/*.log","/location2/*.log"]
    sincedb_path => "/etc/logstash/.sincedb-network"
  }
}

filter {
    grok {
      overwrite => [ "message", "host" ]
      patterns_dir => "/etc/logstash/logstash-2.1.1/vendor/bundle/jruby/1.9/gems/logstash-patterns-core-2.0.2/patterns"
      match => [
        "message", "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:host} %%{CISCOTAG:ciscotag}: %{GREEDYDATA:message}",
        "message", "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:host} %{GREEDYDATA:message}"
      ]
     }
    grok {
      match => [
        "message", "%{CISCOFW106001}",
        "message", "%{CISCOFW106006_106007_106010}",
        "message", "%{CISCOFW106014}",
        "message", "%{CISCOFW106015}",
        "message", "%{CISCOFW106021}",
        "message", "%{CISCOFW106023}",
        "message", "%{CISCOFW106100}",
        "message", "%{CISCOFW110002}",
        "message", "%{CISCOFW302010}",
        "message", "%{CISCOFW302013_302014_302015_302016}",
        "message", "%{CISCOFW302020_302021}",
        "message", "%{CISCOFW305011}",
        "message", "%{CISCOFW313001_313004_313008}",
        "message", "%{CISCOFW313005}",
        "message", "%{CISCOFW402117}",
        "message", "%{CISCOFW402119}",
        "message", "%{CISCOFW419001}",
        "message", "%{CISCOFW419002}",
        "message", "%{CISCOFW500004}",
        "message", "%{CISCOFW602303_602304}",
        "message", "%{CISCOFW710001_710002_710003_710005_710006}",
        "message", "%{CISCOFW713172}",
        "message", "%{CISCOFW733100}",
        "message", "%{GREEDYDATA}"
      ]
    }
    syslog_pri { }
    date {
      "match" => [ "syslog_timestamp", "MMM  d HH:mm:ss",
                   "MMM dd HH:mm:ss" ]
      target => "@timestamp"
    }
    mutate {
      remove_field => [ "syslog_facility", "syslog_facility_code", "syslog_severity", "syslog_severity_code"]
    }
}

output {
    elasticsearch {
    hosts => ["server1","server2","server3"]
    index => "network-%{+YYYY.MM.dd}"
    template => "/etc/logstash/logstash-2.1.1/vendor/bundle/jruby/1.9/gems/logstash-output-elasticsearch-2.2.0-java/lib/logstash/outputs/elasticsearch/elasticsearch-network.json"
    template_name => "network"
 }
}