Telegraf cumulative_sum 处理器插件详解按序列对字段做累积求和【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf本指南围绕 Telegraf 仓库中的cumulative_sum处理器插件源码见 cumulative_sum.go官方文档见 README.md展开完整讲解它的作用、配置方式、运行机制与使用边界。该插件会在每个 metric 被更新时为指定数值字段生成持续递增的累积和xxx_sum非常适合为依赖“单调递增数值”的输出端如计算速率的监控后端提供前置数据处理。读完本文你将掌握cumulative_sum的字段筛选、缓存过期策略、通配符用法以及它与普通字段累加/聚合器之间的本质区别。插件元信息Telegraf v1.35.0引入 · 类别transformation转换· 支持平台all全平台。插件概述与适用场景cumulative_sum是一个典型的状态型转换处理器它按 metric 序列series在内存中缓存历史字段值每当收到一条新 metric 时就把该字段的历史累计值加上当前值并以新字段原字段名_sum追加到 metric 上输出。它的典型应用场景是配合需要单调递增数值的输出端。例如某些输出系统要求指标必须是持续递增的计数器counter而你的采集源提供的是“瞬时增量”或“周期快照”此时就可以用cumulative_sum把增量逐步累加把普通数值伪装成单调递增的序列从而满足输出端的数据模型要求。[!NOTE] 同一序列series内的 metric 是按到达顺序order of arrival累加的而不是按时间戳timestamp排序后累加这意味着乱序到达的指标会按实际到达次序累加累加结果与指标自身时间戳的顺序无关。这一点在高并发、乱序传递的管道场景中需要特别注意。配置详解cumulative_sum的完整配置如下与仓库中的 sample.conf 完全一致# Compute the cumulative sum of the given fields [[processors.cumulative_sum]] ## Numerical fields to be processed (accepting wildcards) # fields [*] ## Interval after which metrics are evicted from the cache and the ## sum values are reset to zero. A zero or unset value will keep the ## metric forever. ## It is strongly recommended to set an expiry interval to avoid ## growing memory usage when varying metric series are processed. # expiry_interval 0s参数一fields —— 参与累加的字段列表类型字符串数组[]string对应源码结构体中的Fields []string \toml:fields 字段。默认值[*]即对 metric 中所有字段都尝试累加。通配符支持 glob 风格的通配符例如fields [bytes_*]可只处理以bytes_开头的字段。实现细节在源码的Init()方法中若Fields为空会默认赋值为[*]随后通过filter.Compile(c.Fields)来自 filter 包编译成匹配器供每次处理时用c.accept.Match(field.Key)判断某字段是否参与累加。参数二expiry_interval —— 缓存过期时间类型duration 字符串如10m、1h对应ExpiryInterval config.Duration。默认值0s未设置。此时缓存条目永不失效累加值会一直累计下去。作用设置后超过该时长未被再次观测到的序列缓存条目会被从内存中清除其累计和归零当该序列后续再次出现时将从新基数重新开始累加。强烈建议配置当采集的 metric 序列组合不断变化例如 tag 值持续新增时若不设置过期时间缓存会随序列数量增长而持续占用内存。官方文档与源码注释均明确建议设置合理的过期区间以避免内存无限增长。一个面向实战的推荐配置示例[[processors.cumulative_sum]] fields [bytes_sent, bytes_received, packets_*] expiry_interval 10m工作原理从源码看累加流程核心逻辑集中在Apply()方法中cumulative_sum.go整个处理流程如下定位序列缓存对每条输入 metric 调用original.HashID()计算序列标识由测量名 tag 集合决定的哈希值在cache map[uint64]*entry中查找该序列的历史累计值若不存在则新建一个空的entry。复制 metric通过original.Copy()生成副本保证原始 metric 不被修改同时保留原有全部字段与 tag。字段筛选与转换遍历副本的FieldList()不匹配fields过滤器或未配置的字段直接保留、不处理对匹配字段调用internal.ToFloat64()实现在 internal/type_conversions.go尝试转为float64。该函数支持string、[]byte、bool、各种有符号/无符号整数、浮点数等类型如bool会转为 1/0转换失败如结构体、切片等不支持类型的字段会被跳过并输出一条 Trace 级别日志计算sum : stored.sums[field.Key] fv用m.AddField(field.Key_sum, sum)追加字段名_sum字段并回写缓存。更新与返回刷新该条目的seen时间戳把更新后的 entry 写回 cache输出携带_sum字段的新 metric并调用original.Accept()标记原 metric 已被消费。过期清理若ExpiryInterval 0以当前时间为基准计算阈值通过maps.DeleteFunc删除所有seen早于阈值的缓存条目。插件通过包内init()函数调用processors.Add(cumulative_sum, ...)完成注册并经由 plugins/processors/all/cumulative_sum.go 统一加载构建标签!custom || processors || processors.cumulative_sum因此它已包含在标准发行版中无需额外编译。官方示例解析README 中给出的示例直观展示了处理前后的差异其中-为输入为输出- net,hostserver01 bytes_sent1000,bytes_received500 - net,hostserver01 bytes_sent2500,bytes_received1500 - net,hostserver01 bytes_sent3000,bytes_received2500 net,hostserver01 bytes_sent1000,bytes_sent_sum1000,bytes_received500,bytes_received_sum500 net,hostserver01 bytes_sent2500,bytes_sent_sum3500,bytes_received1500,bytes_received_sum2000 net,hostserver01 bytes_sent3000,bytes_sent_sum6500,bytes_received2500,bytes_received_sum4500从中可以看到三个关键行为保留原始字段bytes_sent、bytes_received原样保留_sum只是追加字段不会覆盖原始值逐条累积同一序列net,hostserver01的第二条输入中bytes_sent_sum 1000 2500 3500第三条为3500 3000 6500呈现单调递增趋势按到达顺序累加示例输入本身按时间先后到达因此累加顺序与时间顺序一致但若 metric 乱序到达累加依然以到达顺序为准见上文 NOTE。测试用例佐证行为边界一目了然仓库中的 cumulative_sum_test.go 用三组用例精确定义了插件的行为边界是对文档的有力补充all fields keep original fields默认fields为 nil 时所有字段都会被处理。测试显示bool字段healthyfalse被转换为healthy_sum0对应ToFloat64的 bool 处理规则int64字段error_counter10得到error_counter_sum10字符串字段errormachine broken无法转浮点被跳过且不生成_sum。这印证了“字段可转为 float 才会被累加”的实现事实。filter value remove original配置fields[value]后只有value字段生成value_sum其余字段如healthy、error_counter、error原样保留、不参与累加。multiple metrics同一测量名 不同 tag 的序列各自独立累加tagsome tag累计到 3.0taganother tag保持 4.4证明缓存是以HashID()测量名 tag 组合为粒度的。TestCacheExpiry模拟expiry_interval10s时若某序列超过 10 秒未被观测其缓存条目被清除、累计值归零后续再出现时从新基数累加4.4不再叠加进3.2而是重新从4.4开始。使用注意与边界综合文档与源码使用cumulative_sum时需要注意以下几点乱序问题按到达顺序累加而非按时间戳累加。若你的数据源存在乱序投递且对累加顺序敏感请先在上游保证有序或在管道中用其他机制排序。非数值字段只有能通过internal.ToFloat64转成float64的字段才会被累加bool会被转成 0/1。无法转换的字段被静默跳过仅记录 Trace 日志不会报错中断。内存管理强烈建议设置expiry_interval。否则当 tag 组合持续变化时缓存条目只增不减内存占用会随序列规模线性增长。设置过期时间后长期未出现的序列会被回收其累计和自动归零。与聚合器aggregator的区别cumulative_sum是流式转换处理器每收到一条 metric 就立即输出携带累计值的 metric不依赖时间窗口而聚合器如minmax、basicstats等通常在窗口结束后才批量输出。它也不等于“对字段求和后再输出单一值”——它保留原始字段并逐条追加_sum。处理器顺序cumulative_sum属于 processors 阶段其执行顺序受全局插件顺序配置影响具体可参考 docs/CONFIGURATION.md 中的插件排序说明该处理器同样支持所有全局插件配置项如namepass、tagexclude等详见 docs/includes/plugin_config.md。总结cumulative_sum是 Telegraf 中实现“按序列单调递增累加”最直接的处理器一条[[processors.cumulative_sum]]配置配合fields通配筛选与expiry_interval内存治理即可把瞬时增量转换为单调递增计数适配依赖 counter 语义的输出端。理解其“按到达顺序累加、按序列独立缓存、按过期时间回收”三大机制你就能在真实管道中安全、可控地使用它。【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
