Add parser processor (#4551)

plugins/processors/parser/parser.go

@@ -0,0 +1,124 @@
+package parser
+
+import (
+	"log"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/processors"
+)
+
+type Parser struct {
+	parsers.Config
+	DropOriginal bool     `toml:"drop_original"`
+	Merge        string   `toml:"merge"`
+	ParseFields  []string `toml:"parse_fields"`
+	Parser       parsers.Parser
+}
+
+var SampleConfig = `
+  ## The name of the fields whose value will be parsed.
+  parse_fields = []
+
+  ## If true, incoming metrics are not emitted.
+  drop_original = false
+
+  ## If set to override, emitted metrics will be merged by overriding the
+  ## original metric using the newly parsed metrics.
+  merge = "override"
+
+  ## The dataformat to be read from files
+  ## Each data format has its own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_INPUT.md
+  data_format = "influx"
+`
+
+func (p *Parser) SampleConfig() string {
+	return SampleConfig
+}
+
+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
+		}
+	}
+
+	results := []telegraf.Metric{}
+
+	for _, metric := range metrics {
+		newMetrics := []telegraf.Metric{}
+		if !p.DropOriginal {
+			newMetrics = append(newMetrics, metric)
+		}
+
+		for _, key := range p.ParseFields {
+			for _, field := range metric.FieldList() {
+				if field.Key == key {
+					switch value := field.Value.(type) {
+					case string:
+						fromFieldMetric, err := p.parseField(value)
+						if err != nil {
+							log.Printf("E! [processors.parser] could not parse field %s: %v", key, err)
+						}
+
+						for _, m := range fromFieldMetric {
+							if m.Name() == "" {
+								m.SetName(metric.Name())
+							}
+						}
+
+						// multiple parsed fields shouldn't create multiple
+						// metrics so we'll merge tags/fields down into one
+						// prior to returning.
+						newMetrics = append(newMetrics, fromFieldMetric...)
+					default:
+						log.Printf("E! [processors.parser] field '%s' not a string, skipping", key)
+					}
+				}
+			}
+		}
+
+		if len(newMetrics) == 0 {
+			continue
+		}
+
+		if p.Merge == "override" {
+			results = append(results, merge(newMetrics[0], newMetrics[1:]))
+		} else {
+			results = append(results, newMetrics...)
+		}
+	}
+	return results
+}
+
+func merge(base telegraf.Metric, metrics []telegraf.Metric) telegraf.Metric {
+	for _, metric := range metrics {
+		for _, field := range metric.FieldList() {
+			base.AddField(field.Key, field.Value)
+		}
+		for _, tag := range metric.TagList() {
+			base.AddTag(tag.Key, tag.Value)
+		}
+		base.SetName(metric.Name())
+	}
+	return base
+}
+
+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{DropOriginal: false}
+	})
+}