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

[driekhof (Darrel Riekhof)]

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:

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

The filebeat log just gives me this:

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.

[shaunak (Shaunak Kashyap)]

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.

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:

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

Shaunak

[driekhof (Darrel Riekhof)]

@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...