Telegraf Aggregator 与 Processor 插件详解指标处理管道的顺序控制、过滤机制与窗口聚合原理【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf本文基于 Telegraf 官方文档 docs/AGGREGATORS_AND_PROCESSORS.md系统讲解夹在 Input 与 Output 之间的两类处理插件——Processor处理器与 Aggregator聚合器——的工作原理、执行顺序、指标过滤机制与关键配置项并结合仓库源码深入剖析管道的构建流程、skip_processors_*选项的实现逻辑以及聚合窗口的判定算法帮助你在生产环境中正确编排指标处理链、避免指标被二次处理这类隐蔽问题。1. 架构总览插件在数据流中的位置Telegraf 拥有 aggregator 和 processor 两类插件概念它们位于输入插件inputs与输出插件outputs之间允许用户对采集到的指标做额外的加工处理或统计聚合。文档中的架构示意如下┌───────────┐ │ │ │ CPU │───┐ │ │ │ └───────────┘ │ │ ┌───────────┐ │ ┌───────────┐ │ │ │ │ │ │ Memory │───┤ ┌──▶│ InfluxDB │ │ │ │ │ │ │ └───────────┘ │ ┌─────────────┐ ┌─────────────┐ │ └───────────┘ │ │ │ │Aggregators │ │ ┌───────────┐ ┌───────────┐ │ │Processors │ │ - mean │ │ │ │ │ │ │ │ - transform │ │ - quantiles │ │ │ File │ │ MySQL │───┼───▶│ - decorate │────▶│ - min/max │───┼──▶│ │ │ │ │ │ - filter │ │ - count │ │ │ │ └───────────┘ │ │ │ │ │ │ └───────────┘ │ └─────────────┘ └─────────────┘ │ ┌───────────┐ │ │ ┌───────────┐ │ │ │ │ │ │ │ SNMP │───┤ └──▶│ Kafka │ │ │ │ │ │ └───────────┘ │ └───────────┘ │ ┌───────────┐ │ │ │ │ │ Docker │───┘ │ │ └───────────┘从数据流方向看指标依次经过输入插件 → Processor第一轮→ Aggregator → Processor第二轮→ 输出插件。两类插件的语义截然不同后文分别展开。2. 执行顺序Ordering与 skip_processors_* 配置2.1 默认顺序Processor 会被执行两次官方文档明确规定的默认行为是Processor 先运行然后是 Aggregator之后 Processor 再次运行。允许 Processor 在 Aggregator 之后再次运行是为了让用户有机会对聚合出来的指标再做一次处理例如给聚合结果重命名、加标签。但这个行为对新手而言容易造成困惑也可能导致指标出现怪异的结果——文档给出的典型反例是如果某个 Processor 对数据做了缩放scale开启默认顺序后数据会被缩放两次对应两个 Agent 级配置项将skip_processors_before_aggregators设为true可禁用 Processor 的第一轮即 Aggregator 之前运行将skip_processors_after_aggregators设为true可禁用第二轮即 Aggregator 之后运行另一个通用替代方案是使用下文介绍的指标过滤metric filtering。配置示例Agent 段[agent] # 跳过聚合器之后的第二轮 processor 处理 skip_processors_after_aggregators true # 或跳过聚合器之前的第一轮 processor 处理 # skip_processors_before_aggregators true2.2 源码层面的实现证据这两个选项定义在 config/config.go 的Agent配置结构中字段注释与文档描述一致SkipProcessorsAfterAggregators *booltoml:skip_processors_after_aggregatorsBy default, processors are run a second time after aggregators. Changing this setting to true will skip the second run of processors.SkipProcessorsBeforeAggregators booltoml:skip_processors_before_aggregatorsBy default, processors are run a first time before aggregators. Changing this setting to true will skip the first run of processors.在配置加载阶段 config/config.go 中有两处关键逻辑互斥校验两个开关不允许同时为true都禁用则没有意义LoadAll会直接返回错误cannot set both skip_processors_before_aggregators and skip_processors_after_aggregators as true该行为有对应的测试用例见 config/config_test.go。裁剪处理器链若设置了skip_processors_before_aggregators则c.Processors会被清空为空列表第一轮处理链从管道中彻底移除。真正的管道搭建发生在 agent/agent.go 的Run方法中Agent 先启动 outputs若存在聚合器则建立aggregatorUnit当配置了AggProcessors且未跳过时先接入第二轮 Processor 链若存在Processors则在其上游再接入第一轮 Processor 链最后 inputs 写入最上游通道。各段处理逻辑以独立的 goroutinerunProcessors/runAggregators/runOutputs并发运行通过 channel 串联。仓库内的端到端测试用例 agent/testcases/processor-order-explicit/telegraf.conf 以及agent/testcases/下aggregators-skip-processors、aggregators-rerun-processors等目录分别验证了显式排序与跳过行为的最终指标输出。2.3 重要提示默认值将在 v1.40.0 变更agent/agent.go 的Run方法在启动时会检查如果未显式设置skip_processors_after_aggregators会打印黄色警告The default value of skip_processors_after_aggregators will change to true with Telegraf v1.40.0! If you need the current default behavior, please explicitly set the option to false!也就是说当前版本默认仍会执行第二轮 Processor值为false但官方计划在未来版本把聚合后不再重跑 Processor变成默认行为。如果你依赖二次处理聚合结果的能力建议在配置中显式写出该选项以消除歧义。2.4 Processor 的显式顺序order选项同一阶段的多个 Processor 之间还可按order字段显式排序。config/config.go 中对c.Processors与c.AggProcessors执行稳定排序sort.Stable排序键为每个 Processor 配置的Order字段定义于 models/running_processor.go 的ProcessorConfig未显式指定的保持配置文件出现顺序。测试配置 agent/testcases/processor-order-explicit/telegraf.conf 给出了实例[[processors.date]] field_key timestamp date_format 2006-01-02T15:04:05.999999999Z timezone UTC order 2 [[processors.starlark]] source def apply(metric): ... return metrics order 13. 指标过滤Metric Filtering使用指标过滤机制可以控制哪些指标真正进入某个 Processor 或 Aggregator。被过滤掉的指标会绕过该插件原样向下游传递——即过滤是绕过而非丢弃。这一语义在源码中有直接体现。models/running_processor.go 的Add方法中ok, err : rp.Config.Filter.Select(m) if err ! nil { rp.log.Errorf(filtering failed: %v, err) } else if !ok { // pass downstream acc.AddMetric(m) return nil }Filter.Select返回false时指标不做任何修改直接写入下游累积器acc.AddMetric。只有通过筛选的指标才会经过Filter.Modify应用tagpass/tagdrop等修改型选项再交给插件本体。Aggregator侧逻辑相同models/running_aggregator.go 中Select不通过时直接返回原始指标继续流向输出。过滤选项本身namepass、nameprefix、namesuffix、tagpass、tagdrop、taginclude、tagexclude等的完整说明见 docs/CONFIGURATION.md 的 Measurement Filtering 章节。4. Processor 插件随路而过的即时处理Processor 插件在指标经过时立即处理并基于所处理的值即时产出结果例如打印所有经过的指标、给所有指标追加标签等。它的核心特征是无状态窗口不等待不攒批来一条处理一条。4.1 两种插件接口Processor 插件支持两种实现方式定义在仓库根目录的 processor.goProcessor同步式仅实现Apply(in ...Metric) []Metric接口注释指出其效率极高若不需要异步写指标应优先使用StreamingProcessor流式实现Start(acc Accumulator) error、Add(metric, acc)与Stop()适合内部需要 goroutine 做慢速并发的场景例如reverse_dns插件使用了 worker 池。注释中特别提醒不应无限制地派生 goroutine且不需要向下游传递的指标应调用metric.Drop()而不是简单地不调用acc.AddMetric()。运行时包装层 models/running_processor.go 还会为每个实例注册selfstat错误计数与插件日志并支持按order排序。4.2 可用的 Processor 插件完整列表见 plugins/processors/ 目录当前仓库包含部分列举aws_ec2、batch、clone、converter、cumulative_sum、date、dedup、enum、execd、filepath、filter、ifname、lookup、noise、override、parser、pivot、port_name、printer、regex、rename、reverse_dns、round、s2geo、scale、snmp_lookup、split、starlark、strings、tag_limit、template、timestamp、topk、unpivot等。每个插件目录下均带有sample.conf与README.md可直接复制配置并参照注释使用。5. Aggregator 插件窗口聚合与period语义Aggregator 插件比 Processor 复杂得多它们通常产出的是新的聚合指标aggregate metrics如运行均值、最小值、最大值、标准差等。因此所有 Aggregator 插件都配置一个period参数——period是每个聚合值所代表的指标窗口大小输出的聚合指标即为过去period秒内所有指标的聚合值。5.1 关键配置项period聚合窗口长度决定每次推送聚合结果并清空缓存的周期drop_original由于很多用户只关心聚合结果而不关心每一条原始指标该参数为true时 Telegraf 只输出聚合指标、丢弃原始指标taginclude由于聚合是按插件收到的每个measurement field 唯一 tag 组合生成的利用taginclude可以只按指定标签分组聚合例如只按hostname聚合而不按全部标签细粒度展开。示例见 plugins/aggregators/basicstats/sample.conf# Keep the aggregate basicstats of each metric passing through. [[aggregators.basicstats]] ## The period on which to flush clear the aggregator. # period 30s ## If true, the original metric will be dropped by the ## aggregator and will not get sent to the output plugins. # drop_original false ## Configures which basic stats to push as fields # stats [count,min,max,mean,variance,stdev]5.2 插件列表完整列表见 plugins/aggregators/ 目录当前仓库包含basicstats、derivative、final、histogram、merge、minmax、quantile、starlark、valuecounter。5.3 聚合窗口的源码实现纵深原理Aggregator 插件本体只需实现 aggregator.go 中的三个方法RunningAggregator保证Add/Push/Reset不会并发调用插件内无需自行加锁type Aggregator interface { PluginDescriber // Add the metric to the aggregator. Add(in Metric) // Push pushes the current aggregates to the accumulator. Push(acc Accumulator) // Reset resets the aggregators caches and aggregates. Reset() }官方文档特别强调Aggregator 只聚合其 period 窗口内即now() - period之后的指标时间戳早于now() - period的数据无法被纳入聚合。从 models/running_aggregator.go 的AggregatorConfig结构可以进一步看到窗口机制的细节除文档所述的Period与DropOriginal外还包含Delay与Grace两个time.Duration字段以及NameOverride、MeasurementPrefix、MeasurementSuffix、Tags等输出修饰字段。Add方法models/running_aggregator.go的核心判定为if m.Time().Before(r.periodStart.Add(-r.Config.Grace)) || m.Time().After(r.periodEnd.Add(r.Config.Delay)) { // Metric is outside aggregation window; discarding r.MetricsDropped.Incr(1) return r.Config.DropOriginal }即早于periodStart - Grace或晚于periodEnd Delay的指标会被丢弃并计入metrics_dropped自统计这也解释了为什么历史数据进不了聚合结果。此外加入聚合前指标会先做拷贝metric.FromMetric注释说明原因是聚合不能基于历史数据失败投递且等待聚合推送会引入巨大延迟。Push方法models/running_aggregator.go在每次刷新时先计算下一个[since, until]窗口长度为一个period并按period对齐若因时钟调整/休眠导致现在不在下一窗口内则重对齐到当前时间随后调用插件的Push(acc)产出聚合指标最后Reset()清空内部缓存——这与 sample.conf 中 The period on which to flush clear the aggregator 的注释一一对应。运行期自统计aggregate.metrics_pushed、aggregate.metrics_filtered、aggregate.metrics_dropped、aggregate.push_time_ns也在此处注册可用于观察聚合链路的健康状况。6. 实战配置示例组合顺序控制与过滤综合上述机制一个聚合产出均值/最大值原始指标全量透传聚合结果再打标签的典型配置骨架如下以basicstats聚合器为例[agent] interval 10s flush_interval 10s # 明确声明聚合后不再重跑 processor避免依赖默认值 skip_processors_after_aggregators true [[inputs.cpu]] percpu false # 第一轮 processor仅处理 cpu 指标为其加一个标签 [[processors.rename]] order 1 namepass [cpu] # 其余指标绕过本插件原样向下游传递 [[processors.rename.replace]] field usage_idle with idle # 聚合器10 秒窗口只按 cpu 相关指标聚合 [[aggregators.basicstats]] period 10s namepass [cpu] stats [mean, min, max] # 若只想保留聚合结果 # drop_original true [[outputs.file]] files [stdout]要点回顾未配置skip_processors_after_aggregators时processor 会对聚合结果再执行一次当前默认行为v1.40.0 起默认将变为跳过namepass等过滤选项使不匹配的指标绕过插件而不是被丢弃这是实现每个插件只处理部分指标的标准手段聚合结果按measurement field 唯一 tag 组合生成如需粗粒度分组请配合taginclude早于当前 period 窗口的历史数据不会被聚合迟到指标若超出Grace容差会被直接丢弃。7. 小结机制文档/源码依据关键结论默认执行顺序docs/AGGREGATORS_AND_PROCESSORS.md、agent/agent.goProcessor → Aggregator → Processor第二轮处理聚合结果禁用第一轮skip_processors_before_aggregators trueconfig/config.go配置加载时清空c.Processors禁用第二轮skip_processors_after_aggregators trueconfig/config.go两开关互斥均为 true 时报错默认值 v1.40.0 起计划改为true指标过滤models/running_processor.go被过滤的指标绕过插件原样传递而非丢弃聚合窗口models/running_aggregator.go只聚合period窗口内指标窗口外指标计入metrics_dropped原始指标取舍drop_originalmodels/running_aggregator.gotrue时仅输出聚合指标掌握了两轮 Processor 顺序 两个 skip 开关 指标过滤 period 窗口语义这四块拼图就能在 Telegraf 中精确控制指标从采集到落盘的每一段变换路径。【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/13 18:45:34 · 阅读 1.2k Telegraf Aggregator 与 Processor 插件详解：指标处理管道的顺序控制、过滤机制与窗口聚合原理
Telegraf Aggregator 与 Processor 插件详解指标处理管道的顺序控制、过滤机制与窗口聚合原理【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf本文基于 Telegraf 官方文档 docs/AGGREGATORS_AND_PROCESSORS.md系统讲解夹在 Input 与 Output 之间的两类处理插件——Processor处理器与 Aggregator聚合器——的工作原理、执行顺序、指标过滤机制与关键配置项并结合仓库源码深入剖析管道的构建流程、skip_processors_*选项的实现逻辑以及聚合窗口的判定算法帮助你在生产环境中正确编排指标处理链、避免指标被二次处理这类隐蔽问题。1. 架构总览插件在数据流中的位置Telegraf 拥有 aggregator 和 processor 两类插件概念它们位于输入插件inputs与输出插件outputs之间允许用户对采集到的指标做额外的加工处理或统计聚合。文档中的架构示意如下┌───────────┐ │ │ │ CPU │───┐ │ │ │ └───────────┘ │ │ ┌───────────┐ │ ┌───────────┐ │ │ │ │ │ │ Memory │───┤ ┌──▶│ InfluxDB │ │ │ │ │ │ │ └───────────┘ │ ┌─────────────┐ ┌─────────────┐ │ └───────────┘ │ │ │ │Aggregators │ │ ┌───────────┐ ┌───────────┐ │ │Processors │ │ - mean │ │ │ │ │ │ │ │ - transform │ │ - quantiles │ │ │ File │ │ MySQL │───┼───▶│ - decorate │────▶│ - min/max │───┼──▶│ │ │ │ │ │ - filter │ │ - count │ │ │ │ └───────────┘ │ │ │ │ │ │ └───────────┘ │ └─────────────┘ └─────────────┘ │ ┌───────────┐ │ │ ┌───────────┐ │ │ │ │ │ │ │ SNMP │───┤ └──▶│ Kafka │ │ │ │ │ │ └───────────┘ │ └───────────┘ │ ┌───────────┐ │ │ │ │ │ Docker │───┘ │ │ └───────────┘从数据流方向看指标依次经过输入插件 → Processor第一轮→ Aggregator → Processor第二轮→ 输出插件。两类插件的语义截然不同后文分别展开。2. 执行顺序Ordering与 skip_processors_* 配置2.1 默认顺序Processor 会被执行两次官方文档明确规定的默认行为是Processor 先运行然后是 Aggregator之后 Processor 再次运行。允许 Processor 在 Aggregator 之后再次运行是为了让用户有机会对聚合出来的指标再做一次处理例如给聚合结果重命名、加标签。但这个行为对新手而言容易造成困惑也可能导致指标出现怪异的结果——文档给出的典型反例是如果某个 Processor 对数据做了缩放scale开启默认顺序后数据会被缩放两次对应两个 Agent 级配置项将skip_processors_before_aggregators设为true可禁用 Processor 的第一轮即 Aggregator 之前运行将skip_processors_after_aggregators设为true可禁用第二轮即 Aggregator 之后运行另一个通用替代方案是使用下文介绍的指标过滤metric filtering。配置示例Agent 段[agent] # 跳过聚合器之后的第二轮 processor 处理 skip_processors_after_aggregators true # 或跳过聚合器之前的第一轮 processor 处理 # skip_processors_before_aggregators true2.2 源码层面的实现证据这两个选项定义在 config/config.go 的Agent配置结构中字段注释与文档描述一致SkipProcessorsAfterAggregators *booltoml:skip_processors_after_aggregatorsBy default, processors are run a second time after aggregators. Changing this setting to true will skip the second run of processors.SkipProcessorsBeforeAggregators booltoml:skip_processors_before_aggregatorsBy default, processors are run a first time before aggregators. Changing this setting to true will skip the first run of processors.在配置加载阶段 config/config.go 中有两处关键逻辑互斥校验两个开关不允许同时为true都禁用则没有意义LoadAll会直接返回错误cannot set both skip_processors_before_aggregators and skip_processors_after_aggregators as true该行为有对应的测试用例见 config/config_test.go。裁剪处理器链若设置了skip_processors_before_aggregators则c.Processors会被清空为空列表第一轮处理链从管道中彻底移除。真正的管道搭建发生在 agent/agent.go 的Run方法中Agent 先启动 outputs若存在聚合器则建立aggregatorUnit当配置了AggProcessors且未跳过时先接入第二轮 Processor 链若存在Processors则在其上游再接入第一轮 Processor 链最后 inputs 写入最上游通道。各段处理逻辑以独立的 goroutinerunProcessors/runAggregators/runOutputs并发运行通过 channel 串联。仓库内的端到端测试用例 agent/testcases/processor-order-explicit/telegraf.conf 以及agent/testcases/下aggregators-skip-processors、aggregators-rerun-processors等目录分别验证了显式排序与跳过行为的最终指标输出。2.3 重要提示默认值将在 v1.40.0 变更agent/agent.go 的Run方法在启动时会检查如果未显式设置skip_processors_after_aggregators会打印黄色警告The default value of skip_processors_after_aggregators will change to true with Telegraf v1.40.0! If you need the current default behavior, please explicitly set the option to false!也就是说当前版本默认仍会执行第二轮 Processor值为false但官方计划在未来版本把聚合后不再重跑 Processor变成默认行为。如果你依赖二次处理聚合结果的能力建议在配置中显式写出该选项以消除歧义。2.4 Processor 的显式顺序order选项同一阶段的多个 Processor 之间还可按order字段显式排序。config/config.go 中对c.Processors与c.AggProcessors执行稳定排序sort.Stable排序键为每个 Processor 配置的Order字段定义于 models/running_processor.go 的ProcessorConfig未显式指定的保持配置文件出现顺序。测试配置 agent/testcases/processor-order-explicit/telegraf.conf 给出了实例[[processors.date]] field_key timestamp date_format 2006-01-02T15:04:05.999999999Z timezone UTC order 2 [[processors.starlark]] source def apply(metric): ... return metrics order 13. 指标过滤Metric Filtering使用指标过滤机制可以控制哪些指标真正进入某个 Processor 或 Aggregator。被过滤掉的指标会绕过该插件原样向下游传递——即过滤是绕过而非丢弃。这一语义在源码中有直接体现。models/running_processor.go 的Add方法中ok, err : rp.Config.Filter.Select(m) if err ! nil { rp.log.Errorf(filtering failed: %v, err) } else if !ok { // pass downstream acc.AddMetric(m) return nil }Filter.Select返回false时指标不做任何修改直接写入下游累积器acc.AddMetric。只有通过筛选的指标才会经过Filter.Modify应用tagpass/tagdrop等修改型选项再交给插件本体。Aggregator侧逻辑相同models/running_aggregator.go 中Select不通过时直接返回原始指标继续流向输出。过滤选项本身namepass、nameprefix、namesuffix、tagpass、tagdrop、taginclude、tagexclude等的完整说明见 docs/CONFIGURATION.md 的 Measurement Filtering 章节。4. Processor 插件随路而过的即时处理Processor 插件在指标经过时立即处理并基于所处理的值即时产出结果例如打印所有经过的指标、给所有指标追加标签等。它的核心特征是无状态窗口不等待不攒批来一条处理一条。4.1 两种插件接口Processor 插件支持两种实现方式定义在仓库根目录的 processor.goProcessor同步式仅实现Apply(in ...Metric) []Metric接口注释指出其效率极高若不需要异步写指标应优先使用StreamingProcessor流式实现Start(acc Accumulator) error、Add(metric, acc)与Stop()适合内部需要 goroutine 做慢速并发的场景例如reverse_dns插件使用了 worker 池。注释中特别提醒不应无限制地派生 goroutine且不需要向下游传递的指标应调用metric.Drop()而不是简单地不调用acc.AddMetric()。运行时包装层 models/running_processor.go 还会为每个实例注册selfstat错误计数与插件日志并支持按order排序。4.2 可用的 Processor 插件完整列表见 plugins/processors/ 目录当前仓库包含部分列举aws_ec2、batch、clone、converter、cumulative_sum、date、dedup、enum、execd、filepath、filter、ifname、lookup、noise、override、parser、pivot、port_name、printer、regex、rename、reverse_dns、round、s2geo、scale、snmp_lookup、split、starlark、strings、tag_limit、template、timestamp、topk、unpivot等。每个插件目录下均带有sample.conf与README.md可直接复制配置并参照注释使用。5. Aggregator 插件窗口聚合与period语义Aggregator 插件比 Processor 复杂得多它们通常产出的是新的聚合指标aggregate metrics如运行均值、最小值、最大值、标准差等。因此所有 Aggregator 插件都配置一个period参数——period是每个聚合值所代表的指标窗口大小输出的聚合指标即为过去period秒内所有指标的聚合值。5.1 关键配置项period聚合窗口长度决定每次推送聚合结果并清空缓存的周期drop_original由于很多用户只关心聚合结果而不关心每一条原始指标该参数为true时 Telegraf 只输出聚合指标、丢弃原始指标taginclude由于聚合是按插件收到的每个measurement field 唯一 tag 组合生成的利用taginclude可以只按指定标签分组聚合例如只按hostname聚合而不按全部标签细粒度展开。示例见 plugins/aggregators/basicstats/sample.conf# Keep the aggregate basicstats of each metric passing through. [[aggregators.basicstats]] ## The period on which to flush clear the aggregator. # period 30s ## If true, the original metric will be dropped by the ## aggregator and will not get sent to the output plugins. # drop_original false ## Configures which basic stats to push as fields # stats [count,min,max,mean,variance,stdev]5.2 插件列表完整列表见 plugins/aggregators/ 目录当前仓库包含basicstats、derivative、final、histogram、merge、minmax、quantile、starlark、valuecounter。5.3 聚合窗口的源码实现纵深原理Aggregator 插件本体只需实现 aggregator.go 中的三个方法RunningAggregator保证Add/Push/Reset不会并发调用插件内无需自行加锁type Aggregator interface { PluginDescriber // Add the metric to the aggregator. Add(in Metric) // Push pushes the current aggregates to the accumulator. Push(acc Accumulator) // Reset resets the aggregators caches and aggregates. Reset() }官方文档特别强调Aggregator 只聚合其 period 窗口内即now() - period之后的指标时间戳早于now() - period的数据无法被纳入聚合。从 models/running_aggregator.go 的AggregatorConfig结构可以进一步看到窗口机制的细节除文档所述的Period与DropOriginal外还包含Delay与Grace两个time.Duration字段以及NameOverride、MeasurementPrefix、MeasurementSuffix、Tags等输出修饰字段。Add方法models/running_aggregator.go的核心判定为if m.Time().Before(r.periodStart.Add(-r.Config.Grace)) || m.Time().After(r.periodEnd.Add(r.Config.Delay)) { // Metric is outside aggregation window; discarding r.MetricsDropped.Incr(1) return r.Config.DropOriginal }即早于periodStart - Grace或晚于periodEnd Delay的指标会被丢弃并计入metrics_dropped自统计这也解释了为什么历史数据进不了聚合结果。此外加入聚合前指标会先做拷贝metric.FromMetric注释说明原因是聚合不能基于历史数据失败投递且等待聚合推送会引入巨大延迟。Push方法models/running_aggregator.go在每次刷新时先计算下一个[since, until]窗口长度为一个period并按period对齐若因时钟调整/休眠导致现在不在下一窗口内则重对齐到当前时间随后调用插件的Push(acc)产出聚合指标最后Reset()清空内部缓存——这与 sample.conf 中 The period on which to flush clear the aggregator 的注释一一对应。运行期自统计aggregate.metrics_pushed、aggregate.metrics_filtered、aggregate.metrics_dropped、aggregate.push_time_ns也在此处注册可用于观察聚合链路的健康状况。6. 实战配置示例组合顺序控制与过滤综合上述机制一个聚合产出均值/最大值原始指标全量透传聚合结果再打标签的典型配置骨架如下以basicstats聚合器为例[agent] interval 10s flush_interval 10s # 明确声明聚合后不再重跑 processor避免依赖默认值 skip_processors_after_aggregators true [[inputs.cpu]] percpu false # 第一轮 processor仅处理 cpu 指标为其加一个标签 [[processors.rename]] order 1 namepass [cpu] # 其余指标绕过本插件原样向下游传递 [[processors.rename.replace]] field usage_idle with idle # 聚合器10 秒窗口只按 cpu 相关指标聚合 [[aggregators.basicstats]] period 10s namepass [cpu] stats [mean, min, max] # 若只想保留聚合结果 # drop_original true [[outputs.file]] files [stdout]要点回顾未配置skip_processors_after_aggregators时processor 会对聚合结果再执行一次当前默认行为v1.40.0 起默认将变为跳过namepass等过滤选项使不匹配的指标绕过插件而不是被丢弃这是实现每个插件只处理部分指标的标准手段聚合结果按measurement field 唯一 tag 组合生成如需粗粒度分组请配合taginclude早于当前 period 窗口的历史数据不会被聚合迟到指标若超出Grace容差会被直接丢弃。7. 小结机制文档/源码依据关键结论默认执行顺序docs/AGGREGATORS_AND_PROCESSORS.md、agent/agent.goProcessor → Aggregator → Processor第二轮处理聚合结果禁用第一轮skip_processors_before_aggregators trueconfig/config.go配置加载时清空c.Processors禁用第二轮skip_processors_after_aggregators trueconfig/config.go两开关互斥均为 true 时报错默认值 v1.40.0 起计划改为true指标过滤models/running_processor.go被过滤的指标绕过插件原样传递而非丢弃聚合窗口models/running_aggregator.go只聚合period窗口内指标窗口外指标计入metrics_dropped原始指标取舍drop_originalmodels/running_aggregator.gotrue时仅输出聚合指标掌握了两轮 Processor 顺序 两个 skip 开关 指标过滤 period 窗口语义这四块拼图就能在 Telegraf 中精确控制指标从采集到落盘的每一段变换路径。【免费下载链接】telegrafAgent for collecting, processing, aggregating, and writing metrics, logs, and other arbitrary data.项目地址: https://gitcode.com/GitHub_Trending/te/telegraf创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考