feat: migrate value parser to new style (#11407)

plugins/inputs/exec/exec_test.go

@@ -15,8 +15,8 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/influxdata/telegraf/plugins/parsers"
 	"github.com/influxdata/telegraf/plugins/parsers/json"
+	"github.com/influxdata/telegraf/plugins/parsers/value"
 	"github.com/influxdata/telegraf/testutil"
 )
 
@@ -139,11 +139,15 @@ func TestCommandError(t *testing.T) {
 }
 
 func TestExecCommandWithGlob(t *testing.T) {
-	parser, err := parsers.NewValueParser("metric", "string", "", nil)
-	require.NoError(t, err)
+	parser := value.Parser{
+		MetricName: "metric",
+		DataType:   "string",
+	}
+	require.NoError(t, parser.Init())
+
 	e := NewExec()
 	e.Commands = []string{"/bin/ech* metric_value"}
-	e.SetParser(parser)
+	e.SetParser(&parser)
 
 	var acc testutil.Accumulator
 	require.NoError(t, acc.GatherError(e.Gather))
@@ -155,12 +159,15 @@ func TestExecCommandWithGlob(t *testing.T) {
 }
 
 func TestExecCommandWithoutGlob(t *testing.T) {
-	parser, err := parsers.NewValueParser("metric", "string", "", nil)
-	require.NoError(t, err)
+	parser := value.Parser{
+		MetricName: "metric",
+		DataType:   "string",
+	}
+	require.NoError(t, parser.Init())
 
 	e := NewExec()
 	e.Commands = []string{"/bin/echo metric_value"}
-	e.SetParser(parser)
+	e.SetParser(&parser)
 
 	var acc testutil.Accumulator
 	require.NoError(t, acc.GatherError(e.Gather))
@@ -172,11 +179,14 @@ func TestExecCommandWithoutGlob(t *testing.T) {
 }
 
 func TestExecCommandWithoutGlobAndPath(t *testing.T) {
-	parser, err := parsers.NewValueParser("metric", "string", "", nil)
-	require.NoError(t, err)
+	parser := value.Parser{
+		MetricName: "metric",
+		DataType:   "string",
+	}
+	require.NoError(t, parser.Init())
 	e := NewExec()
 	e.Commands = []string{"echo metric_value"}
-	e.SetParser(parser)
+	e.SetParser(&parser)
 
 	var acc testutil.Accumulator
 	require.NoError(t, acc.GatherError(e.Gather))
@@ -188,12 +198,15 @@ func TestExecCommandWithoutGlobAndPath(t *testing.T) {
 }
 
 func TestExecCommandWithEnv(t *testing.T) {
-	parser, err := parsers.NewValueParser("metric", "string", "", nil)
-	require.NoError(t, err)
+	parser := value.Parser{
+		MetricName: "metric",
+		DataType:   "string",
+	}
+	require.NoError(t, parser.Init())
 	e := NewExec()
 	e.Commands = []string{"/bin/sh -c 'echo ${METRIC_NAME}'"}
 	e.Environment = []string{"METRIC_NAME=metric_value"}
-	e.SetParser(parser)
+	e.SetParser(&parser)
 
 	var acc testutil.Accumulator
 	require.NoError(t, acc.GatherError(e.Gather))

plugins/inputs/http/http_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/influxdata/telegraf/plugins/parsers"
 	"github.com/influxdata/telegraf/plugins/parsers/csv"
 	"github.com/influxdata/telegraf/plugins/parsers/json"
+	"github.com/influxdata/telegraf/plugins/parsers/value"
 	"github.com/influxdata/telegraf/testutil"
 )
 
@@ -374,7 +375,12 @@ func TestOAuthClientCredentialsGrant(t *testing.T) {
 			})
 
 			tt.plugin.SetParserFunc(func() (telegraf.Parser, error) {
-				return parsers.NewValueParser("metric", "string", "", nil)
+				p := &value.Parser{
+					MetricName: "metric",
+					DataType:   "string",
+				}
+				err := p.Init()
+				return p, err
 			})
 
 			err = tt.plugin.Init()

plugins/inputs/kafka_consumer/kafka_consumer_test.go

@@ -271,8 +271,12 @@ func (c *FakeConsumerGroupClaim) Messages() <-chan *sarama.ConsumerMessage {
 
 func TestConsumerGroupHandler_Lifecycle(t *testing.T) {
 	acc := &testutil.Accumulator{}
-	parser := value.NewValueParser("cpu", "int", "", nil)
-	cg := NewConsumerGroupHandler(acc, 1, parser, testutil.Logger{})
+	parser := value.Parser{
+		MetricName: "cpu",
+		DataType:   "int",
+	}
+	require.NoError(t, parser.Init())
+	cg := NewConsumerGroupHandler(acc, 1, &parser, testutil.Logger{})
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -301,8 +305,12 @@ func TestConsumerGroupHandler_Lifecycle(t *testing.T) {
 
 func TestConsumerGroupHandler_ConsumeClaim(t *testing.T) {
 	acc := &testutil.Accumulator{}
-	parser := value.NewValueParser("cpu", "int", "", nil)
-	cg := NewConsumerGroupHandler(acc, 1, parser, testutil.Logger{})
+	parser := value.Parser{
+		MetricName: "cpu",
+		DataType:   "int",
+	}
+	require.NoError(t, parser.Init())
+	cg := NewConsumerGroupHandler(acc, 1, &parser, testutil.Logger{})
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
@@ -415,8 +423,12 @@ func TestConsumerGroupHandler_Handle(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			acc := &testutil.Accumulator{}
-			parser := value.NewValueParser("cpu", "int", "", nil)
-			cg := NewConsumerGroupHandler(acc, 1, parser, testutil.Logger{})
+			parser := value.Parser{
+				MetricName: "cpu",
+				DataType:   "int",
+			}
+			require.NoError(t, parser.Init())
+			cg := NewConsumerGroupHandler(acc, 1, &parser, testutil.Logger{})
 			cg.MaxMessageLen = tt.maxMessageLen
 			cg.TopicTag = tt.topicTag
 

plugins/parsers/all/all.go

@@ -7,6 +7,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/parsers/form_urlencoded"
 	_ "github.com/influxdata/telegraf/plugins/parsers/json"
 	_ "github.com/influxdata/telegraf/plugins/parsers/json_v2"
+	_ "github.com/influxdata/telegraf/plugins/parsers/value"
 	_ "github.com/influxdata/telegraf/plugins/parsers/wavefront"
 	_ "github.com/influxdata/telegraf/plugins/parsers/xpath"
 )

plugins/parsers/registry.go

@@ -15,7 +15,6 @@ import (
 	"github.com/influxdata/telegraf/plugins/parsers/prometheusremotewrite"
 	"github.com/influxdata/telegraf/plugins/parsers/temporary/json_v2"
 	"github.com/influxdata/telegraf/plugins/parsers/temporary/xpath"
-	"github.com/influxdata/telegraf/plugins/parsers/value"
 )
 
 // Creator is the function to create a new parser
@@ -202,9 +201,6 @@ func NewParser(config *Config) (Parser, error) {
 	var err error
 	var parser Parser
 	switch config.DataFormat {
-	case "value":
-		parser, err = NewValueParser(config.MetricName,
-			config.DataType, config.ValueFieldName, config.DefaultTags)
 	case "influx":
 		if config.InfluxParserType == "upstream" {
 			parser, err = NewInfluxUpstreamParser()
@@ -301,15 +297,6 @@ func NewGraphiteParser(
 	return graphite.NewGraphiteParser(separator, templates, defaultTags)
 }
 
-func NewValueParser(
-	metricName string,
-	dataType string,
-	fieldName string,
-	defaultTags map[string]string,
-) (Parser, error) {
-	return value.NewValueParser(metricName, dataType, fieldName, defaultTags), nil
-}
-
 func NewDropwizardParser(
 	metricRegistryPath string,
 	timePath string,

plugins/parsers/value/parser.go

@@ -9,16 +9,17 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/metric"
+	"github.com/influxdata/telegraf/plugins/parsers"
 )
 
-type ValueParser struct {
-	MetricName  string
-	DataType    string
-	DefaultTags map[string]string
-	FieldName   string
+type Parser struct {
+	DataType    string            `toml:"data_type"`
+	FieldName   string            `toml:"value_field_name"`
+	MetricName  string            `toml:"-"`
+	DefaultTags map[string]string `toml:"-"`
 }
 
-func (v *ValueParser) Parse(buf []byte) ([]telegraf.Metric, error) {
+func (v *Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
 	vStr := string(bytes.TrimSpace(bytes.Trim(buf, "\x00")))
 
 	// unless it's a string, separate out any fields in the buffer,
@@ -54,7 +55,7 @@ func (v *ValueParser) Parse(buf []byte) ([]telegraf.Metric, error) {
 	return []telegraf.Metric{m}, nil
 }
 
-func (v *ValueParser) ParseLine(line string) (telegraf.Metric, error) {
+func (v *Parser) ParseLine(line string) (telegraf.Metric, error) {
 	metrics, err := v.Parse([]byte(line))
 
 	if err != nil {
@@ -68,19 +69,32 @@ func (v *ValueParser) ParseLine(line string) (telegraf.Metric, error) {
 	return metrics[0], nil
 }
 
-func (v *ValueParser) SetDefaultTags(tags map[string]string) {
+func (v *Parser) SetDefaultTags(tags map[string]string) {
 	v.DefaultTags = tags
 }
 
-func NewValueParser(metricName, dataType, fieldName string, defaultTags map[string]string) *ValueParser {
-	if fieldName == "" {
-		fieldName = "value"
-	}
+// InitFromConfig is a compatibility function to construct the parser the old way
+func (v *Parser) InitFromConfig(config *parsers.Config) error {
+	v.MetricName = config.MetricName
+	v.DefaultTags = config.DefaultTags
+	return v.Init()
+}
 
-	return &ValueParser{
-		MetricName:  metricName,
-		DataType:    dataType,
-		DefaultTags: defaultTags,
-		FieldName:   fieldName,
+func (v *Parser) Init() error {
+	if v.FieldName == "" {
+		v.FieldName = "value"
 	}
+
+	return nil
+}
+
+func init() {
+	parsers.Add("value",
+		func(defaultMetricName string) telegraf.Parser {
+			return &Parser{
+				FieldName:  "value",
+				MetricName: defaultMetricName,
+			}
+		},
+	)
 }