如何在将日志数据转换为JSON之前格式化为elasticsearch？

我通过syslog-ng将系统上的所有事件logging到JSON文件中：

 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）监视，该内容将内容转发给logstash （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部分：

• logrotate – 全部处理后运行postrotate
• 思科ASAlogging缓冲垃圾邮件terminal和控制台线路
• logrotate频率 – 从何时开始
• 什么是审计日志syslog设施？
• syslog.conf – 从/ var / log / messagesredirect日志条目

是否可以调整处理链中的任何元素以启用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"} 

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

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