Graphite tags parser (#8564)

plugins/parsers/graphite/README.md

@@ -1,7 +1,7 @@
 # Graphite
 
 The Graphite data format translates graphite *dot* buckets directly into
-telegraf measurement names, with a single value field, and without any tags.
+telegraf measurement names, with a single value field, and optional tags.
 By default, the separator is left as `.`, but this can be changed using the
 `separator` argument. For more advanced options, Telegraf supports specifying
 [templates](#templates) to translate graphite buckets into Telegraf metrics.

plugins/parsers/graphite/parser.go

@@ -103,15 +103,17 @@ func (p *GraphiteParser) ParseLine(line string) (telegraf.Metric, error) {
 		return nil, fmt.Errorf("received %q which doesn't have required fields", line)
 	}
 
+	parts := strings.Split(fields[0], ";")
+
 	// decode the name and tags
-	measurement, tags, field, err := p.templateEngine.Apply(fields[0])
+	measurement, tags, field, err := p.templateEngine.Apply(parts[0])
 	if err != nil {
 		return nil, err
 	}
 
 	// Could not extract measurement, use the raw value
 	if measurement == "" {
-		measurement = fields[0]
+		measurement = parts[0]
 	}
 
 	// Parse value.
@@ -147,6 +149,24 @@ func (p *GraphiteParser) ParseLine(line string) (telegraf.Metric, error) {
 			}
 		}
 	}
+
+	// Split name and tags
+	if len(parts) >= 2 {
+		for _, tag := range parts[1:] {
+			tagValue := strings.Split(tag, "=")
+			if len(tagValue) != 2 || len(tagValue[0]) == 0 || len(tagValue[1]) == 0 {
+				continue
+			}
+			if strings.IndexAny(tagValue[0], "!^") != -1 {
+				continue
+			}
+			if strings.Index(tagValue[1], "~") == 0 {
+				continue
+			}
+			tags[tagValue[0]] = tagValue[1]
+		}
+	}
+
 	// Set the default tags on the point if they are not already set
 	for k, v := range p.DefaultTags {
 		if _, ok := tags[k]; !ok {

plugins/parsers/graphite/parser_test.go

@@ -178,6 +178,67 @@ func TestParseLine(t *testing.T) {
 			value: 50,
 			time:  testTime,
 		},
+		{
+			test:        "normal case with tag",
+			input:       `cpu.foo.bar;tag1=value1 50 ` + strTime,
+			template:    "measurement.foo.bar",
+			measurement: "cpu",
+			tags: map[string]string{
+				"foo":  "foo",
+				"bar":  "bar",
+				"tag1": "value1",
+			},
+			value: 50,
+			time:  testTime,
+		},
+		{
+			test:        "wrong tag names",
+			input:       `cpu.foo.bar;tag!1=value1;tag^2=value2 50 ` + strTime,
+			template:    "measurement.foo.bar",
+			measurement: "cpu",
+			tags: map[string]string{
+				"foo": "foo",
+				"bar": "bar",
+			},
+			value: 50,
+			time:  testTime,
+		},
+		{
+			test:        "empty tag name",
+			input:       `cpu.foo.bar;=value1 50 ` + strTime,
+			template:    "measurement.foo.bar",
+			measurement: "cpu",
+			tags: map[string]string{
+				"foo": "foo",
+				"bar": "bar",
+			},
+			value: 50,
+			time:  testTime,
+		},
+		{
+			test:        "wrong tag value",
+			input:       `cpu.foo.bar;tag1=~value1 50 ` + strTime,
+			template:    "measurement.foo.bar",
+			measurement: "cpu",
+			tags: map[string]string{
+				"foo": "foo",
+				"bar": "bar",
+			},
+			value: 50,
+			time:  testTime,
+		},
+		{
+			test:        "empty tag value",
+			input:       `cpu.foo.bar;tag1= 50 ` + strTime,
+			template:    "measurement.foo.bar",
+			measurement: "cpu",
+			tags: map[string]string{
+				"foo": "foo",
+				"bar": "bar",
+			},
+			value: 50,
+			time:  testTime,
+		},
 		{
 			test:        "metric only with float value",
 			input:       `cpu 50.554 ` + strTime,
@@ -279,6 +340,20 @@ func TestParse(t *testing.T) {
 			value: 50,
 			time:  testTime,
 		},
+		{
+			test:        "normal case with tag",
+			input:       []byte(`cpu.foo.bar;tag1=value1 50 ` + strTime),
+			template:    "measurement.foo.bar",
+			measurement: "cpu",
+			tags: map[string]string{
+				"foo":  "foo",
+				"bar":  "bar",
+				"tag1": "value1",
+			},
+			value: 50,
+			time:  testTime,
+		},
+
 		{
 			test:        "metric only with float value",
 			input:       []byte(`cpu 50.554 ` + strTime),