配置 Kafka 输出

编辑

Kafka 输出将事件发送到 Apache Kafka。

要使用此输出,请编辑 Packetbeat 配置文件,通过注释掉 Elasticsearch 输出禁用它,并通过取消注释 Kafka 部分启用 Kafka 输出。

对于 Kafka 0.10.0.0+ 版本,消息创建时间戳由 Beats 设置,并且等于事件的初始时间戳。这会影响 Kafka 中的保留策略:例如,如果一个 Beats 事件是 2 周前创建的,保留策略设置为 7 天,并且来自 Beats 的消息今天到达 Kafka,则它将立即被丢弃,因为时间戳值早于过去 7 天。可以通过在消息到达时设置时间戳来更改此行为,这样消息就不会被丢弃,而是保留 7 天。为此,请在 Kafka 配置中将 log.message.timestamp.type 设置为 LogAppendTime(默认 CreateTime)。

配置示例

output.kafka:
  # initial brokers for reading cluster metadata
  hosts: ["kafka1:9092", "kafka2:9092", "kafka3:9092"]

  # message topic selection + partitioning
  topic: '%{[fields.log_topic]}'
  partition.round_robin:
    reachable_only: false

  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

大于 max_message_bytes 的事件将被丢弃。要避免此问题,请确保 Packetbeat 不会生成大于 max_message_bytes 的事件。

兼容性

编辑

此输出可以连接到 Kafka 0.8.2.0 及更高版本。旧版本也可能有效,但不受支持。

配置选项

编辑

您可以在 packetbeat.yml 配置文件的 kafka 部分中指定以下选项

enabled

编辑

enabled 配置是一个布尔值设置,用于启用或禁用输出。如果设置为 false,则禁用输出。

默认值为 true

hosts

编辑

用于获取集群元数据的 Kafka 代理地址列表。集群元数据包含实际发布事件的 Kafka 代理。

version

编辑

Packetbeat 连接时将请求的 Kafka 协议版本。默认为 1.0.0。

有效值是 0.8.2.02.6.0 之间的所有 Kafka 版本。

协议版本控制 Packetbeat 可用的 Kafka 客户端功能;它不会阻止 Packetbeat 连接到比协议版本更新的 Kafka 版本。

有关支持的版本的信息,请参阅 兼容性

username

编辑

用于连接到 Kafka 的用户名。如果配置了用户名,则必须同时配置密码。

password

编辑

用于连接到 Kafka 的密码。

sasl.mechanism

编辑

连接到 Kafka 时使用的 SASL 机制。它可以是以下之一

  • PLAIN,用于 SASL/PLAIN。
  • SCRAM-SHA-256,用于 SCRAM-SHA-256。
  • SCRAM-SHA-512,用于 SCRAM-SHA-512。

如果未设置 sasl.mechanism,则如果提供了 usernamepassword,则使用 PLAIN。否则,禁用 SASL 身份验证。

要使用 GSSAPI 机制通过 Kerberos 进行身份验证,您必须将此字段留空,并使用 kerberos 选项。

topic

编辑

用于生成事件的 Kafka 主题。

您可以使用格式字符串访问任何事件字段来动态设置主题。例如,此配置使用自定义字段 fields.log_topic 为每个事件设置主题

topic: '%{[fields.log_topic]}'

要了解如何将自定义字段添加到事件,请参阅 fields 选项。

有关其他动态设置主题的方法,请参阅 topics 设置。

topics

编辑

主题选择器规则的数组。每个规则指定用于匹配该规则的事件的 topic。在发布期间,Packetbeat 根据数组中第一个匹配的规则为每个事件设置 topic。规则可以包含条件、基于格式字符串的字段和名称映射。如果缺少 topics 设置或没有规则匹配,则使用 topic 字段。

规则设置

topic
要使用的主题格式字符串。如果此字符串包含字段引用,例如 %{[fields.name]},则字段必须存在,否则规则将失败。
mappings
一个字典,它获取 topic 返回的值并将其映射到新名称。
default
如果 mappings 未找到匹配项,则使用的默认字符串值。
when
必须成功才能执行当前规则的条件。处理器支持的所有 条件 也在此处支持。

以下示例根据消息字段是否包含指定的字符串来设置主题

output.kafka:
  hosts: ["localhost:9092"]
  topic: "logs-%{[agent.version]}"
  topics:
    - topic: "critical-%{[agent.version]}"
      when.contains:
        message: "CRITICAL"
    - topic: "error-%{[agent.version]}"
      when.contains:
        message: "ERR"

此配置将生成名为 critical-8.17.0error-8.17.0logs-8.17.0 的主题。

可选格式化字符串,指定 Kafka 事件键。如果配置了,可以使用格式字符串从事件中提取事件键。

有关特定键的选择的含义,请参阅 Kafka 文档;默认情况下,键由 Kafka 集群选择。

partition

编辑

Kafka 输出代理事件分区策略。必须是 randomround_robinhash 之一。默认情况下,使用 hash 分区器。

