# ELK with Kafka

**URL:** <https://discuss.elastic.co/t/elk-with-kafka/53479>\
**Category:** Logstash\
**Created:** [June 21, 2016, 10:10am UTC](https://discuss.elastic.co/t/elk-with-kafka/53479 "2016-06-21T10:10:40Z")\
**Posts on this page:** 2\
**Page:** 1

<div class="post-metadata">

**Author:** ![vienodp](https://avatars.discourse-cdn.com/v4/letter/v/46a35a/32.png) [@vienodp](https://discuss.elastic.co/u/vienodp)\
**Post date:** [June 21, 2016, 10:10am UTC](https://discuss.elastic.co/t/elk-with-kafka/53479/1 "2016-06-21T10:10:40Z")

</div>

Hi ,

I am trying to introduce Kafka in ELK. I have 2 servers  
On first, configured ELK , filebeat and kafka input plugin  
On second, installed Kafka, logstash, kafka output plugin

I am trying to push my second machine's log into kafka and then into elasticsearch which is on first machine.

Somehow filebeat is working fine but using kafka its not working. Here are the input, output files

first machine which is ELK server -

> [root@watlelk01 conf.d]# cat 02-beats-input.conf  
> input {  
> beats {  
> port =\> 5044  
> ssl =\> true  
> ssl\_certificate =\> "/etc/pki/tls/certs/logstash-forwarder.crt"  
> ssl\_key =\> "/etc/pki/tls/private/logstash-forwarder.key"  
> }  
> }

> [root@watlelk01 conf.d]# cat 10-syslog-filter.conf  
> filter {  
> if [type] == "syslog" {  
> grok {  
> match =\> { "message" =\> "%{SYSLOGTIMESTAMP:syslog\_timestamp} %{SYSLOGHOST:syslog\_hostname} %{DATA:syslog\_program}(?:[%{POSINT:syslog\_pid}])?: %{GREEDYDATA:syslog\_message}" }  
> add\_field =\> ["received\_at", "%{@timestamp}"]  
> add\_field =\> ["received\_from", "%{host}"]  
> }  
> syslog\_pri { }  
> date {  
> match =\> ["syslog\_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss"]  
> }  
> }  
> }

> [root@watlelk01 conf.d]# cat 20-logstash-logs-kafka-consumer.conf  
> input {  
> kafka {  
> zk\_connect =\> '9.2.153.25:2181'  
> topic\_id =\> 'logstash\_logs'  
> }  
> }

> output {  
> stdout { codec =\> rubydebug }  
> }  
> [root@watlelk01 conf.d]# cat 30-elasticsearch-output.conf  
> output {  
> elasticsearch {  
> hosts =\> ["localhost:9200"]  
> sniffing =\> true  
> manage\_template =\> false  
> index =\> "%{[@metadata][beat]}-%{+YYYY.MM.dd}"  
> document\_type =\> "%{[@metadata][type]}"  
> }  
> }

On second i.e. kafka machine -

> [root@watlelkyz00 conf.d]# ls  
> 10-logstash-logs-kafka-producer.conf 20-syslog-filter.conf  
> [root@watlelkyz00 conf.d]# cat 10-logstash-logs-kafka-producer.conf  
> input {  
> file {  
> path =\> ["/var/logs/_/_.log"]  
> }  
> }

> output {  
> kafka {  
> bootstrap\_servers =\> 'localhost:9092'  
> topic\_id =\> 'logstash\_logs'  
> }  
> }  
> [root@watlelkyz00 conf.d]# cat 20-syslog-filter.conf  
> filter {  
> if [type] == "syslog" {  
> grok {  
> match =\> { "message" =\> "%{SYSLOGTIMESTAMP:syslog\_timestamp} %{SYSLOGHOST:syslog\_hostname} %{DATA:syslog\_program}(?:[%{POSINT:syslog\_pid}])?: %{GREEDYDATA:syslog\_message}" }  
> add\_field =\> ["received\_at", "%{@timestamp}"]  
> add\_field =\> ["received\_from", "%{host}"]  
> }  
> syslog\_pri { }  
> date {  
> match =\> ["syslog\_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss"]  
> }  
> }  
> }

---

<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:** [July 6, 2017, 4:51am UTC](https://discuss.elastic.co/t/elk-with-kafka/53479/2 "2017-07-06T04:51:31Z")

</div>