# Filter for Kafka and Zookeeper System Logs

**URL:** <https://discuss.elastic.co/t/filter-for-kafka-and-zookeeper-system-logs/124276>\
**Category:** Logstash\
**Created:** [March 16, 2018, 10:49am UTC](https://discuss.elastic.co/t/filter-for-kafka-and-zookeeper-system-logs/124276 "2018-03-16T10:49:17Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![jake\_smith](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jake_smith/32/47016_2.png) [@jake\_smith](https://discuss.elastic.co/u/jake_smith)\
**Post date:** [March 16, 2018, 10:49am UTC](https://discuss.elastic.co/t/filter-for-kafka-and-zookeeper-system-logs/124276/1 "2018-03-16T10:49:17Z")

</div>

Hi all - I'm new to elastic and grok. I'm trying to get Kafka/Zookeeper system logs filtered and parsed by Logstash (shipped by filebeat), then stored in Elasticsearch, so that they may be viewable on Kibana all on CentOS 7.

I imagine I don't need the Kafka plugin as I'm not interested in Kafka topics, only the system related logs (controller.log, system.log, zookeeper.log, etc).

Can I use grok to filter these logs? What would a good starting point for a filter like this be?

Example log:  
`[2018-03-15 22:27:31,877] INFO Subscribing to /brokers/topics path to watch for new topics (kafka.server.KafkaHealthcheck$SessionExpireListener)`

Current input:

> input {  
> beats {  
> port =\> 5044  
> }  
> }

Current output:

> output {  
> elasticsearch {  
> hosts =\> ["localhost:9200"]  
> sniffing =\> true  
> manage\_template =\> false  
> index =\> "%{[@metadata][beat]}-%{+YYYY.MM.dd}"  
> document\_type =\> "%{[@metadata][type]}"  
> }  
> }

This input and output works for /var/log/messages and /var/log/secure. Any advice will be helpful, thank you.

---

<div class="post-metadata">

**Author:** ![yaauie](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/yaauie/32/23363_2.png) [@yaauie](https://discuss.elastic.co/u/yaauie)\
**Post date:** [March 17, 2018, 5:35am UTC](https://discuss.elastic.co/t/filter-for-kafka-and-zookeeper-system-logs/124276/2 "2018-03-17T05:35:12Z")

</div>

> [@jake\_smith](#):
>
> [2018-03-15 22:27:31,877] INFO Subscribing to /brokers/topics path to watch for new topics (kafka.server.KafkaHealthcheck$SessionExpireListener)

_Caveat: it's pretty tough to come up with a generic pattern that will match many messages when we're only given a single example, but I hope this is enough to get you started._

You could use the [grok constructor](http://grokconstructor.appspot.com/do/construction), but for a log message this simple I would probably start with the [dissect filter](https://www.elastic.co/guide/en/logstash/current/plugins-filters-dissect.html), followed by the [date filter](https://www.elastic.co/guide/en/logstash/current/plugins-filters-date.html) to set the event's timestamp:

```auto
filter {
  dissect {
    "mapping" => {
      "message" => "[%{[@metadata][ts]}] %{level} %{message}"
    }
  }
  date {
    "match" => ["[@metadata][ts]", "yyyy-MM-dd HH:mm:ss,SSS"]
  }
}

```

---

<div class="post-metadata">

**Author:** ![jake\_smith](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jake_smith/32/47016_2.png) [@jake\_smith](https://discuss.elastic.co/u/jake_smith)\
**Post date:** [March 21, 2018, 2:42pm UTC](https://discuss.elastic.co/t/filter-for-kafka-and-zookeeper-system-logs/124276/3 "2018-03-21T14:42:45Z")

</div>

Hey - I appreciate the reply! It's definitely enough to get me started, thank you for posting. I'll start with this and go from there.

---

<div class="post-metadata">

**Author:** ![jake\_smith](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/jake_smith/32/47016_2.png) [@jake\_smith](https://discuss.elastic.co/u/jake_smith)\
**Post date:** [March 23, 2018, 11:35am UTC](https://discuss.elastic.co/t/filter-for-kafka-and-zookeeper-system-logs/124276/4 "2018-03-23T11:35:40Z")

</div>

I've created a filter "20-filter-kafka" with the above pattern, however, nothing is coming in that I can see on Kibana.

To simplify the process, I've reduced my output file to:

> output {  
> elasticsearch {  
> hosts =\> ["localhost:9200"]  
> }  
> }

Filebeat.yml prospectors:

> filebeat.prospectors:
> 
> Each - is a prospector. Most options can be set at the prospector level, so  
> you can use different prospectors for various configurations.  
> Below are the prospector specific configurations.
> 
> - type: log

In Kibana my index pattern is:

> filebeat-\*

However, I have tried:

> "\*  
> Both with a time filter field name log\_stamp

logstash-plain.log:

> [2018-03-23T11:37:57,704][INFO][logstash.modules.scaffold] Initializing module {:module\_name=\>"fb\_apache", :directory=\>"/usr/share/logstash/modules/fb\_apache/configuration"}  
> [2018-03-23T11:37:57,712][INFO][logstash.modules.scaffold] Initializing module {:module\_name=\>"netflow", :directory=\>"/usr/share/logstash/modules/netflow/configuration"}  
> [2018-03-23T11:37:57,970][WARN][logstash.config.source.multilocal] Ignoring the 'pipelines.yml' file because modules or command line options are specified  
> [2018-03-23T11:37:58,093][INFO][logstash.runner] Starting Logstash {"logstash.version"=\>"6.2.2"}  
> [2018-03-23T11:37:58,275][INFO][logstash.agent] Successfully started Logstash API endpoint {:port=\>9600}  
> [2018-03-23T11:37:58,986][INFO][logstash.pipeline] Starting pipeline {:pipeline\_id=\>"main", "pipeline.workers"=\>2, "pipeline.batch.size"=\>125, "pipeline.batch.delay"=\>50}  
> [2018-03-23T11:37:59,456][INFO][logstash.outputs.elasticsearch] Elasticsearch pool URLs updated {:changes=\>{:removed=\>, :added=\>[[http://localhost:9200/](http://localhost:9200/)]}}  
> [2018-03-23T11:37:59,480][INFO][logstash.outputs.elasticsearch] Running health check to see if an Elasticsearch connection is working {:healthcheck\_url=\>[http://localhost:9200/](http://localhost:9200/), :path=\>"/"}  
> [2018-03-23T11:37:59,652][WARN][logstash.outputs.elasticsearch] Restored connection to ES instance {:url=\>"[http://localhost:9200/](http://localhost:9200/)"}  
> [2018-03-23T11:37:59,708][INFO][logstash.outputs.elasticsearch] ES Output version determined {:es\_version=\>nil}  
> [2018-03-23T11:37:59,709][WARN][logstash.outputs.elasticsearch] Detected a 6.x and above cluster: the `type` event field won't be used to determine the document \_type {:es\_version=\>6}  
> [2018-03-23T11:37:59,714][INFO][logstash.outputs.elasticsearch] Using mapping template from {:path=\>nil}  
> [2018-03-23T11:37:59,716][INFO][logstash.outputs.elasticsearch] Attempting to install template {:manage\_template=\>{"template"=\>"logstash-_", "version"=\>60001, "settings"=\>{"index.refresh\_interval"=\>"5s"}, "mappings"=\>{"default"=\>{"dynamic\_templates"=\>[{"message\_field"=\>{"path\_match"=\>"message", "match\_mapping\_type"=\>"string", "mapping"=\>{"type"=\>"text", "norms"=\>false}}}, {"string\_fields"=\>{"match"=\>"_", "match\_mapping\_type"=\>"string", "mapping"=\>{"type"=\>"text", "norms"=\>false, "fields"=\>{"keyword"=\>{"type"=\>"keyword", "ignore\_above"=\>256}}}}}], "properties"=\>{"@timestamp"=\>{"type"=\>"date"}, "@version"=\>{"type"=\>"keyword"}, "geoip"=\>{"dynamic"=\>true, "properties"=\>{"ip"=\>{"type"=\>"ip"}, "location"=\>{"type"=\>"geo\_point"}, "latitude"=\>{"type"=\>"half\_float"}, "longitude"=\>{"type"=\>"half\_float"}}}}}}}}  
> [2018-03-23T11:37:59,727][INFO][logstash.outputs.elasticsearch] New Elasticsearch output {:class=\>"LogStash::Outputs::Elasticsearch", :hosts=\>["[//localhost:9200](https://localhost:9200)"]}  
> [2018-03-23T11:38:00,196][INFO][logstash.inputs.beats] Beats inputs: Starting input listener {:address=\>"0.0.0.0:5044"}  
> [2018-03-23T11:38:00,373][INFO][logstash.pipeline] Pipeline started succesfully {:pipeline\_id=\>"main", :thread=\>"#\<Thread:0x2c3a9acf@/usr/share/logstash/logstash-core/lib/logstash/pipeline.rb:246 sleep\>"}  
> [2018-03-23T11:38:00,452][INFO][logstash.agent] Pipelines running {:count=\>1, :pipelines=\>["main"]}  
> [2018-03-23T11:38:00,469][INFO][org.logstash.beats.Server] Starting server on port: 5044

I've had my ELK stack working successfully with other filters so I think my connection issue is fine. /var/log/logstash/logstash-plain.log seems normal. Is there something I can paste or check that will help us figure out what's wrong?

I appreciate the help! Thank you!

-edit- It now works. I needed to create a new index pattern with logstash-\*. I'm not sure if it was always there, but I didn't see it before. I was hopeful by making a "\*index pattern would also allow the logs to show but it did not. Regardless, I've got something to work with now and for that, thank you very much for your reply earlier.

---

<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:** [April 20, 2018, 11:36am UTC](https://discuss.elastic.co/t/filter-for-kafka-and-zookeeper-system-logs/124276/5 "2018-04-20T11:36:05Z")

</div>

Closed on April 20, 2018, 11:36am UTC

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