filebeat+kafka+ELK 5.0

filebeat和ELK全用了5.0了,kafka是0.10.1.0。
filebeat产生的消息用output:kafka到达kafka后是这样:
{"@timestamp":"2016-11-28T06:13:29.168Z","beat":{"hostname":"localhost.localdomain","name":"192.168.138.3","version":"5.0.1"},"input_type":"log","message":"Nov 28 09:09:47 localhost kernel: type=1305 audit(1480295387.539:3): audit_pid=1196 old=0 auid=4294967295 ses=4294967295 res=1","offset":1475347,"source":"/var/log/messages","type":"syslog"}
 
logstash用input-kafka后是这样
{
    "@timestamp" => 2016-11-28T06:11:44.896Z,
         "geoip" => {},
      "@version" => "1",
       "message" => "{\"@timestamp\":\"2016-11-28T06:13:29.168Z\",\"beat\":{\"hostname\":\"localhost.localdomain\",\"name\":\"192.168.138.3\",\"version\":\"5.0.1\"},\"input_type\":\"log\",\"message\":\"Nov 28 09:09:47 localhost kernel: type=1305 audit(1480295387.539:3): audit_pid=1196 old=0 auid=4294967295 ses=4294967295 res=1\",\"offset\":1475347,\"source\":\"/var/log/messages\",\"type\":\"syslog\"}",
          "tags" => [
        [0] "_grokparsefailure",
        [1] "_geoip_lookup_failure"
    ]
}
 
全部都在message里了,完全没有其他beat的field。
beat通过kafka到logstash,要怎么解析呢?我这是只用了input-kafka,可以嵌套个input-beats吗?
filter里面是没有beat插件的
 
已邀请:

dennishood

赞同来自: medcl

妈蛋,很神奇,每次来这儿一发问题,过几分钟就自己解决了。

jhin

赞同来自:

请问下。。filebeat写到kafka,我启动filebeat抓了日志,我在写入日志就不抓了(也不是不抓,能够输入到文件里,但是kafka就没有日志更新),重启一下filebeat,新写的日志又会一古脑的传到kafka,再写又不传,如此。。
 

jhin

赞同来自:

看到了麻烦留个联系方式。。或者加下我的qq951982186,,谢谢了

forsaken627 - 90后打字员

赞同来自:

您好,可以看一下你的logstash用input-kafka的这段配置么?我使用input-kafka遇到了一些问题

wangohyes

赞同来自:

[2016-12-08T23:03:50,675][ERROR][logstash.inputs.kafka    ] Unknown setting 'zk_connect' for kafka
[2016-12-08T23:03:50,678][ERROR][logstash.inputs.kafka    ] Unknown setting 'topic_id' for kafka
[2016-12-08T23:03:50,678][ERROR][logstash.inputs.kafka    ] Unknown setting 'reset_beginning' for kafka
用的kafka0.10.0.0,logstash版本是5.0.2,启动会bao报错,有人遇到这个问题了嘛?

要回复问题请先登录注册