« Logstash 输出设置 远程 Elasticsearch 输出 »
Elastic 文档 ›舰队和 Elastic Agent 指南 [8.14] ›在舰队中集中管理 Elastic Agent ›舰队设置

Kafka 输出设置

编辑

Kafka 输出设置编辑

指定这些设置以通过安全连接将数据发送到 Kafka。在舰队 输出设置 中，确保选择了 Kafka 输出类型。

常规设置编辑

Kafka 版本

Elastic Agent 在连接时将请求的 Kafka 协议版本。默认为 1.0.0。目前支持从 0.8.2.0 到 2.6.0 的 Kafka 版本，但是当选择版本 2.6.0 时，最新的 Kafka 版本 (3.x.x) 预计是兼容的。

主机

您的 Elastic Agent 将用于连接到一个或多个 Kafka 代理的地址。使用格式 主机:端口（不带任何协议 http://）。单击 添加行 以指定其他地址。

示例

  • localhost:9092
  • mykafkahost:9092

有关默认端口和其他配置详细信息，请参阅 舰队服务器 文档。
身份验证设置编辑

选择 Elastic Agent 用于向 Kafka 进行身份验证的机制。

无

Elastic Agent 和 Kafka 之间不使用身份验证。这是默认选项。在生产环境中，建议选择一种身份验证方法。

明文

设置此选项可将 Elastic Agent 和 Kafka 之间的流量作为明文发送，不使用任何传输层安全。

这是未设置任何身份验证时的默认选项。

加密

设置此选项可使 Elastic Agent 和 Kafka 之间的流量使用传输层安全。

选择 加密 后，服务器 SSL 证书颁发机构 和 验证模式 模式选项将变为可用。

用户名 / 密码

使用用户名和密码连接到 Kafka。

提供您的用户名和密码，并为您的登录凭据选择一个 SASL（简单身份验证和安全层）机制。

启用 SCRAM 后，Elastic Agent 使用 SCRAM 机制对用户凭据进行身份验证。SCRAM 基于 IETF RFC5802 标准，该标准描述了一种用于对用户进行身份验证的质询-响应机制。

  • Plain - 不使用 SCRAM 进行身份验证
  • SCRAM-SHA-256 - 使用 SHA-256 哈希函数
  • SCRAM-SHA-512 - 使用 SHA-512 哈希函数

为了防止未经授权的访问，您的 Kafka 密码存储为一个秘密值。虽然建议使用秘密存储，但您可以选择覆盖此设置，并将密码作为明文存储在代理策略定义中。秘密存储需要舰队服务器版本 8.12 或更高版本。

请注意，此设置也可以存储为秘密值或明文，用于预配置的输出。有关更多信息，请参阅 Kibana 指南中的 预配置设置。

SSL

使用安全套接字层 (SSL) 协议进行身份验证。为您的 SSL 证书提供以下详细信息

客户端 SSL 证书

为客户端生成的证书。复制并粘贴证书的全部内容。这是所有代理将用于连接到 Kafka 的证书。

如果每个客户端都有唯一的证书，则可以在此处放置该证书的本地路径。代理在建立与 Kafka 的连接时，将选择该位置的证书。

客户端 SSL 证书密钥

为客户端生成的私钥。这必须是 PKCS 8 密钥。复制并粘贴证书密钥的全部内容。这是所有代理将用于连接到 Kafka 的证书密钥。

如果每个客户端都有唯一的证书密钥，则可以在此处放置该证书密钥的本地路径。代理在建立与 Kafka 的连接时，将选择该位置的证书密钥。

为了防止未经授权的访问，证书密钥存储为一个秘密值。虽然建议使用秘密存储，但您可以选择覆盖此设置，并将密钥作为明文存储在代理策略定义中。秘密存储需要舰队服务器版本 8.12 或更高版本。

请注意，此设置也可以存储为秘密值或明文，用于预配置的输出。有关更多信息，请参阅 Kibana 指南中的 预配置设置。

服务器 SSL 证书颁发机构

用于连接到 Kafka 的 CA 证书。这是用于为 Kafka 生成证书和密钥的 CA。复制并粘贴 CA 证书的全部内容。

此设置是可选的。当选择身份验证 无 和 明文 选项时，此设置不可用。

单击 添加行 以指定其他证书颁发机构。

验证模式

控制服务器证书的验证。有效值为

完全
验证提供的证书是否由受信任的机构 (CA) 签名，并验证服务器的主机名（或 IP 地址）是否与证书中标识的名称匹配。
无
对服务器的证书不执行任何验证。此模式会禁用 SSL/TLS 的许多安全优势，因此应谨慎考虑后再使用。它主要用作尝试解决 TLS 错误时的临时诊断机制；强烈建议不要在生产环境中使用它。
严格
验证提供的证书是否由受信任的机构 (CA) 签名，并验证服务器的主机名（或 IP 地址）是否与证书中标识的名称匹配。如果使用者可选名称为空，则返回错误。
证书
验证提供的证书是否由受信任的机构 (CA) 签名，但不执行任何主机名验证。

默认值为 完全。当选择身份验证 无 和 明文 选项时，此设置不可用。
分区设置编辑

创建的分区数由 Kafka 代理根据主题列表自动设置。然后，记录将随机、轮询或根据计算的哈希值发布到分区。

随机

将记录随机发布到 Kafka 输出代理事件分区。指定在分区器选择新分区之前要发布到同一分区的事件数。

轮询

以轮询方式将记录发布到 Kafka 输出代理事件分区。指定在分区器选择新分区之前要发布到同一分区的事件数。

哈希

根据从指定字段列表计算出的哈希值，将记录发布到 Kafka 输出代理事件分区。如果未指定字段，则使用 Kafka 事件键值。
主题设置编辑

