Входные данные tcp Logstash не переданы в эластичный поиск

После успешной настройки ELK с использованием файловых входов, пересылки logstash и просмотра журналов в потоке Kibana с нескольких серверов я попытался настроить вход TCP:

tcp {
    codec => "json"
    host => "localhost"
    port => 9250
    tags => ["sensu"]
  }

Отправитель - sensu, а сообщения действительно в формате JSON - это проверяется командой tcpdump.

Журнал Logstash указывает, что соединения приняты:

{:timestamp=>"2015-06-15T14:03:39.832000+1000", :message=>"Accepted connection", :client=>"127.0.0.1:38065", :server=>"localhost:9250", :level=>:debug, :file=>"logstash/inputs/tcp.rb", :line=>"146", :method=>"client_thread"}
{:timestamp=>"2015-06-15T14:03:39.962000+1000", :message=>"config LogStash::Codecs::JSONLines/@charset = \"UTF-8\"", :level=>:debug, :file=>"logstash/config/mixin.rb", :line=>"112", :method=>"config_init"}
{:timestamp=>"2015-06-15T14:03:39.963000+1000", :message=>"config LogStash::Codecs::Line/@charset = \"UTF-8\"", :level=>:debug, :file=>"logstash/config/mixin.rb", :line=>"112", :method=>"config_init"}

Тем не менее, данные, кажется, не идут дальше и не могут быть найдены в Кибане.

Я зашел так далеко, что отключил другие входные данные, а затем наблюдал осколок в asticsearch (curl 'localhost:9200/_cat/shards'), размер которого не увеличивался.

По этой ссылке я на правильном пути, но, вероятно, просто делаю что-то глупое... Заранее спасибо.

logstash.conf:

input {
  file {
    path => ["/var/log/messages", "/var/log/secure", "/var/log/iptables"]
    type => "syslog"
    start_position => "end"
  }

  lumberjack {
    port => 5043
    type => "logs"
    ssl_certificate => "/etc/pki/tls/certs/logstash-forwarder.crt"
    ssl_key => "/etc/pki/tls/private/logstash-forwarder.key"
  }

  tcp {
    codec => "json"
    host => "localhost"
    port => 9250
    tags => ["sensu"]
  }

}

output {
  elasticsearch {
    host => "localhost"
    cluster => "webCluster"
  }
}

elasticsearch.yml:

cluster.name: webCluster
node.name: "bossNode"
node.master: true
node.data: true
index.number_of_shards: 1
index.number_of_replicas: 0
network.host: localhost
Источник

1 ответ

Решение

Еще через несколько разочаровывающих дней я пришел к выводу, что кодек json/json_lines сломан - возможно, только при использовании с входами tcp.

Тем не менее, я нашел обходной путь, используя фильтр:

filter {
  if ("sensu" in [tags]) {
    json {
      "source" => "message"
    }
  }
}

Это и несколько мутаций дают эффект, которого я изначально пытался достичь. Для потомков вот мой рабочий logstash.conf, который объединяет журналы и данные метрик процессора / памяти из sensu:

input {
  file {
    path => [
      "/var/log/messages"
      , "/var/log/secure"
    ]
    type => "syslog"
    start_position => "end"
  }

  file {
    path => "/var/log/iptables"
    type => "iptables"
    start_position => "end"
  }

  file {
    path => ["/var/log/httpd/access_log"
        ,"/var/log/httpd/ssl_access_log"
    ]
    type => "apache_access"
    start_position => "end"
  }

  file {
    path => [
      "/var/log/httpd/error_log"
      , "/var/log/httpd/ssl_error_log"
    ]
    type => "apache_error"
    start_position => "end"
  }

  lumberjack {
    port => 5043
    type => "logs"
    ssl_certificate => "/etc/pki/tls/certs/logstash-forwarder.crt"
    ssl_key => "/etc/pki/tls/private/logstash-forwarder.key"
  }

  tcp {
    host => "localhost"
    port => 9250
    mode => "server"
    tags => ["sensu"]
  }

}

filter {
  if ("sensu" in [tags]) {
    json {
      "source" => "message"
    }
    mutate {
      rename => { "[check][name]" => "type" }
      replace => { "host" => "%{[client][address]}" }
      split => { "[check][output]" => " " }
      add_field => { "output" => "%{[check][output][1]}" }
      remove_field => [ "[client]", "[check]", "occurrences" ]
    }
  } else if([type] == "apache_access") {
    grok {
      match => { "message" => "%{IP:client}" }
    }
  }
}

filter {
  mutate {
    convert => { "output" => "float" }
  }
}

output {
  elasticsearch {
    host => "localhost"
    cluster => "webCluser"
  }
}

Несвязанный с выдачей: "выходные данные" принимаются в виде нескольких значений, разделенных пробелами, отсюда и операция "разбиение". Второй элемент используется, а затем конвертируется в float, так что Kibana хорошо его отображает (что я усвоил на собственном опыте).

Источник
Другие вопросы по тегам