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

> Explore Telegraf's plugin architecture. Understand how Inputs, Processors, Aggregators, and Outputs work together in a modular data pipeline. Learn how Telegraf processes metrics efficiently.

- Repository: [InfluxData/telegraf](https://github.com/influxdata/telegraf)
- Tags: internals
- Published: 2026-05-14

---

**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/main/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/main/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/main/aggregator.go)](https://github.com/influxdata/telegraf/blob/master/aggregator.go) |
| **Output** | `Write([]Metric) error` | [[`output.go`](https://github.com/influxdata/telegraf/blob/main/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/main/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/main/input.go)](https://github.com/influxdata/telegraf/blob/master/input.go), the **Input** interface defines:

```go
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/main/plugins/inputs/apache/apache.go)](https://github.com/influxdata/telegraf/blob/master/plugins/inputs/apache/apache.go):

```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/main/plugins/processors/all/processors.go)](https://github.com/influxdata/telegraf/blob/master/plugins/processors/all/processors.go), the **Processor** interface requires:

```go
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

```go
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/main/aggregator.go)](https://github.com/influxdata/telegraf/blob/master/aggregator.go) specifies three lifecycle hooks:

```go
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:

```go
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/main/output.go)](https://github.com/influxdata/telegraf/blob/master/output.go) defines connection-oriented methods:

```go
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/main/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/main/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:

```go
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:

```go
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:

```go
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/main/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/main/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/main/aggregator.go)](https://github.com/influxdata/telegraf/blob/master/aggregator.go).