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

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 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 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/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/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).

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/master/plugins/processors/unpivot/unpivot.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.

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
• plugins/processors/all/<myplugin>.go — Import file with build tags ( ensures init() runs)
• docs/PROCESSORS.md — Developer documentation (follow existing style guidelines)

Required Files

• <plugin>.go: Main implementation with struct definition and interface methods
• sample.conf: TOML configuration example embedded via //go:embed
• 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:

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

Summary

• Implement either telegraf.Processor for batch operations or telegraf.StreamingProcessor for async workflows, both defined in processor.go
• Register your plugin using processors.Add() or processors.AddStreaming() inside an init() function in plugins/processors/registry.go
• Embed a 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, 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.

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.

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