Parser processor

plugins/processors/parser/README.md

@@ -0,0 +1,14 @@
+# Parser Processor Plugin
+This plugin parses through fields in a pre-formatted string
+
+## Configuration
+```var SampleConfig = `
+
+[processors.parser]
+  ## specify the name of the field[s] whose value will be parsed
+  parse_fields = []
+
+  data_format = "logfmt"
+  ## additional configurations for parser go here
+`
+```

plugins/processors/parser/parser.go

@@ -0,0 +1,73 @@
+package parser
+
+import (
+	"fmt"
+	"log"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/processors"
+)
+
+type Parser struct {
+	config      parsers.Config
+	parseFields []string `toml:"parse_fields"`
+	Parser      parsers.Parser
+}
+
+// holds a default sample config
+var SampleConfig = `
+
+## specify the name of the field[s] whose value will be parsed
+parse_fields = []
+
+[processors.parser.config]
+  data_format = "logfmt"
+  ## additional configurations for parser go here
+`
+
+// returns the default config
+func (p *Parser) SampleConfig() string {
+	return SampleConfig
+}
+
+// returns a brief description of the processor
+func (p *Parser) Description() string {
+	return "Parse a value in a specified field/tag(s) and add the result in a new metric"
+}
+
+func (p *Parser) Apply(metrics ...telegraf.Metric) []telegraf.Metric {
+	if p.Parser == nil {
+		var err error
+		p.Parser, err = parsers.NewParser(&p.config)
+		if err != nil {
+			log.Printf("E! [processors.parser] could not create parser: %v", err)
+			return metrics
+		}
+	}
+
+	for _, metric := range metrics {
+		for _, key := range p.parseFields {
+			value := metric.Fields()[key]
+			strVal := fmt.Sprintf("%v", value)
+			nMetrics, err := p.parseField(strVal)
+			if err != nil {
+				log.Printf("E! [processors.parser] could not parse field %v: %v", key, err)
+				return metrics
+			}
+			metrics = append(metrics, nMetrics...)
+		}
+	}
+	return metrics
+
+}
+
+func (p *Parser) parseField(value string) ([]telegraf.Metric, error) {
+	return p.Parser.Parse([]byte(value))
+}
+
+func init() {
+	processors.Add("parser", func() telegraf.Processor {
+		return &Parser{}
+	})
+}

plugins/processors/parser/parser_test.go

@@ -0,0 +1,198 @@
+package parser
+
+import (
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/metric"
+	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/stretchr/testify/require"
+)
+
+//compares metrics without comparing time
+func compareMetrics(t *testing.T, metrics1 []telegraf.Metric, metrics2 []telegraf.Metric) {
+	for i, m1 := range metrics1 {
+		m2 := metrics2[i]
+		require.True(t, reflect.DeepEqual(m1.Tags(), m2.Tags()))
+		require.True(t, reflect.DeepEqual(m1.Fields(), m2.Fields()))
+		//require.True(t, m1.Name() == m2.Name())
+	}
+}
+
+func Metric(v telegraf.Metric, err error) telegraf.Metric {
+	if err != nil {
+		panic(err)
+	}
+	return v
+}
+
+func TestApply(t *testing.T) {
+	tests := []struct {
+		name        string
+		parseFields []string
+		config      parsers.Config
+		input       telegraf.Metric
+		expected    []telegraf.Metric
+	}{
+		{
+			name:        "parse one field",
+			parseFields: []string{"test_name"},
+			config: parsers.Config{
+				DataFormat: "logfmt",
+			},
+			input: Metric(
+				metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"test_name": `ts=2018-07-24T19:43:40.275Z lvl=info msg="http request" method=POST`,
+					},
+					time.Unix(0, 0))),
+			expected: []telegraf.Metric{
+				Metric(metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"test_name": `ts=2018-07-24T19:43:40.275Z lvl=info msg="http request" method=POST`,
+					},
+					time.Unix(0, 0))),
+				Metric(metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"ts":     "2018-07-24T19:43:40.275Z",
+						"lvl":    "info",
+						"msg":    "http request",
+						"method": "post",
+					},
+					time.Unix(0, 0))),
+			},
+		},
+		{
+			name:        "parse two fields",
+			parseFields: []string{"field_1", "field_2"},
+			config: parsers.Config{
+				DataFormat: "logfmt",
+			},
+			input: Metric(
+				metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"field_1": "ts=2018-07-24T19:43:40.275Z",
+						"field_2": "lvl=info",
+					},
+					time.Unix(0, 0))),
+			expected: []telegraf.Metric{
+				Metric(metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"field_1": "ts=2018-07-24T19:43:40.275Z",
+						"field_2": "lvl=info",
+					},
+					time.Unix(0, 0))),
+				Metric(metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"ts":  "2018-07-24T19:43:40.275Z",
+						"lvl": "info",
+					},
+					time.Unix(0, 0))),
+			},
+		},
+		{
+			name:        "Fail to parse one field but parses other",
+			parseFields: []string{"good", "bad"},
+			config: parsers.Config{
+				DataFormat: "logfmt",
+			},
+			input: Metric(
+				metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"good": "lvl=info",
+						"bad":  "why",
+					},
+					time.Unix(0, 0))),
+			expected: []telegraf.Metric{
+				Metric(metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"good": "lvl=info",
+						"bad":  "why",
+					},
+					time.Unix(0, 0))),
+				Metric(metric.New(
+					"success",
+					map[string]string{},
+					map[string]interface{}{
+						"lvl": "info",
+					},
+					time.Unix(0, 0))),
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		parser := Parser{
+			config:      tt.config,
+			parseFields: tt.parseFields,
+		}
+
+		output := parser.Apply(tt.input)
+
+		compareMetrics(t, output, tt.expected)
+	}
+}
+
+func TestBadApply(t *testing.T) {
+	tests := []struct {
+		name        string
+		parseFields []string
+		config      parsers.Config
+		input       telegraf.Metric
+		expected    []telegraf.Metric
+	}{
+		{
+			name:        "field not found",
+			parseFields: []string{"bad_field"},
+			config: parsers.Config{
+				DataFormat: "logfmt",
+			},
+			input: Metric(
+				metric.New(
+					"bad",
+					map[string]string{},
+					map[string]interface{}{
+						"some_field": 5,
+					},
+					time.Unix(0, 0))),
+			expected: []telegraf.Metric{
+				Metric(metric.New(
+					"bad",
+					map[string]string{},
+					map[string]interface{}{
+						"some_field": 5,
+					},
+					time.Unix(0, 0))),
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		parser := Parser{
+			config:      tt.config,
+			parseFields: tt.parseFields,
+		}
+
+		output := parser.Apply(tt.input)
+
+		compareMetrics(t, output, tt.expected)
+	}
+}