Logging Zeek 输出至 Kafka 技术文档

2024-12-23 03:58:47作者：翟江哲Frasier

以下是一份详细的技术文档，介绍了如何安装、使用以及配置Logging Zeek Output to Kafka项目。本文档涵盖以下部分：

安装指南
项目使用说明
项目API使用文档
项目安装方式

1. 安装指南

`zkg` 安装方式

zkg是推荐的安装机制，因为它将动态检索、构建、测试并加载插件。请注意，在安装后，您仍需要激活和配置插件。

安装librdkafka，Kafka的原生客户端库。本插件经过对librdkafka v1.4.2版本的测试。

如果要在Kerberized Kafka环境中使用此插件，还需要安装libsasl2，并在configure脚本中传递--enable-sasl参数。
```
$ curl -L https://github.com/edenhill/librdkafka/archive/v1.4.2.tar.gz | tar xvz
$ cd librdkafka-1.4.2/
$ ./configure --enable-sasl
$ make
$ sudo make install
```
按照此处的快速入门指南配置zkg。

使用zkg install命令安装插件。

$ zkg install apache/metron-bro-plugin-kafka --version master

运行以下命令以确保插件安装成功。
```
$ zeek -N Apache::Kafka
```

手动安装方式

只有在安装和配置zkg不合理的情况下，才应手动安装插件。如果您运行的zeek环境没有Internet连接，请研究bundles或创建内部包源。

以下说明如果对您将此作为包（如deb或rpm）分发也可能有所帮助。

安装librdkafka，Kafka的原生客户端库。本插件经过对librdkafka v1.4.2版本的测试。

如果要在Kerberized Kafka环境中使用此插件，还需要安装libsasl2，并在configure脚本中传递--enable-sasl参数。
```
$ curl -L https://github.com/edenhill/librdkafka/archive/v1.4.2.tar.gz | tar xvz
$ cd librdkafka-1.4.2/
$ ./configure --enable-sasl
$ make
$ sudo make install
```

使用以下命令构建插件。

$ ./configure --with-librdkafka=$librdkafka_root
$ make
$ sudo make install

运行以下命令以确保插件安装成功。
```
$ zeek -N Apache::Kafka
```

2. 项目使用说明

以下示例展示了使用该插件的不同方式。只需将以下zeek脚本语言添加到您的local.zeek文件中，以演示示例。

在运行Zeek集群时，强烈建议利用一个或多个Zeek日志记录器，如此处所示，以将日志活动与manager线程分离。

示例1 - 将日志列表发送到Kafka

此示例的目标是将所有HTTP和DNS记录发送到名为zeek的Kafka主题。

@load packages/metron-bro-plugin-kafka/Apache/Kafka
redef Kafka::logs_to_send = set(HTTP::LOG, DNS::LOG);
redef Kafka::kafka_conf = table(
    ["metadata.broker.list"] = "server1.example.com:9092,server2.example.com:9092"
);

示例2 - 发送所有活动日志

此插件具有将所有活动日志发送到“zeek”Kafka主题的功能。

@load packages/metron-bro-plugin-kafka/Apache/Kafka
redef Kafka::send_all_active_logs = T;
redef Kafka::kafka_conf = table(
    ["metadata.broker.list"] = "localhost:9092"
);

示例3 - 发送所有活动日志与排除列表

您还可以指定一个黑名单，以确保某些日志不会被发送到Kafka，无论Kafka::send_all_active_logs和Kafka::logs_to_send配置如何。在此示例中，我们将发送除Conn日志之外的所有启用日志。

@load packages/metron-bro-plugin-kafka/Apache/Kafka
redef Kafka::send_all_active_logs = T;
redef Kafka::logs_to_exclude = set(Conn::LOG);
redef Kafka::topic_name = "zeek";
redef Kafka::kafka_conf = table(
    ["metadata.broker.list"] = "localhost:9092"
);

示例4 - 每个日志流发送到唯一主题

也可以将每个日志流发送到唯一命名的主题。在此示例中，将所有HTTP记录发送到名为http的Kafka主题，将所有DNS记录发送到名为dns的单独Kafka主题。

@load packages/metron-bro-plugin-kafka/Apache/Kafka
redef Kafka::topic_name = "";
redef Kafka::tag_json = T;

event zeek_init() &priority=-10
{
    # 处理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);

    # 处理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);
}

示例5 - Zeek日志过滤

您可能需要配置zeek以过滤掉具有特定特征的日志消息，使其不被发送到kafka主题。例如，Apache Metron目前不支持默认增强中的IPv6源或目标IP地址，因此从Kafka中过滤这些日志消息可能会有所帮助（尽管有多种方法可以解决这个问题）。在此示例中，我们将执行此操作，并假设以下标准zeek kafka插件配置：

所有zeek日志都发送到默认的zeek主题。
每个JSON消息都标记有相应的日志类型（例如http、dns或conn），通过将Kafka::tag_json设置为true。
如果日志消息包含128字节长的源或目标IP地址，则该日志不会被发送到Kafka。

@load packages/metron-bro-plugin-kafka/Apache/Kafka
redef Kafka::tag_json = T;

event zeek_init() &priority=-10
{
    # 处理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"
        )
    ]);

    # 处理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"
        )
    ]);

    # 处理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"
        )
    ]);
}

注意

logs_to_send与$pred互斥，因此对于每个要设置$pred的日志，您必须单独设置Log::add_filter，并避免将该日志包括在logs_to_send中。
可以使用is_v6_addr()函数在您的$pred中识别IP地址是否为IPv6。
或者，如果您使用Apache Metron来拉取数据，您可以采取其他方法。