Telegraf Disque Input Plugin 接入指南从配置到指标采集源码解析【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf本指南围绕 Telegraf 内置的 Disque 输入插件展开讲解如何通过[[inputs.disque]]采集 Disque实验性分布式内存消息代理实例的运行时状态指标包括连接配置、URI 写法、指标字段含义并结合仓库源码剖析其底层采集原理与测试验证方式。阅读本文后你将能够独立完成 Disque 监控指标的接入、字段定制与故障排查。插件概览Disque 是 Redis 作者 antirez 开发的一款实验性、分布式、纯内存消息代理常用于消息队列场景。Telegraf 的 Disque 输入插件负责连接一个或多个 Disque 实例执行INFO命令获取服务端统计信息并将其转换为 Telegraf 指标输出。引入版本Telegraf v0.10.0插件类型inputmessaging 类平台支持所有平台 all插件注册入口位于 plugins/inputs/all/disque.go通过构建标签inputs.disque控制是否启用默认随完整构建一并编译注册//go:build !custom || inputs || inputs.disque import _ github.com/influxdata/telegraf/plugins/inputs/disque // register plugin配置方法基础配置在 Telegraf 配置文件中加入如下片段即可启用采集# Read metrics from one or many disque servers [[inputs.disque]] ## An array of URI to gather stats about. Specify an ip or hostname ## with optional port and password. ## ie disque://localhost, disque://10.10.3.33:18832, 10.0.0.1:10000, etc. ## If no servers are specified, then localhost is used as the host. servers [localhost]配置模板与 README 完全一致原始文件见 plugins/inputs/disque/sample.conf并通过//go:embed嵌入到插件二进制中见 disque.go执行telegraf --usage disque或生成配置时可直接查看。servers 参数详解servers是一个 URI 字符串数组每个元素表示一个要采集的 Disque 节点支持以下几种写法写法示例说明localhost仅主机名自动使用默认端口10.0.0.1:10000主机 端口无 scheme 的裸地址写法disque://localhost带 scheme 的标准 URIdisque://10.10.3.33:18832带 scheme 与自定义端口disque://user:passwordhost:port在 URI 中携带用户名与密码用于 AUTH 认证关键默认值见 disque.go默认端口7711Disque 服务默认监听端口定义于defaultPort常量默认主机若servers数组为空或未配置则回退到localhost并使用默认端口连接超时defaultTimeout 5 * time.Seconddisque.go。与全局配置选项的协同与所有 Telegraf 插件一样[[inputs.disque]]同样支持全局与插件级配置项例如通过name_override修改指标名、使用tags为指标补充自定义标签、通过fieldpass/fielddrop过滤字段以及配置插件执行顺序processor/aggregator 相对顺序。完整说明参见 docs/CONFIGURATION.md#plugins。采集流程与源码实现该插件的核心实现集中在 plugins/inputs/disque/disque.go整个采集链路可概括为解析服务器列表 → 建立 TCP 连接可选 AUTH→ 发送info\r\n→ 按 RESP 批量字符串协议读取 → 解析字段 → 写入 Accumulator。1. 多服务器并行采集Gather方法遍历Servers对每个地址执行url.Parse。若 URI 没有 scheme例如裸地址10.0.0.1:10000则自动回退为tcpschemedisque.go每个服务器在一个独立的 goroutine 中执行gatherServer通过sync.WaitGroup等待全部完成各节点采集互不阻塞disque.go。因此配置多个 Disque 节点时总采集耗时约等于最慢节点的耗时。2. 连接复用与 AUTH 认证连接在gatherServer内部按需建立并缓存在结构体字段d.c中disque.go若地址未携带端口自动追加:7711使用net.DialTimeout(tcp, addr.Host, defaultTimeout)建立连接若 URI 中包含密码disque://user:passwordhost会先发送AUTH password\r\n并读取服务端响应行——只有以RESP 简单字符串成功应答开头才继续否则返回认证错误每个采集周期通过SetDeadline续约 5 秒超时避免连接悬挂。3. 发送 INFO 命令并解析响应连接就绪后插件发送info\r\n命令disque.go。Disque 返回 RESP 批量字符串以$开头、后跟字节长度插件读取长度、逐行消费跳过空行和以#开头的节标题行再用SplitN(line, :, 2)拆出键值对disque.go。4. 字段白名单与类型转换并非 INFO 输出的所有键都会上报插件通过tracking映射表做了白名单过滤disque.go。解析出的值优先尝试strconv.ParseUint转换为无符号整数失败则回退strconv.ParseFloat转为浮点数从而保证内存、CPU 时间等小数指标不会被截断disque.go。每条指标携带标签disque_host取值为服务器 URI 的完整字符串如disque://10.10.3.33:18832便于在多节点部署中区分数据来源disque.go。指标输出与字段说明插件输出单条指标名disque标签为disque_host字段分为整数型与浮点型两类。下表给出完整字段清单、INFO 来源节及字段重命名情况字段名输出INFO 原始键类型含义uptimeuptime_in_secondsinteger服务运行时长秒clientsconnected_clientsinteger当前连接客户端数blocked_clientsblocked_clientsinteger处于阻塞状态的客户端数used_memoryused_memoryinteger分配器已使用内存字节used_memory_rssused_memory_rssinteger常驻内存 RSS字节used_memory_peakused_memory_peakinteger历史内存使用峰值字节total_connections_receivedtotal_connections_receivedinteger累计接收连接数total_commands_processedtotal_commands_processedinteger累计处理命令数instantaneous_ops_per_secinstantaneous_ops_per_secinteger瞬时每秒操作数latest_fork_useclatest_fork_usecinteger最近一次 fork 耗时微秒mem_fragmentation_ratiomem_fragmentation_ratiofloat内存碎片率used_cpu_sysused_cpu_sysfloat内核态 CPU 消耗秒used_cpu_userused_cpu_userfloat用户态 CPU 消耗秒used_cpu_sys_childrenused_cpu_sys_childrenfloat子进程内核态 CPU 消耗秒used_cpu_user_childrenused_cpu_user_childrenfloat子进程用户态 CPU 消耗秒registered_jobsregistered_jobsinteger当前注册的 job 总数registered_queuesregistered_queuesinteger当前注册的队列总数注意其中两处字段名发生了重命名uptime_in_seconds → uptime、connected_clients → clients其余字段与 INFO 原始键同名映射定义见 disque.go。由于 INFO 输出同时包含# Server、# Clients、# Memory、# Jobs、# Queues、# Persistence、# Stats、# CPU等多个小节而插件只挑选上述 17 个字段因此可以放心地用fieldpass进一步裁剪或用fielddrop剔除不关心的项。典型输出示例基于 disque_test.go 中模拟的 Disque INFO 响应采集到的指标形如disque,disque_hostdisque://localhost:52534 uptime1452705i,clients31i,blocked_clients13i,used_memory1840104i,used_memory_rss3227648i,used_memory_peak89603656i,total_connections_received5062777i,total_commands_processed12308396i,instantaneous_ops_per_sec18i,latest_fork_usec1644i,registered_jobs360i,registered_queues12i,mem_fragmentation_ratio1.75,used_cpu_sys19585.73,used_cpu_user11255.96,used_cpu_sys_children1.75,used_cpu_user_children1.91 1680000000000000000可以看到整数型字段以i后缀输出浮点型字段直接输出小数非常适合直接对接 InfluxDB 等时序数据库。测试验证插件配套的集成测试位于 plugins/inputs/disque/disque_test.goTestDisqueGeneratesMetricsIntegration在本地启动一个模拟 TCP 服务监听info\r\n请求并回放一段完整的 Disque INFO 文本随后校验Gather产出的全部 17 个字段与期望值一致含 uint64 与 float64 两种类型TestDisqueCanPullStatsFromMultipleServersIntegration验证多服务器场景下插件能否正常拉取统计信息。两个用例均标注为集成测试运行go test -tagsintegration ./plugins/inputs/disque/之类带完整依赖的测试环境时可执行在-short模式下会被跳过。测试中的testOutput同时是理解 Disque INFO 响应格式的最佳参考展示了插件实际消费的协议报文结构。使用建议与注意事项端口确认若 Disque 实例使用非默认端口必须在 URI 中显式写出例如disque://10.10.3.33:18832否则将回退到默认的7711认证配置密码通过 URI 携带disque://user:passwordhost:port插件会在首次连接时发送AUTH命令认证失败会以错误形式上报到 Accumulator可在 Telegraf 日志中排查采集频率插件本身没有独立间隔配置采集周期由全局[agent]的interval决定由于每次采集都复用长连接并续约 5 秒超时高频采集的开销相对可控字段筛选建议结合fieldpass只保留关注的核心指标如clients、used_memory、instantaneous_ops_per_sec、registered_jobs、registered_queues降低存储与传输开销适用前提Disque 本身是实验性项目插件文档同样标注为实验性质用于生产前请评估消息代理与采集链路两方面的稳定性。延伸阅读插件完整实现plugins/inputs/disque/disque.go配置模板plugins/inputs/disque/sample.conf集成测试与 INFO 响应样例plugins/inputs/disque/disque_test.go插件注册方式plugins/inputs/all/disque.goTelegraf 通用插件配置全局配置、字段/标签过滤、插件顺序docs/CONFIGURATION.md#plugins【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考