Передача журнала приложений JSON в удаленный LogStash через NXLog на Windows

Я пытался передать логи из приложения Windows, которые уже отформатированы в формате JSON, в logstash через NXlog.

Когда я отправляю файл в Logstash через NXLOG, я получаю множество ошибок в logstash.log:

:message=>"Произошла ошибка. Закрытие подключения",
:client=>"10.xxx.xxx.147:61047",
:exception=>#<IndexError: строка не найдена>

Полный текст ошибки:

{:timestamp=>"2015-04-25T15:15:37.084000-0900", :message=>"Произошла ошибка. Закрытие подключения", :client=>"10.xxx.xxx.147:61047", :exception=>#<IndexError: строка не найдена>, :backtrace=>["org/jruby/RubyString.java:3910:in `[]='", "/opt/logstash/lib/logstash/event.rb:62:in `initialize'", "/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-json_lines-0.1.6/lib/logstash/codecs/json_lines.rb:37:in `decode'", "/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-line-0.1.5/lib/logstash/codecs/line.rb:36:in `decode'", "org/jruby/RubyArray.java:1613:in `each'", "/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-line-0.1.5/lib/logstash/codecs/line.rb:35:in `decode'", "/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-codec-json_lines-0.1.6/lib/logstash/codecs/json_lines.rb:35:in `decode'", "/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-tcp-0.1.3/lib/logstash/inputs/tcp.rb:116:in `handle_socket'", "/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-tcp-0.1.3/lib/logstash/inputs/tcp.rb:145:in `client_thread'", "/opt/logstash/vendor/bundle/jruby/1.9/gems/logstash-input-tcp-0.1.3/lib/logstash/inputs/tcp.rb:143:in `client_thread'"], :level=>:error}
{:timestamp=>"2015-04-25T15:15:38.097000-0900", :message=>"Ошибка разбора JSON. Переход на обычный текст", :error=>#<LogStash::Json::ParserError: Неожиданный конец ввода: ожидается закрывающий маркер для ОБЪЕКТА (из [Источник: [B@26f64966; строка: 1, столбец: 2])
 at [Источник: [B@26f64966; строка: 2, столбец: 5]>, :data=>"  {\r\n", :level=>:info}

Вот моя конфигурация NXLOG:

## Пожалуйста, установите ROOT в папку, в которую был установлен ваш nxlog,
## иначе он не запустится.

define ROOT C:\Program Files (x86)\nxlog

Moduledir %ROOT%\modules
CacheDir %ROOT%\data
Pidfile %ROOT%\data\nxlog.pid
SpoolDir %ROOT%\data
LogFile %ROOT%\data\nxlog.log

<Extension json>
    Module      xm_json
</Extension>

# Внутренние логи Nxlog
<Input internal>
    Module im_internal
    Exec $EventReceivedTime = integer($EventReceivedTime) / 1000000; to_json();
</Input>

# Журнал событий Windows
<Input eventlog>
    Module im_msvistalog
    Exec $EventReceivedTime = integer($EventReceivedTime) / 1000000; to_json();
</Input>

# Логи сервера
<Input Selected_Directory>
    Module      im_file
    File        'E:\\ELK\\logs\\*.json'
    SavePos     False
</Input>

#Вывод журнала событий
<Output out>
   Module om_tcp
   Host 10.xxx.xxx.127
   Port 3515
</Output>

#<output perf_out>
#   Module om_tcp
#   Host 10.xxx.xxx.127
#   Port 3517
#</Output>

#Вывод JSON
<Output out2>
   Module om_tcp
   Host 10.xxx.xxx.127
   Port 3516
</Output> 

<Route 1>
    Path internal, eventlog => out
</Route>

<Route 2>
    Path    Selected_Directory => out2
</Route>

Конфигурация LogStash:

input {
        tcp {
            type   => "eventlog"
            port   => 3515
            codec => json_lines
        }
        tcp {
            type   => "log"
            port   => 3516
            codec => json
        }
    }

output {
    elasticsearch {
        cluster => "MyElkCluster"
        host => "127.0.0.1"
    }
}

Пример формата JSON файла приложения:

[
 {
   "timestamp":"19:54:01.117_0005",
   "type":"N",
   "calllevel":0,
   "thread":772,
   "topic":"ExmpleTopic",
   "level":61,
   "file":"//blah/blah/blah.cpp",
   "function":"functiontext",
   "line":312,
   "message":"Example Message Text",
   "attributes":
    {
      "ThreadName":"1234"
    }
 },
 {
   "timestamp":"20:07:54.038_0691",
   "type":"N",
   "calllevel":0,
   "thread":2324,
   "topic":"ExampleTopic",
   "level":61,
   "file":"//blah/blah/blah.cpp",
   "function":"ExampleFunction",
   "line":2962,
   "message":"Example Message Text",
   "attributes":
    {
      "ThreadName":"1234"
    }
 }
]

У меня есть два дополнительных вопроса, помимо очевидного “помогите разобраться с этой ошибкой”

1. В чем разница между json_lines и json в контексте ввода logstash? Я считал, что json_lines предназначен для потокового текста, а json подразумевает, что я отправлю весь файл сразу.
2. Нужно ли мне добавлять “exec to_json();” в мой nxlog input с именем ‘selected_directory’?

Я бы посоветовал попробовать tcp input, забыть об определении кодека во входных данных, это, как правило, плохая идея, как я выяснил:

input {
  tcp {
    type   => "eventlog"
    port   => 3515
  }
}
filter {
  multiline {
    pattern => "^\s"
    what => "previous"
  }
  json {
    "source" => "message"
  }
}

output {
  elasticsearch {
    cluster => "MyElkCluster"
    host => "127.0.0.1"
  }
}

Так что в этой конфигурации она будет принимать трафик на определенном tcp порту, не устанавливая кодек, затем передавать входные данные в многострочный фильтр, который будет искать строки, которые начинаются с пробела, и если найдет, будет присоединять их к предыдущей строке. Эта новая строка, созданная многострочным фильтром, затем будет передана в json фильтр, который должен быть способен разобрать запись.

Я также получал ошибку при использовании кодека json_lines, которая говорила:

IndexError: строка не найдена

и решил ее, убедившись, что генерируемая мною JSON строка не содержит символов новой строки (например, “\n”), кроме тех случаев, когда они следуют сразу после JSON объекта, в качестве разделителя.