feat(aggregators.final): Add option to disable appending _final

plugins/aggregators/final/README.md

@@ -32,6 +32,9 @@ See the [CONFIGURATION.md][CONFIGURATION.md] for more details.
   ## aggregator and will not get sent to the output plugins.
   # drop_original = false
 
+  ## If true, _final is added to every field name
+  # rename_fields = true
+
   ## The time that a series is not updated until considering it final. Ignored
   ## when output_strategy is "periodic".
   # series_timeout = "5m"

plugins/aggregators/final/final.go

@@ -17,6 +17,7 @@ var sampleConfig string
 type Final struct {
 	OutputStrategy string          `toml:"output_strategy"`
 	SeriesTimeout  config.Duration `toml:"series_timeout"`
+	RenameFields   bool            `toml:"rename_fields"`
 
 	// The last metric for all series which are active
 	metricCache map[uint64]telegraf.Metric
@@ -25,6 +26,7 @@ type Final struct {
 func NewFinal() *Final {
 	return &Final{
 		SeriesTimeout: config.Duration(5 * time.Minute),
+		RenameFields:  true,
 	}
 }
 
@@ -64,10 +66,16 @@ func (m *Final) Push(acc telegraf.Accumulator) {
 			// younger than that. So skip the output for this period.
 			continue
 		}
-		fields := map[string]interface{}{}
-		for _, field := range metric.FieldList() {
-			fields[field.Key+"_final"] = field.Value
+		var fields map[string]any
+		if m.RenameFields {
+			fields = map[string]interface{}{}
+			for _, field := range metric.FieldList() {
+				fields[field.Key+"_final"] = field.Value
+			}
+		} else {
+			fields = metric.Fields()
 		}
+
 		acc.AddFields(metric.Name(), fields, metric.Tags(), metric.Time())
 		delete(m.metricCache, id)
 	}

plugins/aggregators/final/final_test.go

@@ -159,6 +159,7 @@ func TestOutputStrategyTimeout(t *testing.T) {
 	final := &Final{
 		OutputStrategy: "timeout",
 		SeriesTimeout:  config.Duration(30 * time.Second),
+		RenameFields:   true,
 	}
 	require.NoError(t, final.Init())
 
@@ -215,6 +216,7 @@ func TestOutputStrategyPeriodic(t *testing.T) {
 	final := &Final{
 		OutputStrategy: "periodic",
 		SeriesTimeout:  config.Duration(30 * time.Second),
+		RenameFields:   true,
 	}
 	require.NoError(t, final.Init())
 
@@ -266,9 +268,42 @@ func TestOutputStrategyPeriodic(t *testing.T) {
 		metric.New(
 			"m",
 			tags,
-			map[string]interface{}{"a_final": int64(4)},
+			map[string]interface{}{
+				"a_final": 4,
+			},
 			now.Add(time.Second*-20),
 		),
 	}
 	testutil.RequireMetricsEqual(t, expected, acc.GetTelegrafMetrics(), testutil.SortMetrics())
 }
+
+func TestRenameFieldsFalse(t *testing.T) {
+	final := &Final{
+		OutputStrategy: "periodic",
+		SeriesTimeout:  config.Duration(30 * time.Second),
+		RenameFields:   false,
+	}
+
+	require.NoError(t, final.Init())
+
+	now := time.Now()
+	tags := map[string]string{"foo": "bar"}
+	m1 := metric.New("m",
+		tags,
+		map[string]any{"a": 3},
+		now.Add(time.Second*-90))
+
+	var acc testutil.Accumulator
+	final.Add(m1)
+	final.Push(&acc)
+	expected := []telegraf.Metric{
+		metric.New(
+			"m",
+			tags,
+			map[string]any{"a": 3},
+			now.Add(time.Second*-90),
+		),
+	}
+
+	testutil.RequireMetricsEqual(t, expected, acc.GetTelegrafMetrics(), testutil.SortMetrics())
+}

plugins/aggregators/final/sample.conf

@@ -7,6 +7,9 @@
   ## aggregator and will not get sent to the output plugins.
   # drop_original = false
 
+  ## If true, _final is added to every field name
+  # rename_fields = true
+
   ## The time that a series is not updated until considering it final. Ignored
   ## when output_strategy is "periodic".
   # series_timeout = "5m"