How Telegraf's Plugin Architecture Works: Inputs, Processors, Aggregators, and Outputs

Telegraf's plugin architecture is a modular data pipeline built on four distinct Go interfaces—Input, Processor, Aggregator, and Output—where each plugin registers itself via package-level Add functions and processes metrics in a strict Input → Processor → Aggregator → Output execution order.

Telegraf is an open-source metrics collector built around a flexible Telegraf plugin architecture that separates data collection, transformation, aggregation, and delivery into discrete components. This design allows developers to extend functionality by implementing simple interfaces defined in the core repository, without modifying the agent's internal execution engine.

Core Plugin Interfaces

Every plugin type implements a specific Go interface and embeds PluginDescriber for metadata introspection. The primary interfaces are defined in the root directory and plugins/ subdirectories:

Interface	Primary Method	Source File
Input	Gather(Accumulator) error	[input.go](https://github.com/influxdata/telegraf/blob/master/input.go)
Processor	Apply(in Metric, out Plugin) error	[plugins/processors/all/processors.go](https://github.com/influxdata/telegraf/blob/master/plugins/processors/all/processors.go)
Aggregator	Add(in Metric) / Push(Accumulator) / Reset()	[aggregator.go](https://github.com/influxdata/telegraf/blob/master/aggregator.go)
Output	Write([]Metric) error	[output.go](https://github.com/influxdata/telegraf/blob/master/output.go)

The PluginDescriber interface, referenced in [config/plugin_selector.go](https://github.com/influxdata/telegraf/blob/master/config/plugin_selector.go), requires Name(), Description(), and SampleConfig() methods, enabling the TOML configuration loader to validate plugin blocks and generate documentation.

Input Plugins

Input plugins serve as the entry point for metrics collection, implementing the Gather method to pull data from system sources or remote APIs.

Interface and Registration

In [input.go](https://github.com/influxdata/telegraf/blob/master/input.go), the Input interface defines:

type Input interface {
    PluginDescriber
    Gather(Accumulator) error
}

Each input registers via an init() function that calls inputs.Add, populating the global inputs.Plugins map. For example, in [plugins/inputs/apache/apache.go](https://github.com/influxdata/telegraf/blob/master/plugins/inputs/apache/apache.go):

func init() {
    inputs.Add("apache", func() telegraf.Input {
        return &Apache{}
    })
}

During startup, the configuration parser looks up the plugin name in the registry, instantiates the struct, and schedules Gather calls on the configured interval. The Accumulator argument provides thread-safe methods like AddFields() to buffer metrics for downstream processing.

Processor Plugins

Processor plugins sit between inputs and aggregators, enabling real-time transformation of individual metrics.

The Apply Method

Defined in [plugins/processors/all/processors.go](https://github.com/influxdata/telegraf/blob/master/plugins/processors/all/processors.go), the Processor interface requires:

type Processor interface {
    PluginDescriber
    Apply(in telegraf.Metric, out telegraf.Plugin) error
}

Processors are stateless by design. They receive a metric, optionally modify fields or tags, and forward it by calling out.AddMetric(). Multiple processors execute sequentially in the order defined in the TOML [[processors]] array.

Registration Pattern

func init() {
    processors.Add("rename", func() telegraf.Processor {
        return &Rename{}
    })
}

Aggregator Plugins

Aggregator plugins collect metrics over time windows, computing statistical summaries before emitting rolled-up data.

Add, Push, and Reset Lifecycle

The Aggregator interface in [aggregator.go](https://github.com/influxdata/telegraf/blob/master/aggregator.go) specifies three lifecycle hooks:

type Aggregator interface {
    PluginDescriber
    Add(in telegraf.Metric)
    Push(acc telegraf.Accumulator)
    Reset()
}

The RunningAggregator wrapper in the core engine guarantees these methods are never called concurrently, eliminating the need for mutexes in plugin implementations. After each flush interval, Telegraf calls Push() to write aggregated points, followed by Reset() to clear internal state.

Registration

Aggregators register using the same pattern as inputs:

func init() {
    aggregators.Add("minmax", func() telegraf.Aggregator {
        return &MinMax{}
    })
}

Output Plugins

Output plugins handle the final delivery of metrics to external databases, message queues, or file systems.

Connection Lifecycle

The Output interface in [output.go](https://github.com/influxdata/telegraf/blob/master/output.go) defines connection-oriented methods:

type Output interface {
    PluginDescriber
    Connect() error
    Write([]telegraf.Metric) error
    Close() error
}

Telegraf calls Connect() once at startup, then batches metrics into Write() calls at the flush interval. Some outputs implement AggregatingOutput, allowing them to receive pre-aggregated data directly via the aggregator interface methods rather than raw metric slices.

Pipeline Execution Flow

The wiring logic in [cmd/telegraf/telegraf.go](https://github.com/influxdata/telegraf/blob/master/cmd/telegraf/telegraf.go) orchestrates the data flow:

1. Configuration Parsing: The config package reads TOML and instantiates plugins using the registration maps (inputs.Plugins, processors.Plugins, etc.) and label selectors from [config/plugin_selector.go](https://github.com/influxdata/telegraf/blob/master/config/plugin_selector.go).
2. Gather Phase: Inputs call Gather(), writing to an accumulator that feeds metrics into the processor chain.
3. Processing: Each metric traverses processors in sequence via Apply().
4. Aggregation: Processed metrics are passed to Aggregator.Add(). At flush time, Push() emits aggregated data.
5. Writing: Final metrics are batched and sent via Output.Write().

Practical Implementation Examples

Minimal Input Plugin

This custom input generates a static metric:

package myinputs

import (
    "github.com/influxdata/telegraf"
    "github.com/influxdata/telegraf/plugins/inputs"
)

type MyCPU struct{}

func (c *MyCPU) Description() string { return "Collects a fake CPU metric" }
func (c *MyCPU) SampleConfig() string { return "" }

func (c *MyCPU) Gather(acc telegraf.Accumulator) error {
    acc.AddFields("mycpu", 
        map[string]interface{}{"value": 42}, 
        map[string]string{"host": "localhost"})
    return nil
}

func init() {
    inputs.Add("mycpu", func() telegraf.Input { return &MyCPU{} })
}

Tag Rename Processor

A processor that renames tags:

type Rename struct {
    From string `toml:"from"`
    To   string `toml:"to"`
}

func (p *Rename) Apply(in telegraf.Metric, out telegraf.Plugin) error {
    if v, ok := in.GetTag(p.From); ok {
        in.AddTag(p.To, v)
        in.RemoveTag(p.From)
    }
    out.AddMetric(in)
    return nil
}

func init() {
    processors.Add("rename", func() telegraf.Processor { return &Rename{} })
}

Min/Max Aggregator

Tracks statistical extremes over the flush window:

type MinMax struct {
    min, max float64
    seen     bool
}

func (a *MinMax) Add(in telegraf.Metric) {
    v, ok := in.GetField("value").(float64)
    if !ok { return }
    
    if !a.seen {
        a.min, a.max = v, v
        a.seen = true
        return
    }
    if v < a.min { a.min = v }
    if v > a.max { a.max = v }
}

func (a *MinMax) Push(acc telegraf.Accumulator) {
    if a.seen {
        acc.AddFields("minmax", 
            map[string]interface{}{"min": a.min, "max": a.max}, 
            nil)
    }
}

func (a *MinMax) Reset() { a.seen = false }

func init() {
    aggregators.Add("minmax", func() telegraf.Aggregator { return &MinMax{} })
}

Summary

• Telegraf's plugin architecture consists of four interface types—Input, Processor, Aggregator, and Output—each defined in the root package or plugins/ subdirectories.
• Plugins register themselves via init() functions calling package-specific Add methods (e.g., inputs.Add), populating global maps used by the configuration loader in [config/plugin_selector.go](https://github.com/influxdata/telegraf/blob/master/config/plugin_selector.go).
• Data flows unidirectionally: Input → Processor (sequential) → Aggregator (windowed) → Output.
• The RunningAggregator wrapper ensures aggregator plugins are single-threaded, while processors are expected to be stateless and handle metrics individually via Apply().
• All plugins must satisfy the PluginDescriber interface to provide metadata for TOML validation and documentation generation.

Frequently Asked Questions

What is the execution order of plugins in Telegraf's architecture?

Telegraf processes metrics in a strict pipeline: Inputs gather data first, then metrics pass through Processors in the order they appear in the TOML configuration. Next, Aggregators collect metrics over their configured periods, emitting rolled-up values. Finally, Outputs write the resulting metrics to external destinations. This sequence is hardcoded in the agent loop within [cmd/telegraf/telegraf.go](https://github.com/influxdata/telegraf/blob/master/cmd/telegraf/telegraf.go).

How do I register a custom plugin in Telegraf?

Create a Go file in the appropriate plugins/ subdirectory (e.g., plugins/inputs/myplugin/) and implement the required interface. In an init() function, call the registration helper—such as inputs.Add("myplugin", func() telegraf.Input { return &MyPlugin{} }). Telegraf automatically discovers plugin packages compiled into the binary and makes them available for use in the configuration file.

Are Telegraf processor plugins thread-safe?

Yes, but with specific constraints. Processor plugins implement the Apply method and are expected to be stateless; they receive metrics one at a time and should not maintain mutable state between calls. The core engine ensures that Apply is called sequentially for each metric in a single pipeline, though separate pipelines may run concurrently. In contrast, Aggregator plugins are explicitly protected by the RunningAggregator wrapper, which serializes calls to Add, Push, and Reset.

What is the difference between a processor and an aggregator in Telegraf?

Processors transform individual metrics in real-time as they flow through the pipeline—modifying tags, fields, or dropping metrics entirely. Aggregators collect multiple metrics over a time window (e.g., 30 seconds), compute statistical summaries like averages or percentiles, and emit new aggregate metrics at flush intervals. Processors handle one-to-one or one-to-many metric transformations, while aggregators implement many-to-one reduction logic using the Add/Push/Reset lifecycle defined in [aggregator.go](https://github.com/influxdata/telegraf/blob/master/aggregator.go).

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}