Refactor handling of MinMax functionality into RunningAggregator

allows for easier addition of a sliding window at a later time.

Also makes `period` be a generic argument for all aggregator plugins.

internal/models/running_aggregator.go

@@ -7,8 +7,21 @@ import (
 )
 
 type RunningAggregator struct {
-	Aggregator telegraf.Aggregator
-	Config     *AggregatorConfig
+	a      telegraf.Aggregator
+	Config *AggregatorConfig
+
+	metrics chan telegraf.Metric
+}
+
+func NewRunningAggregator(
+	a telegraf.Aggregator,
+	conf *AggregatorConfig,
+) *RunningAggregator {
+	return &RunningAggregator{
+		a:       a,
+		Config:  conf,
+		metrics: make(chan telegraf.Metric, 100),
+	}
 }
 
 // AggregatorConfig containing configuration parameters for the running
@@ -22,6 +35,9 @@ type AggregatorConfig struct {
 	MeasurementSuffix string
 	Tags              map[string]string
 	Filter            Filter
+
+	Period time.Duration
+	Delay  time.Duration
 }
 
 func (r *RunningAggregator) Name() string {
@@ -56,10 +72,10 @@ func (r *RunningAggregator) MakeMetric(
 	return m
 }
 
-// Apply applies the given metric to the aggregator.
+// Add applies the given metric to the aggregator.
 // Before applying to the plugin, it will run any defined filters on the metric.
 // Apply returns true if the original metric should be dropped.
-func (r *RunningAggregator) Apply(in telegraf.Metric) bool {
+func (r *RunningAggregator) Add(in telegraf.Metric) bool {
 	if r.Config.Filter.IsActive() {
 		// check if the aggregator should apply this metric
 		name := in.Name()
@@ -74,6 +90,49 @@ func (r *RunningAggregator) Apply(in telegraf.Metric) bool {
 		in, _ = telegraf.NewMetric(name, tags, fields, t)
 	}
 
-	r.Aggregator.Apply(in)
+	r.metrics <- in
 	return r.Config.DropOriginal
 }
+func (r *RunningAggregator) add(in telegraf.Metric) {
+	r.a.Add(in)
+}
+
+func (r *RunningAggregator) push(acc telegraf.Accumulator) {
+	r.a.Push(acc)
+}
+
+func (r *RunningAggregator) reset() {
+	r.a.Reset()
+}
+
+func (r *RunningAggregator) Run(
+	acc telegraf.Accumulator,
+	shutdown chan struct{},
+) {
+	if r.Config.Delay == 0 {
+		r.Config.Delay = time.Millisecond * 100
+	}
+	if r.Config.Period == 0 {
+		r.Config.Period = time.Second * 30
+	}
+
+	time.Sleep(r.Config.Delay)
+	periodT := time.NewTicker(r.Config.Period)
+	defer periodT.Stop()
+
+	for {
+		select {
+		case <-shutdown:
+			if len(r.metrics) > 0 {
+				// wait until metrics are flushed before exiting
+				continue
+			}
+			return
+		case m := <-r.metrics:
+			r.add(m)
+		case <-periodT.C:
+			r.push(acc)
+			r.reset()
+		}
+	}
+}

internal/models/running_aggregator_test.go

@@ -2,26 +2,28 @@ package models
 
 import (
 	"fmt"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/testutil"
 
 	"github.com/stretchr/testify/assert"
 )
 
-func TestApply(t *testing.T) {
+func TestAdd(t *testing.T) {
 	a := &TestAggregator{}
-	ra := RunningAggregator{
-		Config: &AggregatorConfig{
-			Name: "TestRunningAggregator",
-			Filter: Filter{
-				NamePass: []string{"*"},
-			},
+	ra := NewRunningAggregator(a, &AggregatorConfig{
+		Name: "TestRunningAggregator",
+		Filter: Filter{
+			NamePass: []string{"*"},
 		},
-		Aggregator: a,
-	}
+	})
 	assert.NoError(t, ra.Config.Filter.Compile())
+	acc := testutil.Accumulator{}
+	go ra.Run(&acc, make(chan struct{}))
 
 	m := ra.MakeMetric(
 		"RITest",
@@ -30,21 +32,64 @@ func TestApply(t *testing.T) {
 		telegraf.Untyped,
 		time.Now(),
 	)
-	assert.False(t, ra.Apply(m))
-	assert.Equal(t, int64(101), a.sum)
+	assert.False(t, ra.Add(m))
+
+	for {
+		if atomic.LoadInt64(&a.sum) > 0 {
+			break
+		}
+	}
+	assert.Equal(t, int64(101), atomic.LoadInt64(&a.sum))
 }
 
-func TestApplyDropOriginal(t *testing.T) {
-	ra := RunningAggregator{
-		Config: &AggregatorConfig{
-			Name: "TestRunningAggregator",
-			Filter: Filter{
-				NamePass: []string{"RI*"},
-			},
-			DropOriginal: true,
+func TestAddAndPushOnePeriod(t *testing.T) {
+	a := &TestAggregator{}
+	ra := NewRunningAggregator(a, &AggregatorConfig{
+		Name: "TestRunningAggregator",
+		Filter: Filter{
+			NamePass: []string{"*"},
 		},
-		Aggregator: &TestAggregator{},
+		Period: time.Millisecond * 500,
+	})
+	assert.NoError(t, ra.Config.Filter.Compile())
+	acc := testutil.Accumulator{}
+	shutdown := make(chan struct{})
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		ra.Run(&acc, shutdown)
+	}()
+
+	m := ra.MakeMetric(
+		"RITest",
+		map[string]interface{}{"value": int(101)},
+		map[string]string{},
+		telegraf.Untyped,
+		time.Now(),
+	)
+	assert.False(t, ra.Add(m))
+
+	for {
+		if acc.NMetrics() > 0 {
+			break
+		}
 	}
+	acc.AssertContainsFields(t, "TestMetric", map[string]interface{}{"sum": int64(101)})
+
+	close(shutdown)
+	wg.Wait()
+}
+
+func TestAddDropOriginal(t *testing.T) {
+	ra := NewRunningAggregator(&TestAggregator{}, &AggregatorConfig{
+		Name: "TestRunningAggregator",
+		Filter: Filter{
+			NamePass: []string{"RI*"},
+		},
+		DropOriginal: true,
+	})
 	assert.NoError(t, ra.Config.Filter.Compile())
 
 	m := ra.MakeMetric(
@@ -54,9 +99,9 @@ func TestApplyDropOriginal(t *testing.T) {
 		telegraf.Untyped,
 		time.Now(),
 	)
-	assert.True(t, ra.Apply(m))
+	assert.True(t, ra.Add(m))
 
-	// this metric name doesn't match the filter, so Apply will return false
+	// this metric name doesn't match the filter, so Add will return false
 	m2 := ra.MakeMetric(
 		"foobar",
 		map[string]interface{}{"value": int(101)},
@@ -64,17 +109,15 @@ func TestApplyDropOriginal(t *testing.T) {
 		telegraf.Untyped,
 		time.Now(),
 	)
-	assert.False(t, ra.Apply(m2))
+	assert.False(t, ra.Add(m2))
 }
 
 // make an untyped, counter, & gauge metric
 func TestMakeMetricA(t *testing.T) {
 	now := time.Now()
-	ra := RunningAggregator{
-		Config: &AggregatorConfig{
-			Name: "TestRunningAggregator",
-		},
-	}
+	ra := NewRunningAggregator(&TestAggregator{}, &AggregatorConfig{
+		Name: "TestRunningAggregator",
+	})
 	assert.Equal(t, "aggregators.TestRunningAggregator", ra.Name())
 
 	m := ra.MakeMetric(
@@ -136,15 +179,21 @@ type TestAggregator struct {
 	sum int64
 }
 
-func (t *TestAggregator) Description() string                  { return "" }
-func (t *TestAggregator) SampleConfig() string                 { return "" }
-func (t *TestAggregator) Start(acc telegraf.Accumulator) error { return nil }
-func (t *TestAggregator) Stop()                                {}
+func (t *TestAggregator) Description() string  { return "" }
+func (t *TestAggregator) SampleConfig() string { return "" }
+func (t *TestAggregator) Reset()               {}
 
-func (t *TestAggregator) Apply(in telegraf.Metric) {
+func (t *TestAggregator) Push(acc telegraf.Accumulator) {
+	acc.AddFields("TestMetric",
+		map[string]interface{}{"sum": t.sum},
+		map[string]string{},
+	)
+}
+
+func (t *TestAggregator) Add(in telegraf.Metric) {
 	for _, v := range in.Fields() {
 		if vi, ok := v.(int64); ok {
-			t.sum += vi
+			atomic.AddInt64(&t.sum, vi)
 		}
 	}
 }