feat: Migrate influx and influx_upstream parsers to new style

plugins/common/shim/output.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 
 	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 )
 
 // AddOutput adds the input to the shim. Later calls to Run() will run this.
@@ -23,7 +23,8 @@ func (s *Shim) AddOutput(output telegraf.Output) error {
 }
 
 func (s *Shim) RunOutput() error {
-	parser, err := parsers.NewInfluxParser()
+	parser := influx.Parser{}
+	err := parser.Init()
 	if err != nil {
 		return fmt.Errorf("Failed to create new parser: %w", err)
 	}

plugins/common/shim/processor_test.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/metric"
-	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/plugins/serializers"
 )
 
@@ -53,7 +53,8 @@ func testSendAndReceive(t *testing.T, fieldKey string, fieldValue string) {
 	}()
 
 	serializer, _ := serializers.NewInfluxSerializer()
-	parser, _ := parsers.NewInfluxParser()
+	parser := influx.Parser{}
+	require.NoError(t, parser.Init())
 
 	m := metric.New("thing",
 		map[string]string{

plugins/inputs/cloud_pubsub/cloud_pubsub_test.go

@@ -7,7 +7,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
-	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/testutil"
 )
 
@@ -19,7 +19,8 @@ const (
 func TestRunParse(t *testing.T) {
 	subID := "sub-run-parse"
 
-	testParser, _ := parsers.NewInfluxParser()
+	testParser := &influx.Parser{}
+	require.NoError(t, testParser.Init())
 
 	sub := &stubSub{
 		id:       subID,
@@ -63,7 +64,8 @@ func TestRunParse(t *testing.T) {
 func TestRunBase64(t *testing.T) {
 	subID := "sub-run-base64"
 
-	testParser, _ := parsers.NewInfluxParser()
+	testParser := &influx.Parser{}
+	require.NoError(t, testParser.Init())
 
 	sub := &stubSub{
 		id:       subID,
@@ -107,7 +109,8 @@ func TestRunBase64(t *testing.T) {
 func TestRunInvalidMessages(t *testing.T) {
 	subID := "sub-invalid-messages"
 
-	testParser, _ := parsers.NewInfluxParser()
+	testParser := &influx.Parser{}
+	require.NoError(t, testParser.Init())
 
 	sub := &stubSub{
 		id:       subID,
@@ -154,7 +157,8 @@ func TestRunOverlongMessages(t *testing.T) {
 
 	acc := &testutil.Accumulator{}
 
-	testParser, _ := parsers.NewInfluxParser()
+	testParser := &influx.Parser{}
+	require.NoError(t, testParser.Init())
 
 	sub := &stubSub{
 		id:       subID,
@@ -201,7 +205,8 @@ func TestRunErrorInSubscriber(t *testing.T) {
 
 	acc := &testutil.Accumulator{}
 
-	testParser, _ := parsers.NewInfluxParser()
+	testParser := &influx.Parser{}
+	require.NoError(t, testParser.Init())
 
 	sub := &stubSub{
 		id:       subID,

plugins/inputs/cloud_pubsub_push/cloud_pubsub_push_test.go

@@ -18,6 +18,7 @@ import (
 	"github.com/influxdata/telegraf/config"
 	"github.com/influxdata/telegraf/models"
 	"github.com/influxdata/telegraf/plugins/parsers"
+	_ "github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/testutil"
 )
 

plugins/inputs/execd/execd_test.go

@@ -16,7 +16,7 @@ import (
 	"github.com/influxdata/telegraf/config"
 	"github.com/influxdata/telegraf/metric"
 	"github.com/influxdata/telegraf/models"
-	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/plugins/serializers"
 	"github.com/influxdata/telegraf/testutil"
 )
@@ -42,8 +42,8 @@ func TestSettingConfigWorks(t *testing.T) {
 }
 
 func TestExternalInputWorks(t *testing.T) {
-	influxParser, err := parsers.NewInfluxParser()
-	require.NoError(t, err)
+	influxParser := &influx.Parser{}
+	require.NoError(t, influxParser.Init())
 
 	exe, err := os.Executable()
 	require.NoError(t, err)
@@ -76,8 +76,8 @@ func TestExternalInputWorks(t *testing.T) {
 }
 
 func TestParsesLinesContainingNewline(t *testing.T) {
-	parser, err := parsers.NewInfluxParser()
-	require.NoError(t, err)
+	parser := &influx.Parser{}
+	require.NoError(t, parser.Init())
 
 	metrics := make(chan telegraf.Metric, 10)
 	defer close(metrics)

plugins/inputs/http_listener_v2/http_listener_v2_test.go

@@ -17,8 +17,8 @@ import (
 	"github.com/stretchr/testify/require"
 
 	"github.com/influxdata/telegraf/config"
-	"github.com/influxdata/telegraf/plugins/parsers"
 	"github.com/influxdata/telegraf/plugins/parsers/form_urlencoded"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/testutil"
 )
 
@@ -46,7 +46,8 @@ var (
 )
 
 func newTestHTTPListenerV2() *HTTPListenerV2 {
-	parser, _ := parsers.NewInfluxParser()
+	parser := &influx.Parser{}
+	_ = parser.Init()
 
 	listener := &HTTPListenerV2{
 		Log:            testutil.Logger{},
@@ -70,7 +71,8 @@ func newTestHTTPAuthListener() *HTTPListenerV2 {
 }
 
 func newTestHTTPSListenerV2() *HTTPListenerV2 {
-	parser, _ := parsers.NewInfluxParser()
+	parser := &influx.Parser{}
+	_ = parser.Init()
 
 	listener := &HTTPListenerV2{
 		Log:            testutil.Logger{},
@@ -109,7 +111,8 @@ func createURL(listener *HTTPListenerV2, scheme string, path string, rawquery st
 }
 
 func TestInvalidListenerConfig(t *testing.T) {
-	parser, _ := parsers.NewInfluxParser()
+	parser := &influx.Parser{}
+	require.NoError(t, parser.Init())
 
 	listener := &HTTPListenerV2{
 		Log:            testutil.Logger{},
@@ -319,7 +322,8 @@ func TestWriteHTTPNoNewline(t *testing.T) {
 }
 
 func TestWriteHTTPExactMaxBodySize(t *testing.T) {
-	parser, _ := parsers.NewInfluxParser()
+	parser := &influx.Parser{}
+	require.NoError(t, parser.Init())
 
 	listener := &HTTPListenerV2{
 		Log:            testutil.Logger{},
@@ -344,7 +348,8 @@ func TestWriteHTTPExactMaxBodySize(t *testing.T) {
 }
 
 func TestWriteHTTPVerySmallMaxBody(t *testing.T) {
-	parser, _ := parsers.NewInfluxParser()
+	parser := &influx.Parser{}
+	require.NoError(t, parser.Init())
 
 	listener := &HTTPListenerV2{
 		Log:            testutil.Logger{},

plugins/inputs/influxdb_v2_listener/influxdb_v2_listener.go

@@ -249,7 +249,12 @@ func (h *InfluxDBV2Listener) handleWrite() http.HandlerFunc {
 		var metrics []telegraf.Metric
 		var err error
 		if h.ParserType == "upstream" {
-			parser := influx_upstream.NewParser()
+			parser := influx_upstream.Parser{}
+			err = parser.Init()
+			if err != ErrEOF && err != nil {
+				h.Log.Debugf("Error initializing parser: %v", err.Error())
+				return
+			}
 			parser.SetTimeFunc(influx_upstream.TimeFunc(h.timeFunc))
 
 			if precisionStr != "" {
@@ -259,13 +264,17 @@ func (h *InfluxDBV2Listener) handleWrite() http.HandlerFunc {
 
 			metrics, err = parser.Parse(bytes)
 		} else {
-			metricHandler := influx.NewMetricHandler()
-			parser := influx.NewParser(metricHandler)
+			parser := influx.Parser{}
+			err = parser.Init()
+			if err != ErrEOF && err != nil {
+				h.Log.Debugf("Error initializing parser: %v", err.Error())
+				return
+			}
 			parser.SetTimeFunc(h.timeFunc)
 
 			if precisionStr != "" {
 				precision := getPrecisionMultiplier(precisionStr)
-				metricHandler.SetTimePrecision(precision)
+				parser.SetTimePrecision(precision)
 			}
 
 			metrics, err = parser.Parse(bytes)

plugins/inputs/kafka_consumer_legacy/kafka_consumer_legacy_integration_test.go

@@ -8,7 +8,7 @@ import (
 	"github.com/Shopify/sarama"
 	"github.com/stretchr/testify/require"
 
-	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/testutil"
 )
 
@@ -43,8 +43,9 @@ func TestReadsMetricsFromKafkaIntegration(t *testing.T) {
 		PointBuffer:    100000,
 		Offset:         "oldest",
 	}
-	p, _ := parsers.NewInfluxParser()
-	k.SetParser(p)
+	parser := &influx.Parser{}
+	require.NoError(t, parser.Init())
+	k.SetParser(parser)
 
 	// Verify that we can now gather the sent message
 	var acc testutil.Accumulator

plugins/inputs/kafka_consumer_legacy/kafka_consumer_legacy_test.go

@@ -6,8 +6,8 @@ import (
 
 	"github.com/Shopify/sarama"
 
-	"github.com/influxdata/telegraf/plugins/parsers"
 	"github.com/influxdata/telegraf/plugins/parsers/graphite"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/plugins/parsers/json"
 	"github.com/influxdata/telegraf/testutil"
 
@@ -44,9 +44,10 @@ func TestRunParser(t *testing.T) {
 	k.acc = &acc
 	defer close(k.done)
 
-	var err error
-	k.parser, err = parsers.NewInfluxParser()
-	require.NoError(t, err)
+	parser := &influx.Parser{}
+	require.NoError(t, parser.Init())
+	k.parser = parser
+
 	go k.receiver()
 	in <- saramaMsg(testMsg)
 	acc.Wait(1)
@@ -61,9 +62,10 @@ func TestRunParserInvalidMsg(t *testing.T) {
 	k.acc = &acc
 	defer close(k.done)
 
-	var err error
-	k.parser, err = parsers.NewInfluxParser()
-	require.NoError(t, err)
+	parser := &influx.Parser{}
+	require.NoError(t, parser.Init())
+	k.parser = parser
+
 	go k.receiver()
 	in <- saramaMsg(invalidMsg)
 	acc.WaitError(1)
@@ -95,9 +97,10 @@ func TestRunParserAndGather(t *testing.T) {
 	k.acc = &acc
 	defer close(k.done)
 
-	var err error
-	k.parser, err = parsers.NewInfluxParser()
-	require.NoError(t, err)
+	parser := &influx.Parser{}
+	require.NoError(t, parser.Init())
+	k.parser = parser
+
 	go k.receiver()
 	in <- saramaMsg(testMsg)
 	acc.Wait(1)

plugins/inputs/mqtt_consumer/mqtt_consumer_test.go

@@ -8,6 +8,7 @@ import (
 	mqtt "github.com/eclipse/paho.mqtt.golang"
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/testutil"
 	"github.com/stretchr/testify/require"
 )
@@ -438,11 +439,11 @@ func TestTopicTag(t *testing.T) {
 			plugin.TopicTag = tt.topicTag()
 			plugin.TopicParsing = tt.topicParsing
 
-			parser, err := parsers.NewInfluxParser()
-			require.NoError(t, err)
+			parser := &influx.Parser{}
+			require.NoError(t, parser.Init())
 			plugin.SetParser(parser)
 
-			err = plugin.Init()
+			err := plugin.Init()
 			require.Equal(t, tt.expectedError, err)
 			if tt.expectedError != nil {
 				return

plugins/inputs/nsq_consumer/nsq_consumer_test.go

@@ -14,7 +14,7 @@ import (
 	"github.com/nsqio/go-nsq"
 	"github.com/stretchr/testify/require"
 
-	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/influxdata/telegraf/testutil"
 )
 
@@ -49,7 +49,8 @@ func TestReadsMetricsFromNSQ(t *testing.T) {
 		Nsqd:                   []string{"127.0.0.1:4155"},
 	}
 
-	p, _ := parsers.NewInfluxParser()
+	p := &influx.Parser{}
+	require.NoError(t, p.Init())
 	consumer.SetParser(p)
 	var acc testutil.Accumulator
 	require.Len(t, acc.Metrics, 0, "There should not be any points")

plugins/inputs/socket_listener/socket_listener.go

@@ -20,6 +20,7 @@ import (
 	tlsint "github.com/influxdata/telegraf/plugins/common/tls"
 	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/parsers/influx"
 )
 
 // DO NOT REMOVE THE NEXT TWO LINES! This is required to embed the sampleConfig data.
@@ -389,7 +390,8 @@ func (sl *SocketListener) Stop() {
 }
 
 func newSocketListener() *SocketListener {
-	parser, _ := parsers.NewInfluxParser()
+	parser := &influx.Parser{}
+	_ = parser.Init()
 
 	return &SocketListener{
 		Parser: parser,