Confluent Cloud Kafka 可观测性最佳实践

Confluent Cloud 介绍

Confluent Cloud 是一个完全托管的 Apache Kafka 服务，提供高可用性和可扩展性，旨在简化数据流处理和实时数据集成。用户可以轻松创建和管理 Kafka 集群，而无需担心基础设施的维护和管理。Confluent Cloud 支持多种数据源和目标，允许用户在云环境中实现数据流的无缝传输和处理。此外，它还提供强大的工具和功能，如 Schema Registry、Kafka Connect 和 KSQL，使开发者能够快速构建和部署流处理应用程序。通过 Confluent Cloud，企业可以更高效地利用实时数据，推动业务创新和数字化转型。

Confluent Cloud 旨在为开发者和企业提供一种简单、高效的方式来构建、部署和管理实时数据管道和流式应用程序。以下是 Confluent Cloud 的几个关键特点：

完全托管：Confluent Cloud 负责处理 Kafka 集群的维护和管理工作，包括配置、扩展、更新和故障恢复，让用户可以专注于应用程序的开发。
高可用性：提供高可靠性和容错能力，确保数据流和应用程序的稳定运行。
可扩展性：根据需求自动或手动扩展资源，轻松应对流量高峰。
安全性：提供加密、身份验证和授权机制，保护数据安全和隐私。
性能优化：自动优化性能，包括负载均衡和分区管理，以提高数据处理效率。
成本效益：按使用量付费，无需预先投资昂贵的硬件和维护成本。

可观测性对于 Confluent Cloud 尤为重要，它能够提供对数据流和系统性能的深入洞察，实时跟踪其性能指标，确保数据流的稳定性和可靠性，及时发现并解决潜在问题，优化资源分配，并保障业务连续性。通过监控，可以获得关键洞察，比如流量模式、延迟、错误率等，从而帮助维护服务质量，预防系统故障，并支持数据驱动的决策制定。

观测云

观测云是一款专为 IT 工程师打造的全链路可观测产品，它集成了基础设施监控、应用程序性能监控和日志管理，为整个技术栈提供实时可观察性。这款产品能够帮助工程师全面了解端到端的用户体验追踪，了解应用内函数的每一次调用，以及全面监控云时代的基础设施。此外，观测云还具备快速发现系统安全风险的能力，为数字化时代提供安全保障。

部署 DataKit

DataKit 是一个开源的、跨平台的数据收集和监控工具，由观测云开发并维护。它旨在帮助用户收集、处理和分析各种数据源，如日志、指标和事件，以便进行有效的监控和故障排查。DataKit 支持多种数据输入和输出格式，可以轻松集成到现有的监控系统中。

登录观测云控制台，在集成 -> DataKit 选择对应安装方式，当前采用 Linux 主机部署 DataKit 。

指标采集

Confluent Cloud 指标集成配置

1、登陆 Confluent Cloud ，在 ADMINISTRATION 下选择 Metrics 。

2、在 Metrics 页点击按钮 New integration ，选择监控类型 Prometheus 。

3、点击 Generate Cloud API key 按钮，生成 API Key 。

4、Resources 选择 All Kafka clusters，生成 Prometheus 的 scrape_configs 。

scrape_configs:- job_name: Confluent Cloudscrape_interval: 1mscrape_timeout: 1mhonor_timestamps: truestatic_configs:- targets:- api.telemetry.confluent.cloudscheme: httpsbasic_auth:username: H5BO.....password: RDCgMwguHMy.....metrics_path: /v2/metrics/cloud/exportparams:"resource.kafka.id":- lkc-xxxx

5、点击页面 Copy 按钮进行内容复制。

DataKit 采集器配置

由于 Confluent Cloud 能够直接暴露 metrics url ，所以可以直接通过 prom 采集器进行采集。

进入 DataKit 安装目录下的 conf.d/prom ，复制 prom.conf.sample 为 confluent_cloud.conf 。

cp prom.conf.sample confluent_cloud.conf

调整 confluent_cloud.conf 内容如下：

# {"version": "1.61.0-testing_testing-graphite-metric-set", "desc": "do NOT edit this line"}[[inputs.prom]]## Exporter URLs.urls = ["https://api.telemetry.confluent.cloud/v2/metrics/cloud/export?resource.kafka.id=lkc-xxxx"]source = "confluent_cloud"## Add HTTP headers to data pulling (Example basic authentication).[inputs.prom.http_headers]Authorization = "Basic QkXXXXXXXXXXXX"interval = "60s"

调整参数说明：

urls: 将 Confluent Cloud 复制的内容调整下，拼接成 url ，如果有多个 kafka 资源，则用逗号分割。
Authorization: 将用户名和密码转化成 Basic Authorization 格式，可以使用转化工具在线生成Basic Auth、http基础认证密码生成、 basic access authentication在线生成、Basic Auth密码在线解密--查错网。
interval: 调整为 60s ，由于 Confluent Cloud API 限制，这个值不能小于 60s ，低于这个值将导致无法采集数据。

关键指标

confluent 指标集

指标名称	描述	单位
`kafka_server_active_connection_count`	活跃认证连接数	count
`kafka_server_consumer_lag_offsets`	组成员提交的偏移量与分区的高水位标记之间的滞后量。	count
`kafka_server_partition_count`	分区数量	count
`kafka_server_received_bytes`	从网络接收的客户数据字节数的增量计数。每个样本是自上一个数据样本以来接收的字节数。计数每60秒采样一次。	byte
`kafka_server_received_records`	接收的记录数的增量计数。每个样本是自上一个数据样本以来接收的记录数。计数每60秒采样一次。	count
`kafka_server_request_bytes`	指定请求类型通过网络发送的总请求字节数的增量计数。每个样本是自上一个数据点以来发送的字节数。计数每60秒采样一次。	byte
`kafka_server_request_count`	通过网络接收的请求数的增量计数。每个样本是自上一个数据点以来接收的请求数。计数每60秒采样一次。	count
`kafka_server_response_bytes`	指定响应类型通过网络发送的总响应字节数的增量计数。每个样本是自上一个数据点以来发送的字节数。计数每60秒采样一次。	byte
`kafka_server_rest_produce_request_bytes`	Kafka REST产生调用发送的总请求字节数的增量计数。	byte
`kafka_server_retained_bytes`	集群保留的字节数当前计数。计数每60秒采样一次。	byte
`kafka_server_sent_bytes`	通过网络发送的客户数据字节数的增量计数。每个样本是自上一个数据点以来发送的字节数。计数每60秒采样一次。	byte
`kafka_server_sent_records`	发送的记录数的增量计数。每个样本是自上一个数据点以来发送的记录数。计数每60秒采样一次。	count
`kafka_server_successful_authentication_count`	成功认证的增量计数。每个样本是自上一个数据点以来成功认证的数量。计数每60秒采样一次。	count