用 librdkafka Stats 工具集解析与可视化 Kafka 客户端统计 JSONto_csv、graph 与 jq 实战指南【免费下载链接】fluent-bitFast and Lightweight Logs, Metrics and Traces processor for Linux, BSD, OSX and Windows项目地址: https://gitcode.com/GitHub_Trending/fl/fluent-bit导读librdkafka 是 Fluent Bit 等众多日志与指标处理组件所依赖的 C/C Kafka 客户端库其内置的statistics.interval.ms配置与stats_cb回调可以按固定周期输出一份内容详尽的 JSON 统计快照。本文聚焦当前仓库中随 librdkafka 2.15.0 一同发布的 Stats 工具集系统讲解如何把这份 JSON 流转换为适合分析的时间序列 CSV、如何用交互式图表快速定位延迟与队列积压问题以及如何用jq做即席过滤。读完本文你将掌握一套从原始统计 JSON到可视化报表的完整工具链可直接复用到你自己的 librdkafka 客户端诊断场景中。一、工具集概览与数据来源1.1 工具集构成该工具集位于 lib/librdkafka-2.15.0/tests/tools/stats/由四个文件组成职责边界非常清晰文件作用输入 → 输出to_csv.py选择性把 stats JSON 转成 CSVstdin 逐行 JSON →{前缀}_top.csv、{前缀}_brokers.csv、{前缀}_topics.csv、{前缀}_toppars.csvgraph.py基于 pandas Bokeh 绘制 CSV 图表CSV 文件 → 交互式 HTML 图表filter.jq基础的jq过滤脚本原始 stats JSON → 精简后的 JSONrequirements.txtPython 依赖清单pandas、pandas-bokeh、numpy三个脚本的分工构成一条清晰的数据流水线原始 JSON → CSV → 图表其中filter.jq则是面向原始 JSON 的轻量替代方案。1.2 统计 JSON 从何而来README 明确指出这些工具适合解析 librdkafka 通过stats_cb发出的统计信息前提是设置了statistics.interval.ms。更精确的机制说明可以参考两个权威来源头文件 src/rdkafka.h 中rd_kafka_conf_set_stats_cb()的注释统计回调与statistics.interval.ms配合使用应用若在某次回调中返回 0则后续不再触发该回调即统计输出可以被应用主动关闭。若想了解完整 JSON 结构头文件明确指引读者查阅STATISTICS.md。规格文档 STATISTICS.md 对 JSON 对象做了逐字段定义顶层包含name、client_id、typeproducer/consumer、time自 epoch 起的秒数、msg_cnt/msg_size生产者队列中的消息数与字节数等随后是brokers、topics两个大字典以及可选的cgrp消费组与eos幂等生产者对象。除字节与时间单位有特殊说明外所有尺寸字段默认以字节为单位。注意STATISTICS.md 特别提醒由于 librdkafka 内部是异步工作的brokers、toppars与顶层总量之间可能并非严格一致例如顶层tx总量可能小于它所代表的各 brokertx之和。做数据分析时要避免对这类微小的口径差异过度敏感。JSON 的整体骨架如下摘自 STATISTICS.md{ Top-level fields brokers: { brokers fields, toppars: { toppars fields } }, topics: { topic fields, partitions: { partitions fields } } [, cgrp: { cgrp fields } ] [, eos: { eos fields } ] }二、安装依赖工具依赖三个 Python 包README 给出的安装命令为$ python3 -m pip install -r requirements.txt依赖清单requirements.txt包含pandasto_csv.py与graph.py的数据处理基础pandas-bokeh把 pandas DataFrame 直接渲染为 Bokeh 交互式图表的桥接库numpypandas 的底层数值计算依赖。建议在独立虚拟环境中安装避免污染系统 Python 环境。三、to_csv.py把 JSON 流转换为时间序列 CSV3.1 基本用法README 给出的典型调用方式是从日志文件中提取以STATS:开头的行剥离前缀后逐行喂给脚本$ grep -F STATS: file.log | sed -e s/^.*STATS: // | ./to_csv.py test1这条命令等价于应用把每次stats_cb收到的 JSON 以STATS: json的格式写进日志之后用grep -F精确匹配-F表示按固定字符串匹配不解释正则再用sed删除行首到STATS:为止的所有内容得到纯 JSON 行。test1是输出文件前缀脚本会在当前目录生成四个 CSV 文件README 提到这些test*.csv文件会被创建输出文件内容test1_top.csv客户端实例级别的总体指标test1_brokers.csv每个 broker 的连接与请求指标仅保留有 Produce 请求的 brokertest1_topics.csv每个 topic 的批量大小/条数百分位test1_toppars.csv每个 topic 分区toppar的生产队列与传输指标仅保留有实际传输的 toppar3.2 解析逻辑与字段映射源码级阅读 to_csv.py 的parse()函数可以看到它如何把庞大的 JSON 压缩成便于绘图的时间序列。核心设计思想是每个 CSV 的第一列都是0time以%Y-%m-%d %H:%M:%S格式化的 UTC 时间保证多文件之间可以按时间对齐同时用1nodeid、1topic、1partition这类前缀列做分组键配合graph.py的--group-by使用。top 级实例总体取msg_cnt队列中消息数与msg_size队列中消息总字节数并额外计算两个填充率百分比top[msg_cnt_fill] (float(js[msg_cnt]) / js[msg_max]) * 100.0 top[msg_size_fill] (float(js[msg_size]) / js[msg_size_max]) * 100.0即队列占用与阈值msg_max、msg_size_maxSTATISTICS.md 中定义的队列上限之比直观反映生产者队列是否接近饱和。brokers 级跳过req[Produce] 0的 broker即未参与生产的 broker 不画保留的字段包括stateage状态持续时间除以 1000 换算为秒请求队列指标outbuf_cnt等待发送的请求数、outbuf_msg_cnt等待发送的消息数、waitresp_cnt在途等待响应的请求数、waitresp_msg_cnt在途消息数、tx已发送请求总数、wakeupsbroker 线程轮询唤醒次数延迟类百分位rtt往返时间、int_latency内部生产者队列延迟、outbuf_latency请求从入队到写入 socket 的排队延迟均取p99并除以 1000 换算为毫秒throttlebroker 节流时间毫秒取p99与cnt两个合成指标latency_p99 int_latency_p99 outbuf_latency_p99 rtt_p99端到端延迟估算、toppars_cnt len(d[toppars])该 broker 负责的分区数。其中rtt、int_latency、outbuf_latency、throttle都是 STATISTICS.md 中定义的滚动窗口统计对象字段包括min、max、avg、sum、cnt、stddev、p50…p99_99、outofrange本工具统一抽取p99作为代表值。topics 级对每个 topic 取batchsize批量字节数与batchcnt批量消息条数的p99。toppars 级跳过txmsgs 0从未传输消息的分区保留leader当前 leader broker id、msgq_cnt/msgq_bytes一级队列待生产消息数与字节数、xmit_msgq_cnt/xmit_msgq_bytes传输队列就绪消息数与字节数、txmsgs/txbytes累计已传输消息数与字节数、msgs_inflight在途消息数。这些字段与 STATISTICS.md 中 partitions 一节的定义一一对应。3.3 输出行为CsvWriter类负责落盘列按字典序排序OrderedDict(sorted(d.items()))首行写表头之后每行一条记录脚本对 stdin 逐行解析某一行解析失败会打印SKIP 行号: 原因并继续处理后续行不会中断整批数据。这种宽容设计非常适合处理混有杂质的日志输出。四、graph.py把 CSV 渲染成交互式图表4.1 基本用法README 给出的示例是绘制 toppar 图表并按分区分组同时跳过部分列$ ./graph.py --skip *bytes,*msg_cnt,stateage,*msgs,leader --group-by 1partition test1_toppars.csv执行后默认生成out.html用浏览器打开即可交互浏览。4.2 命令行参数详解结合 graph.py 的参数定义参数默认值说明infiles位置参数必填一个或多个 CSV 文件支持同时传入多个文件一起绘图--cols无逗号分隔的列名列表只绘制这些列时间列0time会被自动追加--skip无逗号分隔的列名glob列表跳过匹配的列与--cols互斥脚本会断言两者不能同时指定--group-by无按指定字段如1partition、1nodeid、1topic分组每组一个独立数据序列--chart-cols3图表网格的列数决定多个子图如何排布--plot-width400每个子图的宽度像素--plot-height300每个子图的高度像素--outout.html输出 HTML 文件名4.3 内部行为源码级--skip的匹配基于 Pythonfnmatch的 glob 语义graph.py因此示例中的*bytes,*msg_cnt,stateage,*msgs,leader会跳过所有以bytes结尾、包含msg_cnt、等于stateage、以msgs开头以及等于leader的列从而聚焦到与请求数、唤醒数等相关的指标上。数据读取pd.read_csv(..., parse_dates[0time], index_col0time)即以0time作为时间索引x 轴类型为 datetime刻度格式化为%H:%M:%S。交互能力每个子图注册了hover, box_zoom, wheel_zoom, reset, pan, poly_select, tap, save工具graph.py支持悬停查看数值、框选/滚轮缩放、平移、点选与保存HoverTool 显示时间与纵轴值。图例支持点击隐藏/显示对应曲线legend.click_policy hide。--group-by模式按分组字段对 DataFrame 分组后为每个分组在每个指标列上画一条线图例标注为列名[分组值]例如txmsgs[0]分组模式会跳过0time与分组列本身。图表采用dark_minimal主题多个子图通过pandas_bokeh.plot_grid()按--chart-cols排列成网格一次输出到一个 HTML 文件中。五、filter.jq面向原始 JSON 的即席过滤5.1 用法filter.jq 是面向原始统计 JSON 的轻量处理脚本用法为$ cat stats.json | jq -R -f filter.jq-Rraw input告诉jq把每一行当作原始字符串而非 JSON 输入脚本内部用fromjson?把每行安全地解析成对象解析失败返回空不会中断。5.2 脚本逻辑脚本输出一个精简后的对象包含三个部分time把原始时间戳换算为本地化时间字符串脚本示例中减去3600*5秒即 UTC-5 时区偏移可按需调整brokers过滤出req.Produce 0的 broker以nodeid为键保留state、stateage除以 1000000 换算为秒、connects、rtt.p99、throttle.cnt、outbuf_cnt、outbuf_msg_cnt、waitresp_cnt、Produce/Metadata请求计数以及toppar_cnttopics过滤出batchcnt.cnt 0的 topic以topic名为键保留batchsize.p99、batchcnt.p99并展开每个分区的leader、msgq_cnt、xmit_msgq_cnt、txmsgs、msgs_inflight。它适合在不需要完整 CSV 流水线、只想快速抽查某个时刻的 broker 连接状态或 topic 积压情况时使用。六、完整实战流程6.1 端到端示例假设应用日志中每 5 秒写入一行STATS: {...}完整诊断流程如下# 1) 安装依赖 $ python3 -m pip install -r requirements.txt # 2) 从日志提取统计行并转换为 CSV生成 test1_{top,brokers,topics,toppars}.csv $ grep -F STATS: file.log | sed -e s/^.*STATS: // | ./to_csv.py test1 # 3) 绘制 broker 级延迟RTT、内部队列延迟、节流 $ ./graph.py --group-by 1nodeid test1_brokers.csv # 4) 绘制各分区消息队列积压跳过字节类与累计计数类列 $ ./graph.py --skip *bytes,*msg_cnt,stateage,*msgs,leader \ --group-by 1partition test1_toppars.csv # 5) 或用 jq 快速抽查某时刻的 broker 状态 $ cat stats.json | jq -R -f filter.jq6.2 常见诊断思路生产延迟异常观察test1_brokers.csv中的rtt_p99、int_latency_p99、outbuf_latency_p99与合成列latency_p99若outbuf_latency_p99持续走高说明请求在发送队列中排队通常对应 socket 写阻塞或网络拥塞若int_latency_p99走高说明消息在生产者内部队列滞留可能由queue.buffering.max.messages触顶或linger.ms配置导致。队列积压与节流观察test1_toppars.csv的msgq_cnt、xmit_msgq_cnt、msgs_inflight以及test1_top.csv的msg_cnt_fill/msg_size_fill是否逼近 100%broker 侧throttle_cnt/throttle_p99升高则意味着被 Kafka 侧限流quota。连接稳定性filter.jq输出的state、connects、toppar_cnt可用于快速判断 broker 是否反复断连、分区是否均匀分布。七、与当前仓库的关联Fluent Bit 场景说明本工具集位于仓库内 vendored 的 librdkafka-2.15.0 源码树的测试工具目录tests/tools/stats/与 Fluent Bit 的关系如下Fluent Bit 的 Kafka 输出插件 plugins/out_kafka 编译链接的正是这份内置的 librdkafka 库因此本文介绍的统计字段rtt、outbuf_cnt、waitresp_cnt、msgq_cnt等同样描述着 Fluent Bit 生产数据到 Kafka 时的底层行为。统计 JSON 的产生由使用 librdkafka 的应用层通过statistics.interval.ms与stats_cb决定见 src/rdkafka.h 与 STATISTICS.md。如果你在自己基于 librdkafka 的 C/C 应用中开启了该回调并把 JSON 打印为带STATS:前缀的日志行即可直接套用本文第 36 节的工具链进行离线分析。若希望深入理解某个指标的确切含义例如consumer_lag在read_uncommitted与read_committed下的差异、eos幂等生产者状态机可随时查阅 STATISTICS.md 中cgrp、eos等章节的逐字段定义。结语librdkafka 自带的这套 stats 工具虽然轻量却覆盖了采集 → 清洗 → 可视化 → 抽查的完整诊断链路to_csv.py负责把高频 JSON 压缩为四类时间序列 CSVgraph.py提供带缩放、悬停与分组能力的交互式 HTML 图表filter.jq则为临时抽查保留了最轻量的入口。在 Kafka 客户端出现延迟升高、队列积压或连接抖动时这套工具能帮你把模糊的变慢了快速定位到具体的 broker、topic 与分区维度上。【免费下载链接】fluent-bitFast and Lightweight Logs, Metrics and Traces processor for Linux, BSD, OSX and Windows项目地址: https://gitcode.com/GitHub_Trending/fl/fluent-bit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 用 librdkafka Stats 工具集解析与可视化 Kafka 客户端统计 JSON：to_csv、graph 与 jq 实战指南
用 librdkafka Stats 工具集解析与可视化 Kafka 客户端统计 JSONto_csv、graph 与 jq 实战指南【免费下载链接】fluent-bitFast and Lightweight Logs, Metrics and Traces processor for Linux, BSD, OSX and Windows项目地址: https://gitcode.com/GitHub_Trending/fl/fluent-bit导读librdkafka 是 Fluent Bit 等众多日志与指标处理组件所依赖的 C/C Kafka 客户端库其内置的statistics.interval.ms配置与stats_cb回调可以按固定周期输出一份内容详尽的 JSON 统计快照。本文聚焦当前仓库中随 librdkafka 2.15.0 一同发布的 Stats 工具集系统讲解如何把这份 JSON 流转换为适合分析的时间序列 CSV、如何用交互式图表快速定位延迟与队列积压问题以及如何用jq做即席过滤。读完本文你将掌握一套从原始统计 JSON到可视化报表的完整工具链可直接复用到你自己的 librdkafka 客户端诊断场景中。一、工具集概览与数据来源1.1 工具集构成该工具集位于 lib/librdkafka-2.15.0/tests/tools/stats/由四个文件组成职责边界非常清晰文件作用输入 → 输出to_csv.py选择性把 stats JSON 转成 CSVstdin 逐行 JSON →{前缀}_top.csv、{前缀}_brokers.csv、{前缀}_topics.csv、{前缀}_toppars.csvgraph.py基于 pandas Bokeh 绘制 CSV 图表CSV 文件 → 交互式 HTML 图表filter.jq基础的jq过滤脚本原始 stats JSON → 精简后的 JSONrequirements.txtPython 依赖清单pandas、pandas-bokeh、numpy三个脚本的分工构成一条清晰的数据流水线原始 JSON → CSV → 图表其中filter.jq则是面向原始 JSON 的轻量替代方案。1.2 统计 JSON 从何而来README 明确指出这些工具适合解析 librdkafka 通过stats_cb发出的统计信息前提是设置了statistics.interval.ms。更精确的机制说明可以参考两个权威来源头文件 src/rdkafka.h 中rd_kafka_conf_set_stats_cb()的注释统计回调与statistics.interval.ms配合使用应用若在某次回调中返回 0则后续不再触发该回调即统计输出可以被应用主动关闭。若想了解完整 JSON 结构头文件明确指引读者查阅STATISTICS.md。规格文档 STATISTICS.md 对 JSON 对象做了逐字段定义顶层包含name、client_id、typeproducer/consumer、time自 epoch 起的秒数、msg_cnt/msg_size生产者队列中的消息数与字节数等随后是brokers、topics两个大字典以及可选的cgrp消费组与eos幂等生产者对象。除字节与时间单位有特殊说明外所有尺寸字段默认以字节为单位。注意STATISTICS.md 特别提醒由于 librdkafka 内部是异步工作的brokers、toppars与顶层总量之间可能并非严格一致例如顶层tx总量可能小于它所代表的各 brokertx之和。做数据分析时要避免对这类微小的口径差异过度敏感。JSON 的整体骨架如下摘自 STATISTICS.md{ Top-level fields brokers: { brokers fields, toppars: { toppars fields } }, topics: { topic fields, partitions: { partitions fields } } [, cgrp: { cgrp fields } ] [, eos: { eos fields } ] }二、安装依赖工具依赖三个 Python 包README 给出的安装命令为$ python3 -m pip install -r requirements.txt依赖清单requirements.txt包含pandasto_csv.py与graph.py的数据处理基础pandas-bokeh把 pandas DataFrame 直接渲染为 Bokeh 交互式图表的桥接库numpypandas 的底层数值计算依赖。建议在独立虚拟环境中安装避免污染系统 Python 环境。三、to_csv.py把 JSON 流转换为时间序列 CSV3.1 基本用法README 给出的典型调用方式是从日志文件中提取以STATS:开头的行剥离前缀后逐行喂给脚本$ grep -F STATS: file.log | sed -e s/^.*STATS: // | ./to_csv.py test1这条命令等价于应用把每次stats_cb收到的 JSON 以STATS: json的格式写进日志之后用grep -F精确匹配-F表示按固定字符串匹配不解释正则再用sed删除行首到STATS:为止的所有内容得到纯 JSON 行。test1是输出文件前缀脚本会在当前目录生成四个 CSV 文件README 提到这些test*.csv文件会被创建输出文件内容test1_top.csv客户端实例级别的总体指标test1_brokers.csv每个 broker 的连接与请求指标仅保留有 Produce 请求的 brokertest1_topics.csv每个 topic 的批量大小/条数百分位test1_toppars.csv每个 topic 分区toppar的生产队列与传输指标仅保留有实际传输的 toppar3.2 解析逻辑与字段映射源码级阅读 to_csv.py 的parse()函数可以看到它如何把庞大的 JSON 压缩成便于绘图的时间序列。核心设计思想是每个 CSV 的第一列都是0time以%Y-%m-%d %H:%M:%S格式化的 UTC 时间保证多文件之间可以按时间对齐同时用1nodeid、1topic、1partition这类前缀列做分组键配合graph.py的--group-by使用。top 级实例总体取msg_cnt队列中消息数与msg_size队列中消息总字节数并额外计算两个填充率百分比top[msg_cnt_fill] (float(js[msg_cnt]) / js[msg_max]) * 100.0 top[msg_size_fill] (float(js[msg_size]) / js[msg_size_max]) * 100.0即队列占用与阈值msg_max、msg_size_maxSTATISTICS.md 中定义的队列上限之比直观反映生产者队列是否接近饱和。brokers 级跳过req[Produce] 0的 broker即未参与生产的 broker 不画保留的字段包括stateage状态持续时间除以 1000 换算为秒请求队列指标outbuf_cnt等待发送的请求数、outbuf_msg_cnt等待发送的消息数、waitresp_cnt在途等待响应的请求数、waitresp_msg_cnt在途消息数、tx已发送请求总数、wakeupsbroker 线程轮询唤醒次数延迟类百分位rtt往返时间、int_latency内部生产者队列延迟、outbuf_latency请求从入队到写入 socket 的排队延迟均取p99并除以 1000 换算为毫秒throttlebroker 节流时间毫秒取p99与cnt两个合成指标latency_p99 int_latency_p99 outbuf_latency_p99 rtt_p99端到端延迟估算、toppars_cnt len(d[toppars])该 broker 负责的分区数。其中rtt、int_latency、outbuf_latency、throttle都是 STATISTICS.md 中定义的滚动窗口统计对象字段包括min、max、avg、sum、cnt、stddev、p50…p99_99、outofrange本工具统一抽取p99作为代表值。topics 级对每个 topic 取batchsize批量字节数与batchcnt批量消息条数的p99。toppars 级跳过txmsgs 0从未传输消息的分区保留leader当前 leader broker id、msgq_cnt/msgq_bytes一级队列待生产消息数与字节数、xmit_msgq_cnt/xmit_msgq_bytes传输队列就绪消息数与字节数、txmsgs/txbytes累计已传输消息数与字节数、msgs_inflight在途消息数。这些字段与 STATISTICS.md 中 partitions 一节的定义一一对应。3.3 输出行为CsvWriter类负责落盘列按字典序排序OrderedDict(sorted(d.items()))首行写表头之后每行一条记录脚本对 stdin 逐行解析某一行解析失败会打印SKIP 行号: 原因并继续处理后续行不会中断整批数据。这种宽容设计非常适合处理混有杂质的日志输出。四、graph.py把 CSV 渲染成交互式图表4.1 基本用法README 给出的示例是绘制 toppar 图表并按分区分组同时跳过部分列$ ./graph.py --skip *bytes,*msg_cnt,stateage,*msgs,leader --group-by 1partition test1_toppars.csv执行后默认生成out.html用浏览器打开即可交互浏览。4.2 命令行参数详解结合 graph.py 的参数定义参数默认值说明infiles位置参数必填一个或多个 CSV 文件支持同时传入多个文件一起绘图--cols无逗号分隔的列名列表只绘制这些列时间列0time会被自动追加--skip无逗号分隔的列名glob列表跳过匹配的列与--cols互斥脚本会断言两者不能同时指定--group-by无按指定字段如1partition、1nodeid、1topic分组每组一个独立数据序列--chart-cols3图表网格的列数决定多个子图如何排布--plot-width400每个子图的宽度像素--plot-height300每个子图的高度像素--outout.html输出 HTML 文件名4.3 内部行为源码级--skip的匹配基于 Pythonfnmatch的 glob 语义graph.py因此示例中的*bytes,*msg_cnt,stateage,*msgs,leader会跳过所有以bytes结尾、包含msg_cnt、等于stateage、以msgs开头以及等于leader的列从而聚焦到与请求数、唤醒数等相关的指标上。数据读取pd.read_csv(..., parse_dates[0time], index_col0time)即以0time作为时间索引x 轴类型为 datetime刻度格式化为%H:%M:%S。交互能力每个子图注册了hover, box_zoom, wheel_zoom, reset, pan, poly_select, tap, save工具graph.py支持悬停查看数值、框选/滚轮缩放、平移、点选与保存HoverTool 显示时间与纵轴值。图例支持点击隐藏/显示对应曲线legend.click_policy hide。--group-by模式按分组字段对 DataFrame 分组后为每个分组在每个指标列上画一条线图例标注为列名[分组值]例如txmsgs[0]分组模式会跳过0time与分组列本身。图表采用dark_minimal主题多个子图通过pandas_bokeh.plot_grid()按--chart-cols排列成网格一次输出到一个 HTML 文件中。五、filter.jq面向原始 JSON 的即席过滤5.1 用法filter.jq 是面向原始统计 JSON 的轻量处理脚本用法为$ cat stats.json | jq -R -f filter.jq-Rraw input告诉jq把每一行当作原始字符串而非 JSON 输入脚本内部用fromjson?把每行安全地解析成对象解析失败返回空不会中断。5.2 脚本逻辑脚本输出一个精简后的对象包含三个部分time把原始时间戳换算为本地化时间字符串脚本示例中减去3600*5秒即 UTC-5 时区偏移可按需调整brokers过滤出req.Produce 0的 broker以nodeid为键保留state、stateage除以 1000000 换算为秒、connects、rtt.p99、throttle.cnt、outbuf_cnt、outbuf_msg_cnt、waitresp_cnt、Produce/Metadata请求计数以及toppar_cnttopics过滤出batchcnt.cnt 0的 topic以topic名为键保留batchsize.p99、batchcnt.p99并展开每个分区的leader、msgq_cnt、xmit_msgq_cnt、txmsgs、msgs_inflight。它适合在不需要完整 CSV 流水线、只想快速抽查某个时刻的 broker 连接状态或 topic 积压情况时使用。六、完整实战流程6.1 端到端示例假设应用日志中每 5 秒写入一行STATS: {...}完整诊断流程如下# 1) 安装依赖 $ python3 -m pip install -r requirements.txt # 2) 从日志提取统计行并转换为 CSV生成 test1_{top,brokers,topics,toppars}.csv $ grep -F STATS: file.log | sed -e s/^.*STATS: // | ./to_csv.py test1 # 3) 绘制 broker 级延迟RTT、内部队列延迟、节流 $ ./graph.py --group-by 1nodeid test1_brokers.csv # 4) 绘制各分区消息队列积压跳过字节类与累计计数类列 $ ./graph.py --skip *bytes,*msg_cnt,stateage,*msgs,leader \ --group-by 1partition test1_toppars.csv # 5) 或用 jq 快速抽查某时刻的 broker 状态 $ cat stats.json | jq -R -f filter.jq6.2 常见诊断思路生产延迟异常观察test1_brokers.csv中的rtt_p99、int_latency_p99、outbuf_latency_p99与合成列latency_p99若outbuf_latency_p99持续走高说明请求在发送队列中排队通常对应 socket 写阻塞或网络拥塞若int_latency_p99走高说明消息在生产者内部队列滞留可能由queue.buffering.max.messages触顶或linger.ms配置导致。队列积压与节流观察test1_toppars.csv的msgq_cnt、xmit_msgq_cnt、msgs_inflight以及test1_top.csv的msg_cnt_fill/msg_size_fill是否逼近 100%broker 侧throttle_cnt/throttle_p99升高则意味着被 Kafka 侧限流quota。连接稳定性filter.jq输出的state、connects、toppar_cnt可用于快速判断 broker 是否反复断连、分区是否均匀分布。七、与当前仓库的关联Fluent Bit 场景说明本工具集位于仓库内 vendored 的 librdkafka-2.15.0 源码树的测试工具目录tests/tools/stats/与 Fluent Bit 的关系如下Fluent Bit 的 Kafka 输出插件 plugins/out_kafka 编译链接的正是这份内置的 librdkafka 库因此本文介绍的统计字段rtt、outbuf_cnt、waitresp_cnt、msgq_cnt等同样描述着 Fluent Bit 生产数据到 Kafka 时的底层行为。统计 JSON 的产生由使用 librdkafka 的应用层通过statistics.interval.ms与stats_cb决定见 src/rdkafka.h 与 STATISTICS.md。如果你在自己基于 librdkafka 的 C/C 应用中开启了该回调并把 JSON 打印为带STATS:前缀的日志行即可直接套用本文第 36 节的工具链进行离线分析。若希望深入理解某个指标的确切含义例如consumer_lag在read_uncommitted与read_committed下的差异、eos幂等生产者状态机可随时查阅 STATISTICS.md 中cgrp、eos等章节的逐字段定义。结语librdkafka 自带的这套 stats 工具虽然轻量却覆盖了采集 → 清洗 → 可视化 → 抽查的完整诊断链路to_csv.py负责把高频 JSON 压缩为四类时间序列 CSVgraph.py提供带缩放、悬停与分组能力的交互式 HTML 图表filter.jq则为临时抽查保留了最轻量的入口。在 Kafka 客户端出现延迟升高、队列积压或连接抖动时这套工具能帮你把模糊的变慢了快速定位到具体的 broker、topic 与分区维度上。【免费下载链接】fluent-bitFast and Lightweight Logs, Metrics and Traces processor for Linux, BSD, OSX and Windows项目地址: https://gitcode.com/GitHub_Trending/fl/fluent-bit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考