Logging Bro Output to Kafka

Example 1

The goal in this example is to send all HTTP and DNS records to a Kafka topic named bro.

• Any configuration value accepted by librdkafka can be added to the kafka_conf configuration table.
• By defining topic_name all records will be sent to the same Kafka topic.
• Defining logs_to_send will ensure that only HTTP and DNS records are sent.

@load Bro/Kafka/logs-to-kafka.bro
redef Kafka::logs_to_send = set(HTTP::LOG, DNS::LOG);
redef Kafka::topic_name = "bro";
redef Kafka::kafka_conf = table(
    ["metadata.broker.list"] = "localhost:9092"
);

Example 2

It is also possible to send each log stream to a uniquely named topic. The goal in this example is to send all HTTP records to a Kafka topic named http and all DNS records to a separate Kafka topic named dns.

• The topic_name value must be set to an empty string.
• The $path value of Bro’s Log Writer mechanism is used to define the topic name.
• Any configuration value accepted by librdkafka can be added to the $config configuration table.
• Each log writer accepts a separate configuration table.

@load Bro/Kafka/logs-to-kafka.bro
redef Kafka::topic_name = "";
redef Kafka::tag_json = T;

event bro_init()
{
    # handles HTTP
    local http_filter: Log::Filter = [
        $name = "kafka-http",
        $writer = Log::WRITER_KAFKAWRITER,
        $config = table(
                ["metadata.broker.list"] = "localhost:9092"
        ),
        $path = "http"
    ];
    Log::add_filter(HTTP::LOG, http_filter);

    # handles DNS
    local dns_filter: Log::Filter = [
        $name = "kafka-dns",
        $writer = Log::WRITER_KAFKAWRITER,
        $config = table(
                ["metadata.broker.list"] = "localhost:9092"
        ),
        $path = "dns"
    ];
    Log::add_filter(DNS::LOG, dns_filter);
}

Example 3

You may want to configure bro to filter log messages with certain characteristics from being sent to your kafka topics. For instance, Metron currently doesn’t support IPv6 source or destination IPs in the default enrichments, so it may be helpful to filter those log messages from being sent to kafka (although there are multiple ways to approach this). In this example we will do that that, and are assuming a somewhat standard bro kafka plugin configuration, such that:

• All bro logs are sent to the bro topic, by configuring Kafka::topic_name.
• Each JSON message is tagged with the appropriate log type (such as http, dns, or conn), by setting tag_json to true.
• If the log message contains a 128 byte long source or destination IP address, the log is not sent to kafka.

@load Bro/Kafka/logs-to-kafka.bro
redef Kafka::topic_name = "bro";
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"] = "localhost:9092"
        )
    ]);

    # 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"] = "localhost:9092"
        )
    ]);

    # 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"] = "localhost:9092"
        )
    ]);
}

kafka_conf

The global configuration settings for Kafka. These values are passed through directly to librdkafka. Any valid librdkafka settings can be defined in this table. The full set of valid librdkafka settings are available here.

redef Kafka::kafka_conf = table(
    ["metadata.broker.list"] = "localhost:9092",
    ["client.id"] = "bro"
);

topic_name

The name of the topic in Kafka where all Bro logs will be sent to.

redef Kafka::topic_name = "bro";

max_wait_on_shutdown

The maximum number of milliseconds that the plugin will wait for any backlog of queued messages to be sent to Kafka before forced shutdown.

redef Kafka::max_wait_on_shutdown = 3000;

tag_json

If true, a log stream identifier is appended to each JSON-formatted message. For example, a Conn::LOG message will look like { 'conn' : { ... }}.

redef Kafka::tag_json = T;

debug

A comma separated list of debug contexts in librdkafka which you want to enable. The available contexts are:

• generic
• broker
• topic
• metadata
• queue
• msg
• protocol
• cgrp
• security
• fetch
• feature
• all

SASL

If you are using SASL as a security protocol for kafka, then you must have libsasl or libsasl2 installed. You can tell if sasl is enabled by running the following from the directory in which you have build librdkafka:

examples/rdkafka_example -X builtin.features
builtin.features = gzip,snappy,ssl,sasl,regex

Producer Config

As stated above, you can configure the producer kafka configs in ${BRO_HOME}/share/bro/site/local.bro. There are a few configs necessary to set, which are described here. For an environment where the following is true:

• The broker is node1:6667
• This kafka is using SASL_PLAINTEXT as the security protocol
• The keytab used is the metron keytab
• The service principal for metron is metron@EXAMPLE.COM

The kafka topic bro has been given permission for the metron user to write:

# login using the metron user
kinit -kt /etc/security/keytabs/metron.headless.keytab metron@EXAMPLE.COM
${KAFKA_HOME}/kafka-broker/bin/kafka-acls.sh --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=node1:2181 --add --allow-principal User:metron --topic bro

The following is how the ${BRO_HOME}/share/bro/site/local.bro looks:

@load Bro/Kafka/logs-to-kafka.bro
redef Kafka::logs_to_send = set(HTTP::LOG, DNS::LOG);
redef Kafka::topic_name = "bro";
redef Kafka::tag_json = T;
redef Kafka::kafka_conf = table( ["metadata.broker.list"] = "node1:6667"
                               , ["security.protocol"] = "SASL_PLAINTEXT"
                               , ["sasl.kerberos.keytab"] = "/etc/security/keytabs/metron.headless.keytab"
                               , ["sasl.kerberos.principal"] = "metron@EXAMPLE.COM"
                               );