# How to Write Custom Processor Plugins for Telegraf: A Complete Developer Guide

> Learn how to write custom processor plugins for Telegraf. Implement interfaces, register plugins, and configure settings to extend Telegraf's data processing capabilities.

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

---

**To write custom processor plugins for Telegraf, implement the `telegraf.Processor` interface for synchronous batch processing or the `telegraf.StreamingProcessor` interface for asynchronous workflows, register your plugin using `processors.Add()` or `processors.AddStreaming()` in an `init()` function, and embed a [`sample.conf`](https://github.com/influxdata/telegraf/blob/main/sample.conf) file for configuration discovery.**

Telegraf’s processor plugin architecture enables you to transform, filter, and enrich metrics as they flow between inputs and outputs. Writing custom processor plugins for Telegraf involves creating a Go package that conforms to interfaces defined in [`processor.go`](https://github.com/influxdata/telegraf/blob/main/processor.go) and hooks into the global registry system. This guide explains the exact development patterns used in the influxdata/telegraf repository, including interface implementations, registration mechanics, and required file structures.

## Core Interfaces: Processor vs. StreamingProcessor

Telegraf defines two distinct processing models in [[`processor.go`](https://github.com/influxdata/telegraf/blob/main/processor.go)](https://github.com/influxdata/telegraf/blob/master/processor.go). Both interfaces embed `PluginDescriber`, which automatically provides `SampleConfig()` and `Description()` methods based on your struct tags.

**`telegraf.Processor`** (Synchronous)
- **Method signature**: `Apply(metrics ...telegraf.Metric) []telegraf.Metric`
- **Use case**: Stateless transformations that operate on complete batches of metrics
- **Execution**: Blocking; the agent waits for the entire batch to process before continuing

**`telegraf.StreamingProcessor`** (Asynchronous)
- **Method signatures**: `Start(acc telegraf.Accumulator) error`, `Add(m telegraf.Metric, acc telegraf.Accumulator) error`, `Stop()`
- **Use case**: Stateful operations, background workers, or tasks requiring goroutines (e.g., DNS lookups, external API calls)
- **Execution**: Non-blocking; metrics flow through individually

## Registering Your Plugin with the Global Registry

All processor plugins must register themselves in the global `processors` map located in [[`plugins/processors/registry.go`](https://github.com/influxdata/telegraf/blob/main/plugins/processors/registry.go)](https://github.com/influxdata/telegraf/blob/master/plugins/processors/registry.go). Registration occurs inside an `init()` function to ensure discovery at startup.

- **Legacy processors**: Call `processors.Add(name, creator)` where `creator` returns a `telegraf.Processor`
- **Streaming processors**: Call `processors.AddStreaming(name, creator)` where `creator` returns a `telegraf.StreamingProcessor`

The plugin name passed to these functions must exactly match the TOML configuration stanza (e.g., `processors.Add("myprocessor", ...)` corresponds to `[[processors.myprocessor]]` in [`telegraf.conf`](https://github.com/influxdata/telegraf/blob/main/telegraf.conf)).

## Minimal Processor Implementation

Below is a complete skeleton for a synchronous processor. This pattern mirrors the reference implementation in [[`plugins/processors/unpivot/unpivot.go`](https://github.com/influxdata/telegraf/blob/main/plugins/processors/unpivot/unpivot.go)](https://github.com/influxdata/telegraf/blob/master/plugins/processors/unpivot/unpivot.go).

```go
package myprocessor

import (
	_ "embed"
	"fmt"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/plugins/processors"
)

//go:embed sample.conf
var sampleConfig string

// MyProcessor demonstrates the synchronous Processor interface.
type MyProcessor struct {
	Prefix string `toml:"prefix"` // Maps to "prefix" key in TOML config
}

// Init validates configuration after TOML parsing but before metric processing.
func (p *MyProcessor) Init() error {
	if p.Prefix == "" {
		return fmt.Errorf("prefix must be set")
	}
	return nil
}

// SampleConfig returns the embedded configuration example.
func (p *MyProcessor) SampleConfig() string {
	return sampleConfig
}

// Apply implements the core processing logic.
// It receives a batch of metrics and must return the processed slice.
func (p *MyProcessor) Apply(metrics ...telegraf.Metric) []telegraf.Metric {
	out := make([]telegraf.Metric, 0, len(metrics))
	for _, m := range metrics {
		// Example transformation: prepend prefix to metric name
		newName := fmt.Sprintf("%s_%s", p.Prefix, m.Name())
		m.SetName(newName)
		out = append(out, m)
	}
	return out
}

// init registers the plugin at import time.
func init() {
	processors.Add("myprocessor", func() telegraf.Processor {
		return &MyProcessor{}
	})
}

```

## Implementing Streaming Processors for Asynchronous Work

For operations requiring background goroutines or external I/O, implement `telegraf.StreamingProcessor`. The `reverse_dns` processor in the Telegraf codebase provides a complete production example of this pattern.

```go
package mystream

import (
	"sync"

	"github.com/influxdata/telegraf"
	"github.com/influxdata/telegraf/plugins/processors"
)

// MyStreamingProcessor handles metrics asynchronously.
type MyStreamingProcessor struct {
	acc telegraf.Accumulator
	wg  sync.WaitGroup
}

// Start initializes resources when the Telegraf agent starts.
func (p *MyStreamingProcessor) Start(acc telegraf.Accumulator) error {
	p.acc = acc
	// Initialize workers, open connections, etc.
	return nil
}

// Add receives individual metrics for asynchronous processing.
func (p *MyStreamingProcessor) Add(m telegraf.Metric, acc telegraf.Accumulator) error {
	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		// Perform async work (e.g., DNS lookup, API enrichment)
		// Then write the result:
		// acc.AddMetric(transformedMetric)
	}()
	return nil
}

// Stop waits for all background goroutines to complete during shutdown.
func (p *MyStreamingProcessor) Stop() {
	p.wg.Wait()
}

// init registers as a streaming processor.
func init() {
	processors.AddStreaming("mystream", func() telegraf.StreamingProcessor {
		return &MyStreamingProcessor{}
	})
}

```

## File Structure and Required Conventions

Organize your custom processor according to Telegraf’s standard directory layout to ensure compatibility with the build system and documentation generators.

**Directory Structure**
- `plugins/processors/<myplugin>/` — Source code directory containing Go files and [`sample.conf`](https://github.com/influxdata/telegraf/blob/main/sample.conf)
- `plugins/processors/all/<myplugin>.go` — Import file with build tags ( ensures `init()` runs)
- [`docs/PROCESSORS.md`](https://github.com/influxdata/telegraf/blob/main/docs/PROCESSORS.md) — Developer documentation (follow existing style guidelines)

**Required Files**
- **`<plugin>.go`**: Main implementation with struct definition and interface methods
- **[`sample.conf`](https://github.com/influxdata/telegraf/blob/main/sample.conf)**: TOML configuration example embedded via `//go:embed`
- **[`README.md`](https://github.com/influxdata/telegraf/blob/main/README.md)**: User-facing documentation explaining configuration options

**Naming Conventions**
- The TOML stanza name must match the registration string exactly
- Configuration struct fields use `toml:"field_name"` tags
- Call `metric.Drop()` to filter out metrics; otherwise ensure metrics are forwarded via the returned slice or `acc.AddMetric()`

After placing your code in the correct directory and ensuring it is imported in `plugins/processors/all/*.go`, compile Telegraf and reference your processor in [`telegraf.conf`](https://github.com/influxdata/telegraf/blob/main/telegraf.conf):

```toml
[[processors.myprocessor]]
  prefix = "production"

```

## Summary

- **Implement** either `telegraf.Processor` for batch operations or `telegraf.StreamingProcessor` for async workflows, both defined in [`processor.go`](https://github.com/influxdata/telegraf/blob/main/processor.go)
- **Register** your plugin using `processors.Add()` or `processors.AddStreaming()` inside an `init()` function in [`plugins/processors/registry.go`](https://github.com/influxdata/telegraf/blob/main/plugins/processors/registry.go)
- **Embed** a [`sample.conf`](https://github.com/influxdata/telegraf/blob/main/sample.conf) file and expose it via `SampleConfig()` to enable `telegraf --input-list` discovery
- **Place** source files in `plugins/processors/<name>/` and ensure build tags in `plugins/processors/all/*.go` import your package

## Frequently Asked Questions

### What is the difference between Apply() and Add() in Telegraf processors?

**`Apply()`** is the method signature for synchronous processors that receive complete batches of metrics (as implemented in the `Processor` interface), while **`Add()`** is used by streaming processors to handle metrics one-by-one asynchronously. According to the Telegraf source code in [`processor.go`](https://github.com/influxdata/telegraf/blob/main/processor.go), `Apply()` must return the full processed slice, whereas `Add()` writes results directly to the provided `telegraf.Accumulator` and can spawn background goroutines.

### How do I drop metrics that should not be forwarded to outputs?

Call **`metric.Drop()`** on any metric you want to filter out. In the `Apply()` method, simply exclude the metric from the returned slice. For streaming processors using `Add()`, do not call `acc.AddMetric()` for metrics you wish to discard. Dropping metrics at the processor stage prevents them from reaching aggregators or outputs.

### Why is my custom processor not appearing in `telegraf --input-list`?

Your plugin must be **imported** into the Telegraf binary. Ensure your package is imported in a file under `plugins/processors/all/` (which uses build tags to orchestrate imports), and verify your `init()` function correctly calls `processors.Add()` or `processors.AddStreaming()` with the exact name matching your configuration stanza. Also confirm that your `SampleConfig()` method returns non-empty content embedded from [`sample.conf`](https://github.com/influxdata/telegraf/blob/main/sample.conf).

### Can I maintain state between metrics in a processor?

Yes, but only when using the **`telegraf.StreamingProcessor`** interface. Synchronous processors implementing `Apply()` should remain stateless because they receive independent batches. Streaming processors maintain state through struct fields and can use goroutines managed via `Start()` and `Stop()` lifecycle hooks, as demonstrated in the `reverse_dns` implementation where DNS cache state persists across multiple `Add()` calls.