aggregators: allow historical data in aggregators

docs/AGGREGATORS.md

@@ -50,6 +50,8 @@ var sampleConfig = `
   ## If true drop_original will drop the original metrics and
   ## only send aggregates.
   drop_original = false
+  ## If true, metrics that fall outside the period won't be ignored.
+  # allow_historical = false
 `
 
 func (m *Min) Init() error {

etc/telegraf.conf

@@ -1493,6 +1493,8 @@
 #   ## If true, the original metric will be dropped by the
 #   ## aggregator and will not get sent to the output plugins.
 #   drop_original = false
+#   ## If true, metrics that fall outside the period won't be ignored.
+#   allow_historical = false
 #
 #   ## Configures which basic stats to push as fields
 #   # stats = ["count", "min", "max", "mean", "stdev", "s2", "sum"]
@@ -1505,6 +1507,8 @@
 #   ## If true, the original metric will be dropped by the
 #   ## aggregator and will not get sent to the output plugins.
 #   drop_original = false
+#   ## If true, metrics that fall outside the period won't be ignored.
+#   allow_historical = false
 #
 #   ## The time that a series is not updated until considering it final.
 #   series_timeout = "5m"
@@ -1519,6 +1523,9 @@
 #   ## aggregator and will not get sent to the output plugins.
 #   drop_original = false
 #
+#   ## If true, metrics that fall outside the period won't be ignored.
+#   allow_historical = false
+#
 #   ## If true, the histogram will be reset on flush instead
 #   ## of accumulating the results.
 #   reset = false
@@ -1548,6 +1555,8 @@
 #   ## If true, the original metric will be dropped by the
 #   ## aggregator and will not get sent to the output plugins.
 #   drop_original = false
+#   ## If true, metrics that fall outside the period won't be ignored.
+#   allow_historical = false
 
 
 # # Count the occurrence of values in fields.
@@ -1558,6 +1567,8 @@
 #   ## If true, the original metric will be dropped by the
 #   ## aggregator and will not get sent to the output plugins.
 #   drop_original = false
+#   ## If true, metrics that fall outside the period won't be ignored.
+#   allow_historical = false
 #   ## The fields for which the values will be counted
 #   fields = []
 

internal/config/config.go

@@ -1065,6 +1065,18 @@ func buildAggregator(name string, tbl *ast.Table) (*models.AggregatorConfig, err
 		}
 	}
 
