配置 Kafka 输出

Elastic Stack

注意

supported deployment methods

Fleet 管理的 APM Server 尚不支持 Kafka 输出。

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

要使用此输出，请编辑 APM Server 配置文件，注释掉 Elasticsearch 输出以禁用它，并取消注释 Kafka 部分以启用 Kafka 输出。

配置示例

		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 的事件将被丢弃。为避免此问题，请确保 APM Server 不会生成大于 max_message_bytes 的事件。

兼容性

此输出适用于 0.11 和 2.2.2 之间的所有 Kafka 版本。较旧的版本可能也可以工作，但不提供支持。

配置选项

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

`enabled（已启用）`

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

默认值为 false。

`hosts（主机）`

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

`version（版本）`

APM Server 假定 Kafka 版本针对此运行。默认为 1.0.0。

如果启用了 0.10.0.0+ 版本，则将添加事件时间戳。

有效值是 0.8.2.0 和 2.0.0 之间的所有 Kafka 版本。

有关支持的版本的详细信息，请参阅兼容性。

`username（用户名）`

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

`password（密码）`

用于连接到 Kafka 的密码。

`sasl.mechanism（sasl.机制）`

警告

此功能目前为 Beta 版，可能会发生更改。与正式 GA 功能相比，其设计和代码不够成熟，并且按原样提供，不提供任何担保。 Beta 版功能不受正式 GA 功能的支持 SLA 的约束。

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

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

如果未设置 sasl.mechanism，则在提供 username 和 password 时使用 PLAIN。否则，SASL 身份验证将被禁用。

`topic（主题）`

用于生成事件的 Kafka 主题。

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

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

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

`topics（主题）`

主题选择器规则的数组。每个规则都指定用于匹配规则的事件的 topic。在发布期间，APM Server 根据数组中第一个匹配的规则设置每个事件的 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-9.0.0*、*error-9.0.0* 和 *logs-9.0.0* 的主题。

`key（键）`

可选的格式化字符串，用于指定 Kafka 事件键。如果已配置，则可以使用格式字符串从事件中提取事件键。

有关特定键选择的影响，请参阅 Kafka 文档；默认情况下，键由 Kafka 集群选择。

`partition（分区）`

Kafka 输出代理事件分区策略。必须是 random、round_robin 或 hash 之一。默认情况下，使用 hash 分区器。

random.group_events：设置在分区器随机选择新分区之前，要发布到同一分区的事件数。默认值为 1，表示在每个事件之后，随机选择一个新分区。

round_robin.group_events：设置在分区器选择下一个分区之前，要发布到同一分区的事件数。默认值为 1，表示在每个事件之后，将选择下一个分区。

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

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

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

注意

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

`client_id（客户端 ID）`

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

`worker（工作进程）`

并发的负载均衡 Kafka 输出工作进程的数量。

`codec（编解码器）`

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

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

`metadata（元数据）`

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

refresh_frequency（刷新频率）: 元数据刷新间隔。默认为 10 分钟。
full（完整）: 用于获取元数据的策略，如果此选项为 true，则客户端将维护所有可用主题的完整元数据集合，如果此选项设置为 false，则它只会刷新配置主题的元数据。默认值为 false。
retry.max（重试.最大）: 当集群处于领导者选举中间时，元数据更新重试的总次数。默认为 3。
retry.backoff（重试.退避）: 领导者选举期间重试之间的等待时间。默认为 250ms。