使用此选项可为每个 Elastic Agent 事件设置 Kafka 主题。

默认主题

设置默认主题，用于 Elastic Agent 发送到 Kafka 输出的事件，例如 elastic-agent。
标头设置编辑

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

键

要在 Kafka 标头中设置的键。

值

要在 Kafka 标头中设置的值。

单击 添加标头 以配置要包含在每个 Kafka 消息中的其他标头。

客户端 ID

用于日志记录、调试和审计目的的可配置 ClientID。默认值为 Elastic。客户端 ID 是协议的一部分，用于标识消息的来源。
压缩设置编辑

您可以启用压缩以减少 Kafka 输出量。

编解码器

选择要使用的压缩编解码器。支持的编解码器有 snappy、lz4 和 gzip。

级别

对于 gzip 编解码器，您可以选择压缩级别。级别必须在 1（速度最快）到 9（压缩率最高）的范围内。

提高压缩级别会减少网络使用量，但会增加 CPU 使用量。默认值为 4。
代理设置编辑

为 Kafka 代理配置超时和缓冲区大小值。

代理超时

Kafka 代理在超时之前等待所需 ACK 数量的最长时间（请参阅后面的 ACK 可靠性 设置）。默认值为 30 秒。

代理可达性超时

Elastic Agent 在超时之前等待 Kafka 代理响应的最长时间。默认值为 30 秒。

ACK 可靠性

代理所需的 ACK 可靠性级别。选项有

  • 等待本地提交
  • 等待所有副本提交
  • 不等待

默认设置为 等待本地提交。

请注意，如果 ACK 可靠性设置为 不等待，Kafka 将不会返回任何 ACK。如果发生错误，消息可能会静默丢失。
其他设置编辑

键

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

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

代理

为 Elastic Agent 选择一个代理 URL 以连接到 Kafka。要了解有关代理配置的信息，请参阅 将代理服务器与 Elastic Agent 和 Fleet 一起使用。

高级 YAML 配置

YAML 设置将被添加到使用此输出的每个策略的 Kafka 输出部分。请确保您指定了有效的 YAML。UI 当前不提供验证。

有关可用设置的说明，请参阅 高级 YAML 配置。

将此输出设为 Agent 集成的默认输出

启用此设置后，如果在 Agent 策略 中未设置其他输出，则 Elastic Agent 将使用此输出发送数据。

将此输出设为 Agent 监控的默认输出

启用此设置后，如果在 Agent 策略 中未设置其他输出，则 Elastic Agent 将使用此输出发送 Agent 监控数据。

高级 YAML 配置编辑

设置 描述

backoff.init

（字符串）网络错误后尝试重新连接到 Kafka 之前的等待秒数。等待 backoff.init 秒后，Elastic Agent 尝试重新连接。如果尝试失败，则退避计时器会呈指数级增长，直至达到 backoff.max。成功连接后，退避计时器将重置。

默认值： 1s

backoff.max

（字符串）网络错误后尝试连接到 Kafka 之前的最长等待秒数。

默认值： 60s

bulk_max_size

（整数）单个 Kafka 请求中批量处理的最大事件数。

默认值： 2048

bulk_flush_frequency

（整数）发送批量 Kafka 请求之前的等待时间。0` 表示无延迟。

默认值： 0

channel_buffer_size

（整数）每个 Kafka 代理在输出管道中缓冲的消息数。

默认值： 256

client_id

（字符串）用于日志记录、调试和审计目的的可配置 ClientID。

默认值： Elastic Agent

codec

输出编解码器配置。您可以指定 json 或 format 编解码器。默认情况下，使用 json 编解码器。

json.pretty：如果将 pretty 设置为 true，则事件将以良好的格式进行格式化。默认为 false。

json.escape_html：如果将 escape_html 设置为 true，则字符串中的 html 符号将被转义。默认为 false。

使用启用了漂亮打印的 json 编解码器将事件写入控制台的配置示例

output.console:
  codec.json:
    pretty: true
    escape_html: false

format.string：用于创建自定义格式消息的可配置格式字符串。

使用 format 编解码器将事件时间戳和消息字段打印到控制台的可配置示例

output.console:
  codec.format:
    string: '%{[@timestamp]} %{[message]}'

默认值： json

keep_alive

（字符串）活动网络连接的保持活动时间。如果为 0s，则禁用保持活动时间。

默认值： 0s

max_message_bytes

（整数）JSON 编码消息允许的最大大小。更大的消息将被丢弃。此值应等于或小于代理的 message.max.bytes。

默认值： 1000000（字节）

metadata

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

refresh_frequency
元数据刷新间隔。默认为 10 分钟。
full
获取元数据时使用的策略。如果此选项为 true，则客户端将维护所有可用主题的完整元数据集。如果设置为 false，则它只会刷新已配置主题的元数据。默认为 false。
retry.max
元数据更新重试总次数。默认为 3。
retry.backoff
重试之间的等待时间。默认为 250 毫秒。

queue.mem.events

队列可以存储的事件数。此值应可以被 queue.mem.flush.min_events 或 bulk_max_size 中较小的值整除，以避免向输出发送部分批次。

默认值： 3200 个事件

queue.mem.flush.min_events

flush.min_events 是一个遗留参数，新配置应优先使用 bulk_max_size 来控制批次大小。从 8.13 版本开始，使用 flush.min_events 而不是 bulk_max_size 来限制批次大小永远不会带来性能优势。

默认值： 1600 个事件

queue.mem.flush.timeout

（整数）满足 queue.mem.flush.min_events 的最大等待时间。如果设置为 0 秒，则事件立即可用于输出。

默认值： 10s
« Logstash 输出设置 远程 Elasticsearch 输出 »