› › ›

Kafka 输出设置

编辑

Kafka 输出设置

编辑

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

如果您计划在将 Elastic Agent 输出数据发送到 Kafka 之前使用 Logstash 修改数据，请参阅本页后面的指南，了解如何操作。

常规设置

编辑

Kafka 版本

Elastic Agent 连接时将请求的 Kafka 协议版本。默认为 1.0.0。目前支持 0.8.2.0 到 2.6.0 的 Kafka 版本，但是选择 2.6.0 版本后，预计最新 Kafka 版本 (3.x.x) 也兼容。使用 Kafka 4.0 及更高版本时，版本必须至少设置为 2.1.0。

主机

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

示例

localhost:9092
mykafkahost:9092

有关默认端口和其他配置详细信息，请参阅 Fleet Server 文档。

身份验证设置

编辑

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

无	Elastic Agent 和 Kafka 之间不使用身份验证。这是默认选项。在生产环境中，建议选择身份验证方法。明文将此选项设置为将 Elastic Agent 和 Kafka 之间的流量作为明文发送，不使用任何传输层安全性。这是未设置身份验证时的默认选项。加密将此选项设置为使用传输层安全性对 Elastic Agent 和 Kafka 之间的流量进行加密。选择加密后，服务器 SSL 证书颁发机构和验证模式选项将可用。
用户名/密码	使用用户名和密码连接到 Kafka。提供您的用户名和密码，并为您的登录凭据选择 SASL（简单身份验证和安全层）机制。启用 SCRAM 时，Elastic Agent 使用 SCRAM 机制来验证用户凭据。SCRAM 基于 IETF RFC5802 标准，该标准描述了一种用于验证用户的质询-响应机制。明文 - 不使用 SCRAM 进行身份验证 SCRAM-SHA-256 - 使用 SHA-256 哈希函数 SCRAM-SHA-512 - 使用 SHA-512 哈希函数为了防止未经授权的访问，您的 Kafka 密码将作为机密值存储。虽然建议使用机密存储，但您可以选择覆盖此设置，并将密码作为明文存储在代理策略定义中。机密存储需要 Fleet Server 8.12 或更高版本。请注意，此设置也可以作为机密值或明文存储在预配置的输出中。请参阅 Kibana 指南中的预配置设置，了解更多信息。
SSL	使用安全套接字层 (SSL) 协议进行身份验证。提供以下 SSL 证书详细信息客户端 SSL 证书为客户端生成的证书。复制并粘贴证书的完整内容。这是所有代理将用于连接到 Kafka 的证书。如果每个客户端都有唯一的证书，则可以将该证书的本地路径放在此处。代理将在建立与 Kafka 的连接时选择该位置的证书。客户端 SSL 证书密钥为客户端生成的私钥。这必须是 PKCS 8 密钥。复制并粘贴证书密钥的完整内容。这是所有代理将用于连接到 Kafka 的证书密钥。如果每个客户端都有唯一的证书密钥，则可以将该证书密钥的本地路径放在此处。代理将在建立与 Kafka 的连接时选择该位置的证书密钥。为了防止未经授权的访问，证书密钥将作为机密值存储。虽然建议使用机密存储，但您可以选择覆盖此设置，并将密钥作为明文存储在代理策略定义中。机密存储需要 Fleet Server 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，或者您可以选择根据 Elastic 通用模式 (ECS) 字段动态设置主题。可用字段包括

data_stream_type
data_stream.dataset
data_stream.namespace
@timestamp
event-dataset

您还可以设置自定义字段。如果您正在将 add_fields 处理器用作 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 用于连接到 Kafka 的代理 URL。要了解有关代理配置的信息，请参阅使用 Elastic Agent 和 Fleet 的代理服务器。
高级 YAML 配置	将添加到使用此输出的每个策略的 Kafka 输出部分的 YAML 设置。请确保您指定有效的 YAML。UI 目前不提供验证。请参阅高级 YAML 配置以了解可用设置的说明。
将此输出设为代理集成的默认输出	启用此设置后，如果在代理策略中未设置其他输出，则 Elastic Agent 将使用此输出发送数据。
将此输出设为代理监控的默认输出	启用此设置后，如果在代理策略中未设置其他输出，则 Elastic 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` 完成的最大等待时间。如果设置为 0s，则事件将立即提供给输出。默认值： `10s`

Kafka 输出和使用 Logstash 将数据索引到 Elasticsearch

编辑

如果您正在考虑使用 Logstash 将数据从 kafka 发送到 Elasticsearch，请注意，Elastic Agent 发送到 kafka 的文档结构不得被 Logstash 修改。我们建议在 kafka 输入和 json 编解码器中禁用 ecs_compatibility，以确保输入不会编辑字段及其内容。

集成设置的数据流预计将接收与直接从 Elastic Agent 发送的事件具有相同结构和字段名称的事件。

Elastic Agent 发送到 kafka 的文档结构不得被 Logstash 修改。我们建议在 kafka 输入和 json 编解码器中禁用 ecs_compatibility。

有关更多详细信息，请参阅 Elastic Agent 的 Logstash 输出文档。

inputs {
  kafka {
    ...
    ecs_compatibility => "disabled"
    codec => json { ecs_compatibility => "disabled" }
    ...
  }
}
...

« Logstash 输出设置远程 Elasticsearch 输出 »