Fix dropwizard parsing error for metrics that need escaped (#4142)

If the dropwizard parser cannot convert the metric name into a valid
line protocol series then we will accept the name as is.

metric.go

@@ -54,6 +54,8 @@ type Metric interface {
 	AddField(key string, value interface{})
 	RemoveField(key string)
 
+	SetTime(t time.Time)
+
 	// HashID returns an unique identifier for the series.
 	HashID() uint64
 

metric/metric.go

@@ -202,6 +202,10 @@ func (m *metric) RemoveField(key string) {
 	}
 }
 
+func (m *metric) SetTime(t time.Time) {
+	m.tm = t
+}
+
 func (m *metric) Copy() telegraf.Metric {
 	m2 := &metric{
 		name:      m.name,

plugins/parsers/dropwizard/parser.go

@@ -1,7 +1,6 @@
 package dropwizard
 
 import (
-	"bytes"
 	"encoding/json"
 	"fmt"
 	"log"
@@ -10,6 +9,7 @@ import (
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal/templating"
+	"github.com/influxdata/telegraf/metric"
 	"github.com/influxdata/telegraf/plugins/parsers/influx"
 	"github.com/tidwall/gjson"
 )
@@ -19,8 +19,8 @@ var keyEscaper = strings.NewReplacer(" ", "\\ ", ",", "\\,", "=", "\\=")
 
 // Parser parses json inputs containing dropwizard metrics,
 // either top-level or embedded inside a json field.
-// This parser is using gjon for retrieving paths within the json file.
+// This parser is using gjson for retrieving paths within the json file.
-type Parser struct {
+type parser struct {
 
 	// an optional json path containing the metric registry object
 	// if left empty, the whole json object is parsed as a metric registry
@@ -45,15 +45,28 @@ type Parser struct {
 	// an optional map of default tags to use for metrics
 	DefaultTags map[string]string
 
-	// templating configuration
+	separator      string
-	Separator string
-	Templates []string
-
 	templateEngine *templating.Engine
+
+	timeFunc metric.TimeFunc
+
+	// seriesParser parses line protocol measurement + tags
+	seriesParser *influx.Parser
+}
+
+func NewParser() *parser {
+	handler := influx.NewMetricHandler()
+	seriesParser := influx.NewSeriesParser(handler)
+
+	parser := &parser{
+		timeFunc:     time.Now,
+		seriesParser: seriesParser,
+	}
+	return parser
 }
 
 // Parse parses the input bytes to an array of metrics
-func (p *Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
+func (p *parser) Parse(buf []byte) ([]telegraf.Metric, error) {
 
 	metrics := make([]telegraf.Metric, 0)
 
@@ -100,28 +113,38 @@ func (p *Parser) Parse(buf []byte) ([]telegraf.Metric, error) {
 	return metrics, nil
 }
 
-// InitTemplating initializes the templating support
+func (p *parser) SetTemplates(separator string, templates []string) error {
-func (p *Parser) InitTemplating() error {
+	if len(templates) == 0 {
-	if len(p.Templates) > 0 {
+		p.templateEngine = nil
-		defaultTemplate, _ := templating.NewDefaultTemplateWithPattern("measurement*")
+		return nil
-		templateEngine, err := templating.NewEngine(p.Separator, defaultTemplate, p.Templates)
+	}
-		p.templateEngine = templateEngine
+
+	defaultTemplate, err := templating.NewDefaultTemplateWithPattern("measurement*")
+	if err != nil {
 		return err
 	}
+
+	templateEngine, err := templating.NewEngine(separator, defaultTemplate, templates)
+	if err != nil {
+		return err
+	}
+
+	p.separator = separator
+	p.templateEngine = templateEngine
 	return nil
 }
 
 // ParseLine is not supported by the dropwizard format
-func (p *Parser) ParseLine(line string) (telegraf.Metric, error) {
+func (p *parser) ParseLine(line string) (telegraf.Metric, error) {
 	return nil, fmt.Errorf("ParseLine not supported: %s, for data format: dropwizard", line)
 }
 
 // SetDefaultTags sets the default tags
-func (p *Parser) SetDefaultTags(tags map[string]string) {
+func (p *parser) SetDefaultTags(tags map[string]string) {
 	p.DefaultTags = tags
 }
 
-func (p *Parser) readTags(buf []byte) map[string]string {
+func (p *parser) readTags(buf []byte) map[string]string {
 
 	if p.TagsPath != "" {
 		var tagsBytes []byte
@@ -147,7 +170,7 @@ func (p *Parser) readTags(buf []byte) map[string]string {
 	return tags
 }
 
-func (p *Parser) parseTime(buf []byte) (time.Time, error) {
+func (p *parser) parseTime(buf []byte) (time.Time, error) {
 
 	if p.TimePath != "" {
 		timeFormat := p.TimeFormat
@@ -157,19 +180,19 @@ func (p *Parser) parseTime(buf []byte) (time.Time, error) {
 		timeString := gjson.GetBytes(buf, p.TimePath).String()
 		if timeString == "" {
 			err := fmt.Errorf("time not found in JSON path %s", p.TimePath)
-			return time.Now().UTC(), err
+			return p.timeFunc(), err
 		}
 		t, err := time.Parse(timeFormat, timeString)
 		if err != nil {
 			err = fmt.Errorf("time %s cannot be parsed with format %s, %s", timeString, timeFormat, err)
-			return time.Now().UTC(), err
+			return p.timeFunc(), err
 		}
 		return t.UTC(), nil
 	}
-	return time.Now().UTC(), nil
+	return p.timeFunc(), nil
 }
 
-func (p *Parser) unmarshalMetrics(buf []byte) (map[string]interface{}, error) {
+func (p *parser) unmarshalMetrics(buf []byte) (map[string]interface{}, error) {
 
 	var registryBytes []byte
 	if p.MetricRegistryPath != "" {
@@ -195,11 +218,8 @@ func (p *Parser) unmarshalMetrics(buf []byte) (map[string]interface{}, error) {
 	return jsonOut, nil
 }
 
-func (p *Parser) readDWMetrics(metricType string, dwms interface{}, metrics []telegraf.Metric, tm time.Time) []telegraf.Metric {
+func (p *parser) readDWMetrics(metricType string, dwms interface{}, metrics []telegraf.Metric, tm time.Time) []telegraf.Metric {
-
+	if dwmsTyped, ok := dwms.(map[string]interface{}); ok {
-	switch dwmsTyped := dwms.(type) {
-	case map[string]interface{}:
-		var metricsBuffer bytes.Buffer
 		for dwmName, dwmFields := range dwmsTyped {
 			measurementName := dwmName
 			tags := make(map[string]string)
@@ -207,59 +227,46 @@ func (p *Parser) readDWMetrics(metricType string, dwms interface{}, metrics []te
 			if p.templateEngine != nil {
 				measurementName, tags, fieldPrefix, _ = p.templateEngine.Apply(dwmName)
 				if len(fieldPrefix) > 0 {
-					fieldPrefix = fmt.Sprintf("%s%s", fieldPrefix, p.Separator)
+					fieldPrefix = fmt.Sprintf("%s%s", fieldPrefix, p.separator)
 				}
 			}
-			tags["metric_type"] = metricType
-
-			measurementWithTags := measurementName
-			for tagName, tagValue := range tags {
-				tagKeyValue := fmt.Sprintf("%s=%s", keyEscaper.Replace(tagName), keyEscaper.Replace(tagValue))
-				measurementWithTags = fmt.Sprintf("%s,%s", measurementWithTags, tagKeyValue)
-			}
 
-			fields := make([]string, 0)
+			parsed, err := p.seriesParser.Parse([]byte(measurementName))
-			switch t := dwmFields.(type) {
+			var m telegraf.Metric
-			case map[string]interface{}: // json object
+			if err != nil || len(parsed) != 1 {
-				for fieldName, fieldValue := range t {
+				m, err = metric.New(measurementName, map[string]string{}, map[string]interface{}{}, tm)
-					key := keyEscaper.Replace(fieldPrefix + fieldName)
-					switch v := fieldValue.(type) {
-					case float64:
-						fields = append(fields, fmt.Sprintf("%s=%f", key, v))
-					case string:
-						fields = append(fields, fmt.Sprintf("%s=\"%s\"", key, fieldEscaper.Replace(v)))
-					case bool:
-						fields = append(fields, fmt.Sprintf("%s=%t", key, v))
-					default: // ignore
-					}
-				}
-			default: // ignore
-			}
-
-			metricsBuffer.WriteString(fmt.Sprintf("%s,metric_type=%s ", measurementWithTags, metricType))
-			metricsBuffer.WriteString(strings.Join(fields, ","))
-			metricsBuffer.WriteString("\n")
-		}
-
-		handler := influx.NewMetricHandler()
-		handler.SetTimeFunc(func() time.Time { return tm })
-		parser := influx.NewParser(handler)
-		newMetrics, err := parser.Parse(metricsBuffer.Bytes())
 				if err != nil {
 					log.Printf("W! failed to create metric of type '%s': %s\n", metricType, err)
+					continue
+				}
+			} else {
+				m = parsed[0]
+				m.SetTime(tm)
 			}
 
-		return append(metrics, newMetrics...)
+			m.AddTag("metric_type", metricType)
+			for k, v := range tags {
+				m.AddTag(k, v)
+			}
+
+			if fields, ok := dwmFields.(map[string]interface{}); ok {
+				for k, v := range fields {
+					switch v := v.(type) {
+					case float64, string, bool:
+						m.AddField(fieldPrefix+k, v)
 					default:
+						// ignore
+					}
+				}
+			}
+
+			metrics = append(metrics, m)
+		}
+	}
+
 	return metrics
-	}
-
 }
 
-func arraymap(vs []string, f func(string) string) []string {
+func (p *parser) SetTimeFunc(f metric.TimeFunc) {
-	vsm := make([]string, len(vs))
+	p.timeFunc = f
-	for i, v := range vs {
-		vsm[i] = f(v)
-	}
-	return vsm
 }

plugins/parsers/dropwizard/parser_test.go

@@ -4,13 +4,19 @@ import (
 	"testing"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/metric"
 
 	"fmt"
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
+var TimeFunc = func() time.Time {
+	return time.Unix(0, 0)
+}
+
 // validEmptyJSON is a valid dropwizard json document, but without any metrics
 const validEmptyJSON = `
 {
@@ -24,7 +30,7 @@ const validEmptyJSON = `
 `
 
 func TestParseValidEmptyJSON(t *testing.T) {
-	parser := Parser{}
+	parser := NewParser()
 
 	// Most basic vanilla test
 	metrics, err := parser.Parse([]byte(validEmptyJSON))
@@ -49,7 +55,7 @@ const validCounterJSON = `
 `
 
 func TestParseValidCounterJSON(t *testing.T) {
-	parser := Parser{}
+	parser := NewParser()
 
 	metrics, err := parser.Parse([]byte(validCounterJSON))
 	assert.NoError(t, err)
@@ -87,11 +93,10 @@ const validEmbeddedCounterJSON = `
 func TestParseValidEmbeddedCounterJSON(t *testing.T) {
 	timeFormat := "2006-01-02T15:04:05Z07:00"
 	metricTime, _ := time.Parse(timeFormat, "2017-02-22T15:33:03.662+03:00")
-	parser := Parser{
+	parser := NewParser()
-		MetricRegistryPath: "metrics",
+	parser.MetricRegistryPath = "metrics"
-		TagsPath:           "tags",
+	parser.TagsPath = "tags"
-		TimePath:           "time",
+	parser.TimePath = "time"
-	}
 
 	metrics, err := parser.Parse([]byte(validEmbeddedCounterJSON))
 	assert.NoError(t, err)
@@ -109,11 +114,10 @@ func TestParseValidEmbeddedCounterJSON(t *testing.T) {
 	assert.True(t, metricTime.Equal(metrics[0].Time()), fmt.Sprintf("%s should be equal to %s", metrics[0].Time(), metricTime))
 
 	// now test json tags through TagPathsMap
-	parser2 := Parser{
+	parser2 := NewParser()
-		MetricRegistryPath: "metrics",
+	parser2.MetricRegistryPath = "metrics"
-		TagPathsMap:        map[string]string{"tag1": "tags.tag1"},
+	parser2.TagPathsMap = map[string]string{"tag1": "tags.tag1"}
-		TimePath:           "time",
+	parser2.TimePath = "time"
-	}
 	metrics2, err2 := parser2.Parse([]byte(validEmbeddedCounterJSON))
 	assert.NoError(t, err2)
 	assert.Equal(t, map[string]string{"metric_type": "counter", "tag1": "green"}, metrics2[0].Tags())
@@ -141,7 +145,7 @@ const validMeterJSON1 = `
 `
 
 func TestParseValidMeterJSON1(t *testing.T) {
-	parser := Parser{}
+	parser := NewParser()
 
 	metrics, err := parser.Parse([]byte(validMeterJSON1))
 	assert.NoError(t, err)
@@ -181,7 +185,7 @@ const validMeterJSON2 = `
 `
 
 func TestParseValidMeterJSON2(t *testing.T) {
-	parser := Parser{}
+	parser := NewParser()
 
 	metrics, err := parser.Parse([]byte(validMeterJSON2))
 	assert.NoError(t, err)
@@ -215,7 +219,7 @@ const validGaugeJSON = `
 `
 
 func TestParseValidGaugeJSON(t *testing.T) {
-	parser := Parser{}
+	parser := NewParser()
 
 	metrics, err := parser.Parse([]byte(validGaugeJSON))
 	assert.NoError(t, err)
@@ -254,7 +258,7 @@ const validHistogramJSON = `
 `
 
 func TestParseValidHistogramJSON(t *testing.T) {
-	parser := Parser{}
+	parser := NewParser()
 
 	metrics, err := parser.Parse([]byte(validHistogramJSON))
 	assert.NoError(t, err)
@@ -309,7 +313,7 @@ const validTimerJSON = `
 `
 
 func TestParseValidTimerJSON(t *testing.T) {
-	parser := Parser{}
+	parser := NewParser()
 
 	metrics, err := parser.Parse([]byte(validTimerJSON))
 	assert.NoError(t, err)
@@ -360,7 +364,7 @@ const validAllJSON = `
 `
 
 func TestParseValidAllJSON(t *testing.T) {
-	parser := Parser{}
+	parser := NewParser()
 
 	metrics, err := parser.Parse([]byte(validAllJSON))
 	assert.NoError(t, err)
@@ -369,18 +373,19 @@ func TestParseValidAllJSON(t *testing.T) {
 
 func TestTagParsingProblems(t *testing.T) {
 	// giving a wrong path results in empty tags
-	parser1 := Parser{MetricRegistryPath: "metrics", TagsPath: "tags1"}
+	parser1 := NewParser()
+	parser1.MetricRegistryPath = "metrics"
+	parser1.TagsPath = "tags1"
 	metrics1, err1 := parser1.Parse([]byte(validEmbeddedCounterJSON))
 	assert.NoError(t, err1)
 	assert.Len(t, metrics1, 1)
 	assert.Equal(t, map[string]string{"metric_type": "counter"}, metrics1[0].Tags())
 
 	// giving a wrong TagsPath falls back to TagPathsMap
-	parser2 := Parser{
+	parser2 := NewParser()
-		MetricRegistryPath: "metrics",
+	parser2.MetricRegistryPath = "metrics"
-		TagsPath:           "tags1",
+	parser2.TagsPath = "tags1"
-		TagPathsMap:        map[string]string{"tag1": "tags.tag1"},
+	parser2.TagPathsMap = map[string]string{"tag1": "tags.tag1"}
-	}
 	metrics2, err2 := parser2.Parse([]byte(validEmbeddedCounterJSON))
 	assert.NoError(t, err2)
 	assert.Len(t, metrics2, 1)
@@ -425,23 +430,21 @@ const sampleTemplateJSON = `
 `
 
 func TestParseSampleTemplateJSON(t *testing.T) {
-	parser := Parser{
+	parser := NewParser()
-		Separator: "_",
+	err := parser.SetTemplates("_", []string{
-		Templates: []string{
 		"jenkins.* measurement.metric.metric.field",
 		"vm.* measurement.measurement.pool.field",
-		},
+	})
-	}
+	require.NoError(t, err)
-	parser.InitTemplating()
 
 	metrics, err := parser.Parse([]byte(sampleTemplateJSON))
-	assert.NoError(t, err)
+	require.NoError(t, err)
 
-	assert.Len(t, metrics, 11)
+	require.Len(t, metrics, 11)
 
 	jenkinsMetric := search(metrics, "jenkins", nil, "")
-	assert.NotNil(t, jenkinsMetric, "the metrics should contain a jenkins measurement")
+	require.NotNil(t, jenkinsMetric, "the metrics should contain a jenkins measurement")
-	assert.Equal(t, map[string]interface{}{
+	require.Equal(t, map[string]interface{}{
 		"duration_count":  float64(1),
 		"duration_max":    float64(2),
 		"duration_mean":   float64(3),
@@ -454,21 +457,21 @@ func TestParseSampleTemplateJSON(t *testing.T) {
 		"duration_p999":   float64(10),
 		"duration_stddev": float64(11),
 	}, jenkinsMetric.Fields())
-	assert.Equal(t, map[string]string{"metric_type": "histogram", "metric": "job_building"}, jenkinsMetric.Tags())
+	require.Equal(t, map[string]string{"metric_type": "histogram", "metric": "job_building"}, jenkinsMetric.Tags())
 
 	vmMemoryHeapCommitted := search(metrics, "vm_memory", map[string]string{"pool": "heap"}, "committed_value")
-	assert.NotNil(t, vmMemoryHeapCommitted)
+	require.NotNil(t, vmMemoryHeapCommitted)
-	assert.Equal(t, map[string]interface{}{
+	require.Equal(t, map[string]interface{}{
 		"committed_value": float64(1),
 	}, vmMemoryHeapCommitted.Fields())
-	assert.Equal(t, map[string]string{"metric_type": "gauge", "pool": "heap"}, vmMemoryHeapCommitted.Tags())
+	require.Equal(t, map[string]string{"metric_type": "gauge", "pool": "heap"}, vmMemoryHeapCommitted.Tags())
 
 	vmMemoryNonHeapCommitted := search(metrics, "vm_memory", map[string]string{"pool": "non-heap"}, "committed_value")
-	assert.NotNil(t, vmMemoryNonHeapCommitted)
+	require.NotNil(t, vmMemoryNonHeapCommitted)
-	assert.Equal(t, map[string]interface{}{
+	require.Equal(t, map[string]interface{}{
 		"committed_value": float64(6),
 	}, vmMemoryNonHeapCommitted.Fields())
-	assert.Equal(t, map[string]string{"metric_type": "gauge", "pool": "non-heap"}, vmMemoryNonHeapCommitted.Tags())
+	require.Equal(t, map[string]string{"metric_type": "gauge", "pool": "non-heap"}, vmMemoryNonHeapCommitted.Tags())
 }
 
 func search(metrics []telegraf.Metric, name string, tags map[string]string, fieldName string) telegraf.Metric {
@@ -493,3 +496,105 @@ func containsAll(t1 map[string]string, t2 map[string]string) bool {
 	}
 	return true
 }
+
+func Metric(v telegraf.Metric, err error) telegraf.Metric {
+	if err != nil {
+		panic(err)
+	}
+	return v
+}
+
+func NoError(t *testing.T, err error) {
+	require.NoError(t, err)
+}
+
+func TestDropWizard(t *testing.T) {
+	tests := []struct {
+		name    string
+		input   []byte
+		metrics []telegraf.Metric
+		errFunc func(t *testing.T, err error)
+	}{
+		{
+			name:  "minimal",
+			input: []byte(`{"version": "3.0.0", "counters": {"cpu": {"value": 42}}}`),
+			metrics: []telegraf.Metric{
+				Metric(
+					metric.New(
+						"cpu",
+						map[string]string{
+							"metric_type": "counter",
+						},
+						map[string]interface{}{
+							"value": 42.0,
+						},
+						TimeFunc(),
+					),
+				),
+			},
+			errFunc: NoError,
+		},
+		{
+			name:  "name with space unescaped",
+			input: []byte(`{"version": "3.0.0", "counters": {"hello world": {"value": 42}}}`),
+			metrics: []telegraf.Metric{
+				Metric(
+					metric.New(
+						"hello world",
+						map[string]string{
+							"metric_type": "counter",
+						},
+						map[string]interface{}{
+							"value": 42.0,
+						},
+						TimeFunc(),
+					),
+				),
+			},
+			errFunc: NoError,
+		},
+		{
+			name:  "name with space single slash escaped is not valid JSON",
+			input: []byte(`{"version": "3.0.0", "counters": {"hello\ world": {"value": 42}}}`),
+			errFunc: func(t *testing.T, err error) {
+				require.Error(t, err)
+			},
+		},
+		{
+			name:  "name with space double slash escape",
+			input: []byte(`{"version": "3.0.0", "counters": {"hello\\ world": {"value": 42}}}`),
+			metrics: []telegraf.Metric{
+				Metric(
+					metric.New(
+						"hello world",
+						map[string]string{
+							"metric_type": "counter",
+						},
+						map[string]interface{}{
+							"value": 42.0,
+						},
+						TimeFunc(),
+					),
+				),
+			},
+			errFunc: NoError,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			parser := NewParser()
+			parser.SetTimeFunc(TimeFunc)
+			metrics, err := parser.Parse(tt.input)
+			tt.errFunc(t, err)
+
+			require.Equal(t, len(tt.metrics), len(metrics))
+			for i, expected := range tt.metrics {
+				require.Equal(t, expected.Name(), metrics[i].Name())
+				require.Equal(t, expected.Tags(), metrics[i].Tags())
+				require.Equal(t, expected.Fields(), metrics[i].Fields())
+				require.Equal(t, expected.Time(), metrics[i].Time())
+			}
+		})
+	}
+}

plugins/parsers/influx/machine.go.rl

@@ -221,6 +221,8 @@ discard_line :=
 # main machine.
 align :=
 	(space* comment)* space* measurement_start @hold_recover %eof(yield);
+
+series := measurement tagset $err(parse_error) eol;
 }%%
 
 %% write data;
@@ -242,12 +244,30 @@ type machine struct {
 	p, pe, eof int
 	pb         int
 	handler    Handler
+	initState  int
 	err        error
 }
 
 func NewMachine(handler Handler) *machine {
 	m := &machine{
 		handler: handler,
+		initState: LineProtocol_en_align,
+	}
+
+	%% access m.;
+	%% variable p m.p;
+	%% variable pe m.pe;
+	%% variable eof m.eof;
+	%% variable data m.data;
+	%% write init;
+
+	return m
+}
+
+func NewSeriesMachine(handler Handler) *machine {
+	m := &machine{
+		handler: handler,
+		initState: LineProtocol_en_series,
 	}
 
 	%% access m.;
@@ -269,7 +289,7 @@ func (m *machine) SetData(data []byte) {
 	m.err = nil
 
 	%% write init;
-	m.cs = LineProtocol_en_align
+	m.cs = m.initState
 }
 
 // ParseLine parses a line of input and returns true if more data can be

plugins/parsers/influx/machine_test.go

@@ -1390,3 +1390,80 @@ func BenchmarkMachineProcstat(b *testing.B) {
 		}
 	}
 }
+
+func TestSeriesMachine(t *testing.T) {
+	var tests = []struct {
+		name    string
+		input   []byte
+		results []Result
+		err     error
+	}{
+		{
+			name:    "empty string",
+			input:   []byte(""),
+			results: nil,
+		},
+		{
+			name:  "no tags",
+			input: []byte("cpu"),
+			results: []Result{
+				Result{
+					Name:  Measurement,
+					Value: []byte("cpu"),
+				},
+			},
+		},
+		{
+			name:  "tags",
+			input: []byte("cpu,a=x,b=y"),
+			results: []Result{
+				Result{
+					Name:  Measurement,
+					Value: []byte("cpu"),
+				},
+				Result{
+					Name:  TagKey,
+					Value: []byte("a"),
+				},
+				Result{
+					Name:  TagValue,
+					Value: []byte("x"),
+				},
+				Result{
+					Name:  TagKey,
+					Value: []byte("b"),
+				},
+				Result{
+					Name:  TagValue,
+					Value: []byte("y"),
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handler := &TestingHandler{}
+			fsm := NewSeriesMachine(handler)
+			fsm.SetData(tt.input)
+
+			count := 0
+			for fsm.ParseLine() {
+				if fsm.Err() != nil {
+					handler.AddError(fsm.Err())
+				}
+				count++
+				if count > 20 {
+					break
+				}
+			}
+
+			if fsm.Err() != nil {
+				handler.AddError(fsm.Err())
+			}
+
+			results := handler.Results()
+			require.Equal(t, tt.results, results)
+		})
+	}
+}

plugins/parsers/influx/parser.go

@@ -38,6 +38,7 @@ type Parser struct {
 	handler *MetricHandler
 }
 
+// NewParser returns a Parser than accepts line protocol
 func NewParser(handler *MetricHandler) *Parser {
 	return &Parser{
 		machine: NewMachine(handler),
@@ -45,6 +46,14 @@ func NewParser(handler *MetricHandler) *Parser {
 	}
 }
 
+// NewSeriesParser returns a Parser than accepts a measurement and tagset
+func NewSeriesParser(handler *MetricHandler) *Parser {
+	return &Parser{
+		machine: NewSeriesMachine(handler),
+		handler: handler,
+	}
+}
+
 func (p *Parser) Parse(input []byte) ([]telegraf.Metric, error) {
 	p.Lock()
 	defer p.Unlock()

plugins/parsers/influx/parser_test.go

@@ -616,3 +616,83 @@ func BenchmarkParser(b *testing.B) {
 		})
 	}
 }
+
+func TestSeriesParser(t *testing.T) {
+	var tests = []struct {
+		name      string
+		input     []byte
+		timeFunc  func() time.Time
+		precision time.Duration
+		metrics   []telegraf.Metric
+		err       error
+	}{
+		{
+			name:    "empty",
+			input:   []byte(""),
+			metrics: []telegraf.Metric{},
+		},
+		{
+			name:  "minimal",
+			input: []byte("cpu"),
+			metrics: []telegraf.Metric{
+				Metric(
+					metric.New(
+						"cpu",
+						map[string]string{},
+						map[string]interface{}{},
+						time.Unix(0, 0),
+					),
+				),
+			},
+		},
+		{
+			name:  "tags",
+			input: []byte("cpu,a=x,b=y"),
+			metrics: []telegraf.Metric{
+				Metric(
+					metric.New(
+						"cpu",
+						map[string]string{
+							"a": "x",
+							"b": "y",
+						},
+						map[string]interface{}{},
+						time.Unix(0, 0),
+					),
+				),
+			},
+		},
+		{
+			name:    "missing tag value",
+			input:   []byte("cpu,a="),
+			metrics: []telegraf.Metric{},
+			err: &ParseError{
+				Offset: 6,
+				msg:    ErrTagParse.Error(),
+				buf:    "cpu,a=",
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			handler := NewMetricHandler()
+			handler.SetTimeFunc(DefaultTime)
+			if tt.timeFunc != nil {
+				handler.SetTimeFunc(tt.timeFunc)
+			}
+			if tt.precision > 0 {
+				handler.SetTimePrecision(tt.precision)
+			}
+			parser := NewSeriesParser(handler)
+
+			metrics, err := parser.Parse(tt.input)
+			require.Equal(t, tt.err, err)
+
+			require.Equal(t, len(tt.metrics), len(metrics))
+			for i, expected := range tt.metrics {
+				require.Equal(t, expected.Name(), metrics[i].Name())
+				require.Equal(t, expected.Tags(), metrics[i].Tags())
+			}
+		})
+	}
+}

plugins/parsers/registry.go

@@ -26,11 +26,15 @@ type Parser interface {
 	// Parse takes a byte buffer separated by newlines
 	// ie, `cpu.usage.idle 90\ncpu.usage.busy 10`
 	// and parses it into telegraf metrics
+	//
+	// Must be thread-safe.
 	Parse(buf []byte) ([]telegraf.Metric, error)
 
 	// ParseLine takes a single string metric
 	// ie, "cpu.usage.idle 90"
 	// and parses it into a telegraf metric.
+	//
+	// Must be thread-safe.
 	ParseLine(line string) (telegraf.Metric, error)
 
 	// SetDefaultTags tells the parser to add all of the given tags
@@ -107,9 +111,15 @@ func NewParser(config *Config) (Parser, error) {
 		parser, err = NewCollectdParser(config.CollectdAuthFile,
 			config.CollectdSecurityLevel, config.CollectdTypesDB)
 	case "dropwizard":
-		parser, err = NewDropwizardParser(config.DropwizardMetricRegistryPath,
+		parser, err = NewDropwizardParser(
-			config.DropwizardTimePath, config.DropwizardTimeFormat, config.DropwizardTagsPath, config.DropwizardTagPathsMap, config.DefaultTags,
+			config.DropwizardMetricRegistryPath,
-			config.Separator, config.Templates)
+			config.DropwizardTimePath,
+			config.DropwizardTimeFormat,
+			config.DropwizardTagsPath,
+			config.DropwizardTagPathsMap,
+			config.DefaultTags,
+			config.Separator,
+			config.Templates)
 	default:
 		err = fmt.Errorf("Invalid data format: %s", config.DataFormat)
 	}
@@ -177,17 +187,16 @@ func NewDropwizardParser(
 	templates []string,
 
 ) (Parser, error) {
-	parser := &dropwizard.Parser{
+	parser := dropwizard.NewParser()
-		MetricRegistryPath: metricRegistryPath,
+	parser.MetricRegistryPath = metricRegistryPath
-		TimePath:           timePath,
+	parser.TimePath = timePath
-		TimeFormat:         timeFormat,
+	parser.TimeFormat = timeFormat
-		TagsPath:           tagsPath,
+	parser.TagsPath = tagsPath
-		TagPathsMap:        tagPathsMap,
+	parser.TagPathsMap = tagPathsMap
-		DefaultTags:        defaultTags,
+	parser.DefaultTags = defaultTags
-		Separator:          separator,
+	err := parser.SetTemplates(separator, templates)
-		Templates:          templates,
+	if err != nil {
+		return nil, err
 	}
-	err := parser.InitTemplating()
-
 	return parser, err
 }