[Zeek] Cannot send logs to their individual Kafka topics

[Arda Savran]

Hello folks:

I have successfully been able to send everything to a remote single Kafka
Topic from a local Bro machine and following is my local.bro file to make
that happen:

*##! Local site policy. Customize as appropriate.*
*##!*
*##! This file will not be overwritten when upgrading or reinstalling!*

*#@load packages*

*@load
/usr/local/bro/lib/bro/plugins/packages/metron-bro-plugin-kafka/scripts/Apache/Kafka*
*redef Kafka::send_all_active_logs = T;*
*redef Kafka::tag_json = T;*
*redef Kafka::kafka_conf = table(["metadata.broker.list"] =
"XX.XX.XX.XX:9092");*

However, when I change that to write logs to their individual Kafka topics
I get an error message under stderr.log. Following is my updated local.bro:

*##! Local site policy. Customize as appropriate.*
*##!*
*##! This file will not be overwritten when upgrading or reinstalling!*

*#@load packages*

*#@load
/usr/local/bro/lib/bro/plugins/packages/metron-bro-plugin-kafka/scripts/Apache/Kafka*
*#redef Kafka::send_all_active_logs = T;*
*#redef Kafka::tag_json = T;*
*#redef Kafka::kafka_conf = table(["metadata.broker.list"] =
"XX.XX.XX.XX:9092");*

*###########*
*###########*

*@load
/usr/local/bro/lib/bro/plugins/packages/metron-bro-plugin-kafka/scripts/Apache/Kafka*
*redef Kafka::topic_name = "";*
*redef Kafka::tag_json = T;*
*redef Kafka::debug = "all";*

*event bro_init() &priority=-10*
*{*
*# handles DNS*
*local dns_filter: Log::Filter = [*
*$name = "kafka-dns",*
*$writer = Log::WRITER_KAFKAWRITER,*
*$config = table(["metadata.broker.list"] = "XX.XX.XX.XX:9092"),*
*$path = "dns"*
*];*
*Log::add_filter(DNS::LOG, dns_filter);*
*}*

*###########*
*###########*

I enter "broctl check" and "broctl deploy" after that; but get the
following:

[root at localhost current]# tail -f stderr.log
%7|1554218121.957|STATE|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed
state DOWN -> CONNECT
%7|1554218121.957|BROADCAST|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1554218121.957|BROKERFAIL|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: failed: err:
Local: Broker transport failure: (errno: Connection refused)
%7|1554218121.957|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]:
localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection
refused
%7|1554218121.957|STATE|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed
state CONNECT -> DOWN
%7|1554218121.957|BROADCAST|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1554218121.957|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]:
localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1554218121.957|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]:
localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1554218122.309|NOINFO|rdkafka#producer-1| [thrd:main]: Topic  partition
count is zero: should refresh metadata
%7|1554218122.309|METADATA|rdkafka#producer-1| [thrd:main]: Skipping
metadata refresh of 1 topic(s): no usable brokers
%7|1554218122.957|CONNECT|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: broker in state
DOWN connecting
%7|1554218122.958|CONNECT|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Connecting to
ipv4#127.0.0.1:9092 (plaintext) with socket 29
%7|1554218122.958|STATE|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed
state DOWN -> CONNECT
%7|1554218122.958|BROADCAST|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1554218122.958|BROKERFAIL|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: failed: err:
Local: Broker transport failure: (errno: Connection refused)
%7|1554218122.958|FAIL|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]:
localhost:9092/bootstrap: Connect to ipv4#127.0.0.1:9092 failed: Connection
refused
%7|1554218122.958|STATE|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Broker changed
state CONNECT -> DOWN
%7|1554218122.958|BROADCAST|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: Broadcasting state change
%7|1554218122.958|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]:
localhost:9092/bootstrap: Purging bufq with 0 buffers
%7|1554218122.958|BUFQ|rdkafka#producer-1| [thrd:localhost:9092/bootstrap]:
localhost:9092/bootstrap: Updating 0 buffers on connection reset
%7|1554218122.958|RECONNECT|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Delaying next
reconnect by 301ms
%7|1554218123.259|RECONNECT|rdkafka#producer-1|
[thrd:localhost:9092/bootstrap]: localhost:9092/bootstrap: Delaying next
reconnect by 53ms
%7|1554218123.309|NOINFO|rdkafka#producer-1| [thrd:main]: Topic  partition
count is zero: should refresh metadata

Yes, I have iptables enabled on the local bro machine but it works with the
first configuration option file. How come bro thinks that the kafka broker
is local. It is supposed to send the messages to XX.XX.XX.XX.

Thanks in advance.
-------------- next part --------------
An HTML attachment was scrubbed...
URL: http://mailman.ICSI.Berkeley.EDU/pipermail/zeek/attachments/20190402/1483383a/attachment.html