random.group_events:设置在分区器随机选择新分区之前,要发布到同一分区的事件数。默认值为 1,这意味着在每个事件之后,将随机选择一个新分区。

round_robin.group_events:设置在分区器选择下一个分区之前,要发布到同一分区的事件数。默认值为 1,这意味着在每个事件之后,将选择下一个分区。

hash.hash:用于计算分区哈希值的字段列表。如果未配置字段,则将使用事件的 key 值。

hash.random:如果无法计算哈希或键值,则随机分配事件。

默认情况下,所有分区器都将尝试将事件发布到所有分区。如果某个分区的领导者对于 Beats 变得不可访问,则输出可能会阻塞。所有分区器都支持设置 reachable_only 以覆盖此行为。如果 reachable_only 设置为 true,则事件将仅发布到可用的分区。

发布到可用分区的子集可能会增加资源使用量,因为事件可能会分布不均匀。

headers

编辑

标头是一个键值对,并且可以使用相同的 key 包括多个标头。仅支持字符串值。这些标头将包含在每个生成的 Kafka 消息中。

output.kafka:
  hosts: ["localhost:9092"]
  topic: "logs-%{[agent.version]}"
  headers:
    - key: "some-key"
      value: "some value"
    - key: "another-key"
      value: "another value"

client_id

编辑

用于日志记录、调试和审计的可配置 ClientID。默认为“beats”。

codec

编辑

输出编解码器配置。如果缺少 codec 部分,则事件将以 json 编码。

有关更多信息,请参阅 更改输出编解码器

metadata

编辑

Kafka 元数据更新设置。元数据包含有关要用于发布的代理、主题、分区和活动领导者的信息。

refresh_frequency
元数据刷新间隔。默认为 10 分钟。
full
获取元数据时使用的策略,当此选项为 true 时,客户端将为所有可用主题维护完整的元数据集,如果此选项设置为 false,则仅刷新已配置主题的元数据。默认为 false。
retry.max
当集群处于领导者选举中间时,元数据更新重试的总次数。默认为 3。
retry.backoff
在领导者选举期间重试之间的等待时间。默认为 250 毫秒。

max_retries

编辑

发布失败后重试发布事件的次数。在指定的重试次数之后,事件通常会被丢弃。

max_retries 设置为小于 0 的值,以重试直到所有事件都发布为止。

默认值为 3。

backoff.init

编辑

在发生网络错误后尝试重新发布到 Kafka 之前等待的秒数。等待 backoff.init 秒后,Packetbeat 将尝试重新发布。如果尝试失败,则回退计时器将呈指数增长,直至 backoff.max。成功发布后,将重置回退计时器。默认值为 1 秒。

backoff.max

编辑

在网络错误后尝试重新发布到 Kafka 之前等待的最大秒数。默认值为 60 秒。

bulk_max_size

编辑

单个 Kafka 请求中批量处理的最大事件数。默认值为 2048。

bulk_flush_frequency

编辑

发送批量 Kafka 请求之前等待的时间。0 表示无延迟。默认值为 0。

timeout

编辑

等待 Kafka 代理响应的超时秒数。默认值为 30 秒。

broker_timeout

编辑

代理等待所需 ACK 数量的最大持续时间。默认值为 10 秒。

channel_buffer_size

编辑

每个 Kafka 代理在输出管道中缓冲的消息数量。默认值为 256。

keep_alive

编辑

活动网络连接的保持活动时间。如果为 0 秒,则禁用保持活动。默认值为 0 秒。

compression

编辑

设置输出压缩编解码器。必须是 nonesnappylz4gzipzstd 之一。默认值为 gzip

Azure 事件中心 Kafka 的已知问题

当以 Azure 事件中心的 Kafka 为目标时,请将 compression 设置为 none,因为提供的编解码器不受支持。

compression_level

编辑

设置 gzip 使用的压缩级别。将此值设置为 0 将禁用压缩。压缩级别必须在 1(最佳速度)到 9(最佳压缩)的范围内。

提高压缩级别将减少网络使用量,但会增加 CPU 使用量。

默认值为 4。

max_message_bytes

编辑

JSON 编码消息允许的最大大小。更大的消息将被丢弃。默认值为 1000000(字节)。此值应等于或小于代理的 message.max.bytes

required_acks

编辑

从代理所需的 ACK 可靠性级别。0=无响应,1=等待本地提交,-1=等待所有副本提交。默认值为 1。

注意:如果设置为 0,Kafka 不会返回任何 ACK。错误时消息可能会被静默丢失。

SSL 参数的配置选项,如 Kafka 连接的根 CA。Kafka 主机密钥库应使用 -keyalg RSA 参数创建,以确保它使用 Filebeat 的 Kafka 库支持的密码。有关更多信息,请参阅 SSL

kerberos

编辑

此功能处于测试阶段,可能会发生更改。其设计和代码不如官方 GA 功能成熟,因此按原样提供,不提供任何保证。测试版功能不受官方 GA 功能的支持 SLA 的约束。

Kerberos 身份验证的配置选项。

有关更多信息,请参阅 Kerberos