Elasticsearch 未从 logstash 接收数据

Question

我有一个 ELlastic 搜索服务器：

    {
  "cluster_name" : "elasticsearch",
  "status" : "yellow",
  "timed_out" : false,
  "number_of_nodes" : 76,
  "number_of_data_nodes" : 1,
  "active_primary_shards" : 297,
  "active_shards" : 297,
  "relocating_shards" : 0,
  "initializing_shards" : 0,
  "unassigned_shards" : 297,
  "delayed_unassigned_shards" : 0,
  "number_of_pending_tasks" : 0,
  "number_of_in_flight_fetch" : 0
}

它在具有 30gb Ram 的双核机器上运行，并且正在接收来自 3 到 4 个 logstash 服务器的日志，并且总共有 30 个输入（所有 log-stash 服务器相结合），但是对于大多数输入，日志都被遗漏了，我30-40 分钟内没有日志，我在 Logstash 服务器中收到警告，如:retrying-failed-action-with-response-code-429。此外，弹性搜索服务器的内存使用率非常高。日志文件的心跳非常高..我检查了 gork 模式，他们是正确的 这是我的 conf 文件之一：

 input {

  exec {
    command => "/usr/bin/touch /var/run/logstash-monit/input.touch && /bin/echo OK."
    interval => 60
    type => "heartbeat"
  }

  file {
    type => 'seller-forever'
   path => '/var/log/seller/seller.log'
   sincedb_path => "/opt/logstash/sincedb-access1"
    }
    }
filter {

grok {
   type => "seller-forever"
    match => [ "message", "%{GREEDYDATA:logline} %{GREEDYDATA:extra_fields}" ]
  }

geoip {
        add_tag => [ "GeoIP" ]
        database => "/opt/logstash/GeoLiteCity.dat"
        source => "clientip"
    }
    if [useragent] != "-" and [useragent] != "" {
      useragent {
        add_tag => [ "UA" ]
        source => "useragent"
      }
    }
    if [bytes] == 0 { mutate { remove => "[bytes]" } }
    if [geoip][city_name]      == "" { mutate { remove => "[geoip][city_name]" } }
    if [geoip][continent_code] == "" { mutate { remove => "[geoip][continent_code]" } }
    if [geoip][country_code2]  == "" { mutate { remove => "[geoip][country_code2]" } }
    if [geoip][country_code3]  == "" { mutate { remove => "[geoip][country_code3]" } }
    if [geoip][country_name]   == "" { mutate { remove => "[geoip][country_name]" } }
    if [geoip][latitude]       == "" { mutate { remove => "[geoip][latitude]" } }
    if [geoip][longitude]      == "" { mutate { remove => "[geoip][longitude]" } }
    if [geoip][postal_code]    == "" { mutate { remove => "[geoip][postal_code]" } }
    if [geoip][region_name]    == "" { mutate { remove => "[geoip][region_name]" } }
    if [geoip][time_zone]      == "" { mutate { remove => "[geoip][time_zone]" } }
    if [urlquery]              == "" { mutate { remove => "urlquery" } }

    if "apache_json" in [tags] {
       if [method]    =~ "(HEAD|OPTIONS)" { mutate { remove => "method" } }
        if [useragent] == "-"              { mutate { remove => "useragent" } }
        if [referer]   == "-"              { mutate { remove => "referer" } }
    }
    if "UA" in [tags] {
        if [device] == "Other" { mutate { remove => "device" } }
        if [name]   == "Other" { mutate { remove => "name" } }
        if [os]     == "Other" { mutate { remove => "os" } }
    }

}


output {

stdout { codec => rubydebug }

elasticsearch {
type => "seller-forever"
index => "seller-forever"
host => "10.0.0.89"
protocol => "node"
   }
}

我正在使用 kibana 进行可视化。 我应该如何解决这个问题我应该检查什么？任何帮助将不胜感激，我无法理解该怎么做。

Answer 1

您检查过您的 Logstash 和 Elasticsearch 日志吗？

另一方面，我重写了您的 logstash 配置，因为您使用的某些选项已过时或在我的 Logstash 版本 2.3.2 中已弃用。

我将 mutate 中的 remove 更改为 remov_field（不推荐使用 remove）。 我删除了协议，因为它已过时（节点是默认选项）。

输入 grok 和 elasticsearch 已过时。 您输入的类型正确，因此 Logstash 会将其与您的文件一起发送。 如果您想根据过滤器中的特定类型做某事。你需要使用这样的东西。

filter {
    if [type] == "apacheAccess" {
        grok {
            match => [ "message", "%{message}" ]
        }

您可以使用 2 个选项修复 unassigned_shards。

1. 您可以强制合并，这样可以强制合并一个或多个索引。 curl -XPOST 'http://localhost:9200/_forcemerge'Elasticsearch Documentation: Force Merge
2. 您可以将 index.routing.allocation.disable_allocation 设置为 false。这将禁用分配。 curl -XPUT 'localhost:9200/_settings' \ -d '{"index.routing.allocation.disable_allocation": false}'