+	if node, ok := tbl.Fields["allow_historical"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if b, ok := kv.Value.(*ast.Boolean); ok {
+				var err error
+				conf.AllowHistorical, err = strconv.ParseBool(b.Value)
+				if err != nil {
+					log.Printf("Error parsing boolean value for %s: %s\n", name, err)
+				}
+			}
+		}
+	}
+
 	if node, ok := tbl.Fields["name_prefix"]; ok {
 		if kv, ok := node.(*ast.KeyValue); ok {
 			if str, ok := kv.Value.(*ast.String); ok {
@@ -1101,6 +1113,7 @@ func buildAggregator(name string, tbl *ast.Table) (*models.AggregatorConfig, err
 	delete(tbl.Fields, "period")
 	delete(tbl.Fields, "delay")
 	delete(tbl.Fields, "drop_original")
+	delete(tbl.Fields, "allow_historical")
 	delete(tbl.Fields, "name_prefix")
 	delete(tbl.Fields, "name_suffix")
 	delete(tbl.Fields, "name_override")

internal/models/running_aggregator.go

@@ -55,10 +55,11 @@ func NewRunningAggregator(
 
 // AggregatorConfig is the common config for all aggregators.
 type AggregatorConfig struct {
-	Name         string
-	DropOriginal bool
-	Period       time.Duration
-	Delay        time.Duration
+	Name            string
+	DropOriginal    bool
+	AllowHistorical bool
+	Period          time.Duration
+	Delay           time.Duration
 
 	NameOverride      string
 	MeasurementPrefix string
@@ -135,11 +136,13 @@ func (r *RunningAggregator) Add(m telegraf.Metric) bool {
 	r.Lock()
 	defer r.Unlock()
 
-	if m.Time().Before(r.periodStart) || m.Time().After(r.periodEnd.Add(r.Config.Delay)) {
-		log.Printf("D! [%s] metric is outside aggregation window; discarding. %s: m: %s e: %s",
-			r.Name(), m.Time(), r.periodStart, r.periodEnd)
-		r.MetricsDropped.Incr(1)
-		return r.Config.DropOriginal
+	if !r.Config.AllowHistorical {
+		if m.Time().Before(r.periodStart) || m.Time().After(r.periodEnd.Add(r.Config.Delay)) {
+			log.Printf("D! [%s] metric is outside aggregation window; discarding. %s: m: %s e: %s",
+				r.Name(), m.Time(), r.periodStart, r.periodEnd)
+			r.MetricsDropped.Incr(1)
+			return r.Config.DropOriginal
+		}
 	}
 
 	r.Aggregator.Add(m)

internal/models/running_aggregator_test.go

@@ -89,6 +89,57 @@ func TestAddMetricsOutsideCurrentPeriod(t *testing.T) {
 	require.Equal(t, int64(101), acc.Metrics[0].Fields["sum"])
 }
 
+func TestAllowHistorical(t *testing.T) {
+	a := &TestAggregator{}
+	ra := NewRunningAggregator(a, &AggregatorConfig{
+		Name: "TestRunningAggregator",
+		Filter: Filter{
+			NamePass: []string{"*"},
+		},
+		Period:          time.Millisecond * 500,
+		AllowHistorical: true,
+	})
+	require.NoError(t, ra.Config.Filter.Compile())
+	acc := testutil.Accumulator{}
+	now := time.Now()
+	ra.UpdateWindow(now, now.Add(ra.Config.Period))
+
+	m := testutil.MustMetric("RITest",
+		map[string]string{},
+		map[string]interface{}{
+			"value": int64(101),
+		},
+		now.Add(-time.Hour),
+		telegraf.Untyped,
+	)
+	require.False(t, ra.Add(m))
+
+	// metric after current period
+	m = testutil.MustMetric("RITest",
+		map[string]string{},
+		map[string]interface{}{
+			"value": int64(101),
+		},
+		now.Add(time.Hour),
+		telegraf.Untyped,
+	)
+	require.False(t, ra.Add(m))
+
+	// "now" metric
+	m = testutil.MustMetric("RITest",
+		map[string]string{},
+		map[string]interface{}{
+			"value": int64(101),
+		},
+		time.Now().Add(time.Millisecond*50),
+		telegraf.Untyped)
+	require.False(t, ra.Add(m))
+
+	ra.Push(&acc)
+	require.Equal(t, 1, len(acc.Metrics))
+	require.Equal(t, int64(303), acc.Metrics[0].Fields["sum"])
+}
+
 func TestAddAndPushOnePeriod(t *testing.T) {
 	a := &TestAggregator{}
 	ra := NewRunningAggregator(a, &AggregatorConfig{

plugins/aggregators/basicstats/basicstats.go

@@ -52,6 +52,8 @@ var sampleConfig = `
   ## If true, the original metric will be dropped by the
   ## aggregator and will not get sent to the output plugins.
   drop_original = false
+  ## If true, metrics that fall outside the period won't be ignored.
+  # allow_historical = false
 
   ## Configures which basic stats to push as fields
   # stats = ["count", "min", "max", "mean", "stdev", "s2", "sum"]

plugins/aggregators/final/final.go

@@ -14,6 +14,8 @@ var sampleConfig = `
   ## If true, the original metric will be dropped by the
   ## aggregator and will not get sent to the output plugins.
   drop_original = false
+  ## If true, metrics that fall outside the period won't be ignored.
+  # allow_historical = false
 
   ## The time that a series is not updated until considering it final.
   series_timeout = "5m"

plugins/aggregators/histogram/histogram.go

@@ -73,6 +73,9 @@ var sampleConfig = `
   ## aggregator and will not get sent to the output plugins.
   drop_original = false
 
+  ## If true, metrics that fall outside the period won't be ignored.
+  # allow_historical = false
+
   ## If true, the histogram will be reset on flush instead
   ## of accumulating the results.
   reset = false

plugins/aggregators/minmax/minmax.go

@@ -33,6 +33,8 @@ var sampleConfig = `
   ## If true, the original metric will be dropped by the
   ## aggregator and will not get sent to the output plugins.
   drop_original = false
+  ## If true, metrics that fall outside the period won't be ignored.
+  # allow_historical = false
 `
 
 func (m *MinMax) SampleConfig() string {

plugins/aggregators/valuecounter/valuecounter.go

@@ -36,6 +36,8 @@ var sampleConfig = `
   drop_original = false
   ## The fields for which the values will be counted
   fields = []
+  ## If true, metrics that fall outside the period won't be ignored.
+  # allow_historical = false
 `
 
 // SampleConfig generates a sample config for the ValueCounter plugin