Nightingale/Categraf Kafka 采集插件实战基于 kafka_exporter 的消费延迟Consumer Lag监控配置指南【免费下载链接】nightingaleNightingale is to monitoring and alerting what Grafana is to visualization.项目地址: https://gitcode.com/GitHub_Trending/ni/nightingaleKafka 的核心指标大多通过 JMX 暴露而消费者消费延迟Consumer Lag无法从服务端 JMX 获取需要专门插件从消费者端采集。本文基于 Nightingale 生态的 Categraf 采集器中的 Kafka 插件系统讲解其定位、指标命名差异、延迟速率预估原理以及完整的kafka.toml配置参数并结合仓库内自带告警规则与看板给出可直接落地的消费积压监控方案。插件定位为什么需要它Kafka 的 Broker 侧指标消息吞吐、请求延迟、副本状态等全部通过 JMX 暴露这类指标使用 jolokia 或官方jmx_exporterjar 即可采集本插件并不负责 JMX 指标的采集。本插件的核心职责是采集消费者延迟数据Consumer Lag——即消费组已消费的 offset 与 Topic 最新 offset 之间的差值。该数据由消费者协调与 offset 提交机制产生无法通过 Kafka 服务端的 JMX 拿到因此必须由独立插件通过 Kafka 客户端 API 主动查询各消费组各分区的 offset 状态来获得。插件的来源与版本谱系该插件 fork 自davidmparrott/kafka_exporter下称 davidmparrott 版本而 davidmparrott 版本又 fork 自danielqsj/kafka_exporter下称 danielqsj 版本。danielqsj 版本是原始版本在 Prometheus 生态中被广泛使用。davidmparrott 版本与 danielqsj 版本相比部分指标名称不同对应关系如下davidmparrott 版本本插件默认danielqsj 版本kafka_consumergroup_uncommit_offsetskafka_consumergroup_lagkafka_consumergroup_uncommit_offsets_sumkafka_consumergroup_lag_sumkafka_consumergroup_uncommitted_offsets_zookeeperkafka_consumergroup_lag_zookeeper如果希望沿用 danielqsj 版本的指标命名例如与既有 PromQL 规则或 Grafana 看板兼容在[[instances]]中开启如下配置即可rename_uncommit_offset_to_lag true延迟速率预估指标与积压风险判断davidmparrott 版本相比 danielqsj 版本额外增加了三个对延迟速率做预估计算的指标kafka_consumer_lag_milliskafka_consumer_lag_interpolationkafka_consumer_lag_extrapolation为什么要计算速率因为lag 很大但消费很快并不会产生积压而 lag 很小但消费很慢仍然会积压。仅凭 lag 数值无法判断积压风险通过计算历史消费速率来推断把当前积压消费完还需要多长时间才更合理。由于计算该速率需要将每个分区的历史 offset 存入插值表interpolation table会占用较多内存可通过如下配置关闭这一计算逻辑disable_calculate_lag_rate true关闭后仅保留基础的 lag 指标适用于内存受限或集群分区数量庞大的场景。采集配置快速上手Categraf 的 Kafka 插件配置文件为conf/input.kafka/kafka.toml仓库内对应的实际模板位于 integrations/Kafka/collect/kafka/kafka.toml。最简可用配置如下[[instances]] log_level error kafka_uris [192.168.0.250:9092] labels { clusterkafka-cluster-01, servicekafka }其中kafka_uris为 Kafka 集群的 broker 地址列表host:portlabels为附加到所有指标上的标签cluster是推荐使用的集群名标签——若未显式配置默认取kafka_uris中的第一个地址作为 cluster 值log_level仅作用于 kafka exporter 自身的日志。完整配置参数详解以下为带注释的完整配置与仓库模板一致逐项说明其作用与默认值[[instances]] # # interval global.interval * interval_times # interval_times 1 # append some labels to metrics # cluster is a preferred tag with the cluster name. If none is provided, the first of kafka_uris will be used labels { clusterkafka-cluster-01 } # log level only for kafka exporter log_level error # Address (host:port) of Kafka server. # kafka_uris [127.0.0.1:9092,127.0.0.1:9092,127.0.0.1:9092] kafka_uris [] # Connect using SASL/PLAIN # Default is false # use_sasl false # Only set this to false if using a non-Kafka SASL proxy # Default is true # use_sasl_handshake false # SASL user name # sasl_username username # SASL user password # sasl_password password # The SASL SCRAM SHA algorithm sha256 or sha512 as mechanism # sasl_mechanism # Connect using TLS # use_tls false # The optional certificate authority file for TLS client authentication # ca_file # The optional certificate file for TLS client authentication # cert_file # The optional key file for TLS client authentication # key_file # If true, the servers certificate will not be checked for validity. This will make your HTTPS connections insecure # insecure_skip_verify true # Kafka broker version # Default is 2.0.0 # kafka_version 2.0.0 # if you need to use a group from zookeeper # Default is false # use_zookeeper_lag false # Address array (hosts) of zookeeper server. # zookeeper_uris [] # Metadata refresh interval # Default is 1m # metadata_refresh_interval 1m # Whether show the offset/lag for all consumer group, otherwise, only show connected consumer groups, default is true # Default is true # offset_show_all true # If true, all scrapes will trigger kafka operations otherwise, they will share results. WARN: This should be disabled on large clusters # Default is false # allow_concurrency false # Maximum number of offsets to store in the interpolation table for a partition # Default is 1000 # max_offsets 1000 # How frequently should the interpolation table be pruned, in seconds. # Default is 30 # prune_interval_seconds 30 # Regex filter for topics to be monitored # Default is .* # topics_filter_regex .* # Regex filter for consumer groups to be monitored # Default is .* # groups_filter_regex .* # if rename kafka_consumergroup_uncommitted_offsets to kafka_consumergroup_lag # Default is false # rename_uncommit_offset_to_lag false # if disable calculating lag rate # Default is false # disable_calculate_lag_rate false参数速查表参数默认值说明interval_times1采集间隔倍率实际采集间隔 全局 interval × 该倍率labels空附加到指标上的标签推荐使用cluster标识集群名未配置时取kafka_uris第一个地址log_levelerrorkafka exporter 自身日志级别kafka_uris[]Kafka broker 地址数组host:port必填use_saslfalse是否使用 SASL/PLAIN 认证连接use_sasl_handshaketrue是否使用 SASL 握手仅当连接非 Kafka 的 SASL 代理时才置为falsesasl_username/sasl_password空SASL 用户名 / 密码sasl_mechanism空SASL SCRAM SHA 算法机制取sha256或sha512use_tlsfalse是否启用 TLS 连接ca_file/cert_file/key_file空TLS 客户端认证的 CA、证书、私钥文件insecure_skip_verifytrue是否跳过服务端证书校验跳过会使 HTTPS 连接不安全kafka_version2.0.0Kafka broker 版本use_zookeeper_lagfalse是否从 Zookeeper 获取消费组 offsetzookeeper_uris[]Zookeeper 地址数组metadata_refresh_interval1m元数据刷新间隔offset_show_alltrue是否展示所有消费组的 offset/lagfalse时仅展示有活跃连接的消费组allow_concurrencyfalse是否每次抓取都触发 Kafka 操作为false时多次抓取共享结果。大集群上应保持关闭max_offsets1000每个分区插值表最多缓存的 offset 数量用于速率计算prune_interval_seconds30插值表的清理间隔秒topics_filter_regex.*需要监控的 Topic 正则过滤器groups_filter_regex.*需要监控的消费组正则过滤器rename_uncommit_offset_to_lagfalse是否将kafka_consumergroup_uncommitted_offsets改名为kafka_consumergroup_lag兼容 danielqsj 版本命名disable_calculate_lag_ratefalse是否关闭延迟速率计算关闭可显著降低内存占用其中topics_filter_regex与groups_filter_regex适合集群中 Topic 或消费组数量较多、只想聚焦关键链路的场景offset_show_all false可避免展示大量已下线消费组的陈旧数据。指标清单与告警规则仓库为 Kafka 集成提供了指标元数据定义、告警规则与看板三件套可直接导入 Nightingale 使用指标定义integrations/Kafka/metrics/categraf-base.jsonCategraf 插件告警规则integrations/Kafka/alerts/kafka_by_categraf.jsonExporter 告警规则integrations/Kafka/alerts/kafka_by_exporter.json看板integrations/Kafka/dashboards/kafka_by_categraf.json核心指标插件主要输出以下指标在 metrics/categraf-base.json 中有明确定义指标含义kafka_brokers集群中的 broker 数量kafka_topic_partitions各 Topic 的分区数kafka_topic_partition_replicas各 Topic 分区的副本数kafka_topic_partition_under_replicated_partition各 Topic 中副本未同步的分区数kafka_topic_partition_current_offset各 Topic 分区当前最新 offset生产水位kafka_consumergroup_current_offset各消费组各分区当前已消费 offsetkafka_consumergroup_uncommitted_offsets各消费组的未提交 offset 数即消费延迟 lag默认命名基于上述指标指标定义中还提供了两个常用的吞吐量 PromQL各 Topic 每秒消费消息量sum(irate(kafka_consumergroup_current_offset[3m])) without (partition)各 Topic 每秒生产消息量sum(irate(kafka_topic_partition_current_offset[3m])) without (partition)消费积压告警kafka_by_categraf.json中针对本插件默认指标命名定义了三条告警规则kafka 消费组积压超过 1 万条severity 2max by (topic, consumergroup) (kafka_consumergroup_uncommitted_offsets) 10000评估间隔 15 秒。kafka 分区副本未同步severity 2max by (topic) (kafka_topic_partition_under_replicated_partition) 0评估间隔 15 秒。kafka broker 数量下降severity 1delta(kafka_brokers[10m]) 0评估间隔 60 秒。而 kafka_by_exporter.json 则面向使用 danielqsj 版本命名开启rename_uncommit_offset_to_lag true的场景例如其积压告警写为max by (topic, consumergroup) (kafka_consumergroup_lag) 10000。两套规则可按实际指标命名选择导入避免出现指标不存在导致告警恒不触发的问题。值得注意的是告警规则自带append_tags: [servicekafka, typecategraf]便于在 Nightingale 中按标签区分数据来源。内置看板kafka_by_categraf.json 看板围绕本插件指标构建包含overview行brokersbroker 数、topicstopic 数count(count by (topic) (kafka_topic_partitions{cluster$cluster}))、partitions分区总数、Replicas副本总数throughput行Messages produced per secondsum(rate(kafka_topic_partition_current_offset{cluster$cluster}[1m])) by (topic)、Messages consumed per secondsum(rate(kafka_consumergroup_current_offset{cluster$cluster}[1m])) by (topic)。看板以cluster作为模板变量因此配置插件时务必在labels中设置cluster标签导入看板后即可通过集群筛选查看对应 Kafka 集群的状态。告警后的排查思路仓库在告警规则的annotations与 i18n/en_US.json 中提供了配套的英文 runbook可作为积压告警触发后的标准处置流程参考从告警标签取出consumergroup与topic执行kafka-consumer-groups.sh --describe --group group --bootstrap-server broker查看各分区 LAG 分布积压集中在个别分区说明数据倾斜应检查生产端的分区键设计积压均匀分布则是消费者处理慢——先检查消费端日志是否有异常重试再考虑扩容 consumer 实例实例数不要超过分区数或提升单条消息处理效率确认消费者没有全部掉线MEMBERS为 0 时必须先把消费者拉起来。小结Nightingale 生态中该 Kafka 插件专注于弥补 JMX 采集的空白——消费延迟。通过理解其 fork 谱系带来的指标命名差异、延迟速率预估的计算逻辑以及kafka.toml中每个参数的默认值与适用场景再配合仓库自带的指标定义、告警规则与看板即可快速构建一套既能观测 lag 绝对值、又能评估积压风险的 Kafka 消费监控体系。【免费下载链接】nightingaleNightingale is to monitoring and alerting what Grafana is to visualization.项目地址: https://gitcode.com/GitHub_Trending/ni/nightingale创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考