metron-bro-kafka not producing logs

[#1]

Hi, i am trying to push my bro logs to kafka, i can see bro logs

	tail -f /nsm/bro/logs/current/conn.log

"25.1504200tcp-1.7481615280SHFF0ScADF579600(empty)''

But it is not going to kafka topic, i am using bro kafka pugin from metron repository "https://github.com/apache/metron-bro-plugin-kafka"

My local.bro file is:

@load /nsm/bro/lib/bro/plugins/APACHE_KAFKA/scripts/
redef Kafka::topic_name = "bro-new";
redef Kafka::tag_json = T;
event bro_init() &priority=-5
{
	    #handles HTTP
    Log::add_filter(HTTP::LOG, [
        $name = "kafka-http",
        $writer = Log::WRITER_KAFKAWRITER,
	        $pred(rec: HTTP::Info) = { return ! (( |rec$id$orig_h| == 128 || |rec$id$resp_h| == 128 )); },
    $config = table(["metadata.broker.list"] = "kafkaip:6667")
    ]);

[#2]

You have only configured the plugin to push HTTP logs to Kafka; not Conn logs. If you expect to push the Conn logs, then configure those to be sent like Example 3 in the README. Or just start with a simpler configuration like this, which will send only the Conn logs.

@load packages/metron-bro-plugin-kafka/Apache/Kafka
redef Kafka::logs_to_send = set(Conn::LOG);
redef Kafka::topic_name = "bro";
redef Kafka::kafka_conf = table(
	["metadata.broker.list"] = "kafkaip:6667")
);

[#3 · Accepted solution]

solved this, i was not over writing rdkafka defaults, doing this fixed my problem.
working local.bro

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


# This script logs which scripts were loaded during each run.
@load misc/loaded-scripts


# Apply the default tuning scripts for common tuning settings.
@load tuning/defaults


# Estimate and log capture loss.
@load misc/capture-loss


# Enable logging of memory, packet and lag statistics.
@load misc/stats


# Load the scan detection script.
@load misc/scan


# Detect traceroute being run on the network. This could possibly cause
# performance trouble when there are a lot of traceroutes on your network.
# Enable cautiously.
#@load misc/detect-traceroute


# Generate notices when vulnerable versions of software are discovered.
# The default is to only monitor software found in the address space defined
# as "local".  Refer to the software framework's documentation for more
# information.
@load frameworks/software/vulnerable


# Detect software changing (e.g. attacker installing hacked SSHD).
@load frameworks/software/version-changes


# This adds signatures to detect cleartext forward and reverse windows shells.
@load-sigs frameworks/signatures/detect-windows-shells


# Load all of the scripts that detect software in various protocols.
@load protocols/ftp/software
@load protocols/smtp/software
@load protocols/ssh/software
@load protocols/http/software
# The detect-webapps script could possibly cause performance trouble when
# running on live traffic.  Enable it cautiously.
#@load protocols/http/detect-webapps


# This script detects DNS results pointing toward your Site::local_nets
# where the name is not part of your local DNS zone and is being hosted
# externally.  Requires that the Site::local_zones variable is defined.
@load protocols/dns/detect-external-names


# Script to detect various activity in FTP sessions.
@load protocols/ftp/detect


# Scripts that do asset tracking.
@load protocols/conn/known-hosts
@load protocols/conn/known-services
@load protocols/ssl/known-certs


# This script enables SSL/TLS certificate validation.
@load protocols/ssl/validate-certs


# This script prevents the logging of SSL CA certificates in x509.log
@load protocols/ssl/log-hostcerts-only


# Uncomment the following line to check each SSL certificate hash against the ICSI
# certificate notary service; see http://notary.icsi.berkeley.edu .
# @load protocols/ssl/notary


# If you have libGeoIP support built in, do some geographic detections and
# logging for SSH traffic.
@load protocols/ssh/geo-data
# Detect hosts doing SSH bruteforce attacks.
@load protocols/ssh/detect-bruteforcing
# Detect logins using "interesting" hostnames.
@load protocols/ssh/interesting-hostnames


# Detect SQL injection attacks.
@load protocols/http/detect-sqli


#### Network File Handling ####


# Enable MD5 and SHA1 hashing for all files.
@load frameworks/files/hash-all-files


# Detect SHA1 sums in Team Cymru's Malware Hash Registry.
@load frameworks/files/detect-MHR


# Uncomment the following line to enable detection of the heartbleed attack. Enabling
# this might impact performance a bit.
# @load policy/protocols/ssl/heartbleed


# Uncomment the following line to enable logging of connection VLANs. Enabling
# this adds two VLAN fields to the conn.log file.
# @load policy/protocols/conn/vlan-logging


# Uncomment the following line to enable logging of link-layer addresses. Enabling
# this adds the link-layer address for each connection endpoint to the conn.log file.
# @load policy/protocols/conn/mac-logging


# Uncomment the following line to enable the SMB analyzer.  The analyzer
# is currently considered a preview and therefore not loaded by default.
# @load policy/protocols/smb
@load /nsm/bro/lib/bro/plugins/APACHE_KAFKA/scripts/
redef Kafka::topic_name = "bro-new";
redef Kafka::tag_json = T;
redef Kafka::kafka_conf = table(
    ["metadata.broker.list"] = "10.162.96.32:6667",
    ["client.id"] = "bro"
);
event bro_init() &priority=-5
{
    # handles HTTP
    Log::add_filter(HTTP::LOG, [
        $name = "kafka-http",
        $writer = Log::WRITER_KAFKAWRITER,
        $pred(rec: HTTP::Info) = { return ! (( |rec$id$orig_h| == 128 || |rec$id$resp_h| == 128 )); },
        $config = table(
            ["metadata.broker.list"] = "10.162.96.32:6667"
        )
    ]);


    # handles DNS
    Log::add_filter(DNS::LOG, [
        $name = "kafka-dns",
        $writer = Log::WRITER_KAFKAWRITER,
        $pred(rec: DNS::Info) = { return ! (( |rec$id$orig_h| == 128 || |rec$id$resp_h| == 128 )); },
        $config = table(
            ["metadata.broker.list"] = "10.162.96.32:6667"
        )
    ]);


    # handles Conn
    Log::add_filter(Conn::LOG, [
        $name = "kafka-conn",
        $writer = Log::WRITER_KAFKAWRITER,
        $pred(rec: Conn::Info) = { return ! (( |rec$id$orig_h| == 128 || |rec$id$resp_h| == 128 )); },
        $config = table(
            ["metadata.broker.list"] = "10.162.96.32:6667"
        )
    ]);
}