Как отформатировать данные журнала перед пересылкой их в формате JSON в elasticsearch?

Я записываю все события в системе в файл JSON через syslog-ng:

destination d_json { file("/var/log/all_syslog_in_json.log" perm(0666) template("{\"@timestamp\": \"$ISODATE\", \"facility\": \"$FACILITY\", \"priority\": \"$PRIORITY\", \"level\": \"$LEVEL\", \"tag\": \"$TAG\", \"host\": \"$HOST\", \"program\": \"$PROGRAM\", \"message\": \"$MSG\"}\n")); };

log { source(s_src); destination(d_json); };

Этот файл контролируется logstash (2.0 beta), который перенаправляет контент в elasticsearch (2.0 RC1):

input
{
  file
  {
    path => "/var/log/all_syslog_in_json.log"
    start_position => "beginning"
    codec => json
    sincedb_path => "/etc/logstash/db_for_watched_files.db"
    type => "syslog"
  }

}

output {
    elasticsearch {
        hosts => ["elk.example.com"]
        index => "logs"
    }
}

Затем я визуализирую результаты в kibana.

Эта настройка работает нормально, за исключением того, что kibana не расширяет message часть:

Можно ли настроить какой-либо из элементов цепочки обработки, чтобы разрешить расширение messages (так что его компоненты находятся на том же уровне, что и path или type?

РЕДАКТИРОВАТЬ: по запросу, несколько строк из /var/log/all_syslog_in_json.log

{"@timestamp": "2015-10-21T20:14:05+02:00", "facility": "auth", "priority": "info", "level": "info", "tag": "26", "host": "eu2", "program": "sshd", "message": "Disconnected from 10.8.100.112"}
{"@timestamp": "2015-10-21T20:14:05+02:00", "facility": "authpriv", "priority": "info", "level": "info", "tag": "56", "host": "eu2", "program": "sshd", "message": "pam_unix(sshd:session): session closed for user nagios"}
{"@timestamp": "2015-10-21T20:14:05+02:00", "facility": "authpriv", "priority": "info", "level": "info", "tag": "56", "host": "eu2", "program": "systemd", "message": "pam_unix(systemd-user:session): session closed for user nagios"}

Я считаю, что вы используете неправильный кодек для ввода, вам нужно использовать json_lines из документы:

If you are streaming JSON messages delimited by \n then see the json_lines codec.

Использовать этот кодек вместо. В качестве альтернативы вы можете игнорировать кодек на входе и отправлять их через фильтр json, как я всегда это делаю.

filter {
    json {
        source => "message"
    }
}