# Suricata logs to Filebeat to Kafka topics, by event-type

**URL:** <https://discuss.elastic.co/t/suricata-logs-to-filebeat-to-kafka-topics-by-event-type/215179>\
**Category:** Beats\
**Tags:** filebeat\
**Created:** [January 15, 2020, 5:10pm UTC](https://discuss.elastic.co/t/suricata-logs-to-filebeat-to-kafka-topics-by-event-type/215179 "2020-01-15T17:10:13Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![driekhof](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/driekhof/32/60845_2.png) [@driekhof](https://discuss.elastic.co/u/driekhof)\
**Post date:** [January 15, 2020, 5:10pm UTC](https://discuss.elastic.co/t/suricata-logs-to-filebeat-to-kafka-topics-by-event-type/215179/1 "2020-01-15T17:10:13Z")

</div>

Hello, new to Filebeat. I have a simplified case working: Suricata to Filebeat to Kafka if I hard code one kafka topic name in filebeat.yml. But I'd like to make Filebeat dynamically route events to different topics based on their Suricata event type.

I've taken several wild guesses, but I don't really understand how dynamic filebeat fields work, or what I need to do to set them up. Here's one attempt after enabling the suricata module:

```auto
topic: 'suricata-%{[fields.suricata.eve.event_type]}'

```

The filebeat log just gives me this:

```auto
2020-01-14T23:44:49.550Z INFO kafka/log.go:53 kafka message: Initializing new client
2020-01-14T23:44:49.551Z INFO kafka/log.go:53 kafka message: Successfully initialized new client
2020-01-14T23:44:49.551Z INFO pipeline/output.go:105 Connection to kafka(somehost:9092) established
2020-01-14T23:44:49.551Z ERROR kafka/client.go:144 Dropping event: no topic could be selected
2020-01-14T23:44:49.551Z ERROR kafka/client.go:144 Dropping event: no topic could be selected

```

etc.....

Is this kind of thing possible with filebeat, and if so how? An example would be ideal.

---

<div class="post-metadata">

**Author:** ![shaunak](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/shaunak/32/6643_2.png) [@shaunak](https://discuss.elastic.co/u/shaunak)\
**Post date:** [January 15, 2020, 10:00pm UTC](https://discuss.elastic.co/t/suricata-logs-to-filebeat-to-kafka-topics-by-event-type/215179/2 "2020-01-15T22:00:33Z")

</div>

Hi @driekhof,

Welcome to the Elastic community forums!

Yes, you can use formatted strings in the `kafka` output's `topic` setting. See [https://www.elastic.co/guide/en/beats/filebeat/master/kafka-output.html#topic-option-kafka](https://www.elastic.co/guide/en/beats/filebeat/master/kafka-output.html#topic-option-kafka).

One way to figure out what fields are available would be to temporarily disable the `kafka` output and enable the `console` output. This prints out the events that would be sent as JSON. You can then inspect them to figure out which fields are available and their paths.

In your case, this should work:

```auto
topic: 'suricata-%{[suricata.eve.event_type]}'

```

Shaunak

---

<div class="post-metadata">

**Author:** ![driekhof](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/driekhof/32/60845_2.png) [@driekhof](https://discuss.elastic.co/u/driekhof)\
**Post date:** [January 15, 2020, 11:01pm UTC](https://discuss.elastic.co/t/suricata-logs-to-filebeat-to-kafka-topics-by-event-type/215179/3 "2020-01-15T23:01:32Z")

</div>

> [@shaunak](#):
>
> topic: 'suricata-%{[suricata.eve.event\_type]}'

Thanks, that worked (I was so close, doh!). And thanks for the debugging tip. Now I just need to figure out why I only get stats event\_type...

---

<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:** [February 12, 2020, 11:01pm UTC](https://discuss.elastic.co/t/suricata-logs-to-filebeat-to-kafka-topics-by-event-type/215179/4 "2020-02-12T23:01:32Z")

</div>

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