feat(processors.batch): Add batch processor

plugins/processors/all/batch.go

@@ -0,0 +1,5 @@
+//go:build !custom || processors || processors.batch
+
+package all
+
+import _ "github.com/influxdata/telegraf/plugins/processors/batch" // register plugin

plugins/processors/batch/README.md

@@ -0,0 +1,61 @@
+# Batch Processor Plugin
+
+This processor groups metrics into batches by adding a batch tag. This is 
+useful for parallel processing of metrics where downstream processors, 
+aggregators or outputs can then select a batch using `tagpass` or `metricpass`.
+
+## Global configuration options <!-- @/docs/includes/plugin_config.md -->
+
+In addition to the plugin-specific configuration settings, plugins support
+additional global and plugin configuration settings. These settings are used to
+modify metrics, tags, and field or create aliases and configure ordering, etc.
+See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
+
+[CONFIGURATION.md]: ../../../docs/CONFIGURATION.md#plugins
+
+## Configuration
+
+```toml @sample.conf
+## Batch metrics into separate batches by adding a tag indicating the batch index.
+## Can be used to route batches of metrics to different outputs
+## to parallelize writing of metrics to an output
+## Metrics are distributed across batches using the round-robin scheme.
+[[processors.batch]]
+  ## The name of the tag to use for adding the batch index
+  batch_tag = "my_batch"
+
+  ## The number of batches to create
+  num_batches = 16
+
+  ## Do not assign metrics with an existing batch assignment to a
+  ## different batch. 
+  # skip_existing = false
+```
+
+## Example
+
+The example below uses these settings:
+
+```toml
+[[processors.batch]]
+  ## The tag key to use for batching
+  batch_tag = "batch"
+
+  ## The number of batches to create
+  num_batches = 3
+```
+
+```diff
+- temperature cpu=25
+- temperature cpu=50
+- temperature cpu=75
+- temperature cpu=25
+- temperature cpu=50
+- temperature cpu=75
++ temperature,batch=0 cpu=25
++ temperature,batch=1 cpu=50
++ temperature,batch=2 cpu=75
++ temperature,batch=0 cpu=25
++ temperature,batch=1 cpu=50
++ temperature,batch=2 cpu=75
+```

plugins/processors/batch/batch.go

@@ -0,0 +1,47 @@
+package batch
+
+import (
+	_ "embed"
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/processors"
+	"strconv"
+	"sync/atomic"
+)
+
+//go:embed sample.conf
+var sampleConfig string
+
+type Batch struct {
+	BatchTag     string `toml:"batch_tag"`
+	NumBatches   uint64 `toml:"num_batches"`
+	SkipExisting bool   `toml:"skip_existing"`
+
+	// the number of metrics that have been processed so far
+	count atomic.Uint64
+}
+
+func (*Batch) SampleConfig() string {
+	return sampleConfig
+}
+
+func (b *Batch) Apply(in ...telegraf.Metric) []telegraf.Metric {
+	out := make([]telegraf.Metric, 0, len(in))
+	for _, m := range in {
+		if b.SkipExisting && m.HasTag(b.BatchTag) {
+			continue
+		}
+
+		oldCount := b.count.Add(1) - 1
+		batchId := oldCount % b.NumBatches
+		m.AddTag(b.BatchTag, strconv.FormatUint(batchId, 10))
+		out = append(out, m)
+	}
+
+	return out
+}
+
+func init() {
+	processors.Add("batch", func() telegraf.Processor {
+		return &Batch{}
+	})
+}

plugins/processors/batch/batch_test.go

@@ -0,0 +1,69 @@
+package batch
+
+import (
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/testutil"
+	"github.com/stretchr/testify/require"
+	"testing"
+)
+
+const batchTag = "?internal_batch_idx"
+
+func MakeBatching(batches uint64) *Batch {
+	return &Batch{
+		BatchTag:   batchTag,
+		NumBatches: batches,
+	}
+}
+
+func MakeXMetrics(count int) []telegraf.Metric {
+	ms := make([]telegraf.Metric, 0, count)
+	for range count {
+		ms = append(ms, testutil.MockMetrics()...)
+	}
+
+	return ms
+}
+
+func requireMetricInBatch(t *testing.T, m telegraf.Metric, batch string) {
+	batchTagValue, ok := m.GetTag(batchTag)
+	require.True(t, ok)
+	require.Equal(t, batch, batchTagValue)
+}
+
+func Test_SingleMetricPutInBatch0(t *testing.T) {
+	b := MakeBatching(1)
+	m := testutil.MockMetricsWithValue(1)
+	expectedM := testutil.MockMetricsWithValue(1)
+	expectedM[0].AddTag(batchTag, "0")
+
+	res := b.Apply(m...)
+	testutil.RequireMetricsEqual(t, expectedM, res)
+}
+
+func Test_MetricsSmallerThanBatchSizeAreInDifferentBatches(t *testing.T) {
+	b := MakeBatching(3)
+	ms := MakeXMetrics(2)
+	res := b.Apply(ms...)
+	requireMetricInBatch(t, res[0], "0")
+	requireMetricInBatch(t, res[1], "1")
+}
+
+func Test_MetricsEqualToBatchSizeInDifferentBatches(t *testing.T) {
+	b := MakeBatching(3)
+	ms := MakeXMetrics(3)
+	res := b.Apply(ms...)
+	requireMetricInBatch(t, res[0], "0")
+	requireMetricInBatch(t, res[1], "1")
+	requireMetricInBatch(t, res[2], "2")
+}
+
+func Test_MetricsMoreThanBatchSizeInSameBatch(t *testing.T) {
+	b := MakeBatching(2)
+	ms := MakeXMetrics(3)
+	res := b.Apply(ms...)
+
+	requireMetricInBatch(t, res[0], "0")
+	requireMetricInBatch(t, res[1], "1")
+	requireMetricInBatch(t, res[2], "0")
+}

plugins/processors/batch/sample.conf

@@ -0,0 +1,14 @@
+## Batch metrics into separate batches by adding a tag indicating the batch index.
+## Can be used to route batches of metrics to different outputs
+## to parallelize writing of metrics to an output
+## Metrics are distributed across batches using the round-robin scheme.
+[[processors.batch]]
+  ## The name of the tag to use for adding the batch index
+  batch_tag = "my_batch"
+
+  ## The number of batches to create
+  num_batches = 16
+
+  ## Do not assign metrics with an existing batch assignment to a
+  ## different batch.
+  # skip_existing = false