Elactic搜索未Reciving从logstash

Question

数据I有一个弹性的搜索服务器：从3至4 logstash服务器

{ 
    "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拉姆上运行，并且正在接收日志和具有总共30输入（所有日志存储服务器合并），但对于大多数输入日志错过了，我得到30-40分钟没有日志，我得到像:retrying-failed-action-with-response-code-429 logstash服务器警告。同时弹性搜索服务器有一个非常高的内存使用率。日志文件的心跳非常高..我检查了gork patter并且它们是正确的 这是我的一个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}'