# Re: Filebeat \> Kafka \> Logstash \> ElasticSearch

**URL:** <https://discuss.elastic.co/t/re-filebeat-kafka-logstash-elasticsearch/97888>\
**Category:** Logstash\
**Created:** [August 22, 2017, 10:07am UTC](https://discuss.elastic.co/t/re-filebeat-kafka-logstash-elasticsearch/97888 "2017-08-22T10:07:42Z")\
**Posts on this page:** 6\
**Page:** 1

<div class="post-metadata">

**Author:** ![inhinyera16](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/inhinyera16/32/61625_2.png) [@inhinyera16](https://discuss.elastic.co/u/inhinyera16)\
**Post date:** [August 22, 2017, 10:07am UTC](https://discuss.elastic.co/t/re-filebeat-kafka-logstash-elasticsearch/97888/1 "2017-08-22T10:07:42Z")

</div>

Anyone here why beats data such as beat name, version etc is not included after indexing in elasticsearch?

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [August 23, 2017, 8:55pm UTC](https://discuss.elastic.co/t/re-filebeat-kafka-logstash-elasticsearch/97888/2 "2017-08-23T20:55:38Z")

</div>

You're talking about the `beat` field and its subfields? It should be included. We can't say much without seeing your configuration and a sample event produced by a `stdout { codec => rubydebug }` output.

---

<div class="post-metadata">

**Author:** ![inhinyera16](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/inhinyera16/32/61625_2.png) [@inhinyera16](https://discuss.elastic.co/u/inhinyera16)\
**Post date:** [August 24, 2017, 3:20am UTC](https://discuss.elastic.co/t/re-filebeat-kafka-logstash-elasticsearch/97888/3 "2017-08-24T03:20:44Z")

</div>

> [@magnusbaeck](#):
>
> s? It should be included. We can’t say much without seeing your configuration and a sample event prod

thank you for your reply, magnus.

We have configured our filebeat to output in kafka only:

output.kafka:  
hosts: ["host1:9092","host2:9093","host32:9094"]  
codec.format:  
string: '%{[message]}{[beats]}'  
topic: 'mytopic'

while our logstash config is :

input {  
kafka {  
topics =\> ["mytopic"]  
bootstrap\_servers =\> "host1:9092,host2:9093,host3:9094"  
}  
}

output {  
elasticsearch {  
hosts =\> ["142.122.218.12:9200"]  
index =\> "mycollection"  
}  
}

Filebeat stdout contains the beat field.  
after indexing in elasticsearch, the beat field along its subfields are gone/does not exists.

---

<div class="post-metadata">

**Author:** ![magnusbaeck](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/magnusbaeck/32/44943_2.png) [@magnusbaeck](https://discuss.elastic.co/u/magnusbaeck)\
**Post date:** [August 24, 2017, 2:38pm UTC](https://discuss.elastic.co/t/re-filebeat-kafka-logstash-elasticsearch/97888/4 "2017-08-24T14:38:32Z")

</div>

Please supply everything I asked for. The sample event is missing.

---

<div class="post-metadata">

**Author:** ![inhinyera16](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/inhinyera16/32/61625_2.png) [@inhinyera16](https://discuss.elastic.co/u/inhinyera16)\
**Post date:** [August 29, 2017, 3:24am UTC](https://discuss.elastic.co/t/re-filebeat-kafka-logstash-elasticsearch/97888/6 "2017-08-29T03:24:54Z")

</div>

Sorry if i forgot to include sample event.

But, i have found the solution.

Whenever we use filebeat to send event in logstash, its metadata is also included.  
But, if the source comes from filebeat to kafka, by default, only the message field will be sent to kafka.  
That is why we need to set the filebeat codec format and append the filebeat metadata such as input\_type, filesource, .. etc.... see below config.

output.kafka:  
hosts: ["host:9092"]

codec.format:  
string: 'beat=%{[beat]} offset=%{[offset]} filesource=%{[source]} input\_type=%{[input\_type]} start=%{[message]}'  
topic: 'mytopic'

We can now parse these fields using grok filter in logstash.

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [September 26, 2017, 3:25am UTC](https://discuss.elastic.co/t/re-filebeat-kafka-logstash-elasticsearch/97888/7 "2017-09-26T03:25:00Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.