Parser processor

plugins/processors/all/all.go

@@ -4,6 +4,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/processors/converter"
 	_ "github.com/influxdata/telegraf/plugins/processors/enum"
 	_ "github.com/influxdata/telegraf/plugins/processors/override"
+	_ "github.com/influxdata/telegraf/plugins/processors/parser"
 	_ "github.com/influxdata/telegraf/plugins/processors/printer"
 	_ "github.com/influxdata/telegraf/plugins/processors/regex"
 	_ "github.com/influxdata/telegraf/plugins/processors/topk"

plugins/processors/parser/README.md

@@ -0,0 +1,54 @@
+# Parser Processor Plugin
+This plugin parses defined fields containing the specified data format.
+
+## Configuration
+```toml
+[[processors.parser]]
+  ## specify the name of the field[s] whose value will be parsed
+  parse_fields = ["message"]
+
+  ## specify what to do with the original message. [merge|replace|keep] default=keep
+  original = "merge"
+
+  [processors.parser.config]
+    data_format = "json"
+    ## additional configurations for parser go here
+    tag_keys = ["verb", "request"]
+```
+
+### Tags:
+
+User specified tags may be added by this processor.
+
+### Example Config:
+```toml
+[[inputs.exec]]
+  data_format = "influx"
+  commands = [
+    "echo -en 'thing,host=\"mcfly\" message=\"{\\\"verb\\\":\\\"GET\\\",\\\"request\\\":\\\"/time/to/awesome\\\"}\" 1519652321000000000'"
+  ]
+
+[[processors.parser]]
+  ## specify the name of the field[s] whose value will be parsed
+  parse_fields = ["message"]
+
+  ## specify what to do with the original message. [merge|replace|keep] default=keep
+  original = "merge"
+
+  [processors.parser.config]
+    data_format = "json"
+    ## additional configurations for parser go here
+    tag_keys = ["verb", "request"]
+
+[[outputs.file]]
+  files = ["stdout"]
+```
+
+### Example Output [original=merge]:
+```
+# input = nginx_requests,host="mcfly" message="{\"verb\":\"GET\",\"request\":\"/time/to/awesome\"}" 1519652321000000000
+nginx_requests,host="mcfly",verb="GET",request="/time/to/awesome" message="{\"verb\":\"GET\",\"request\":\"/time/to/awesome\"}" 1519652321000000000
+```
+
+### Caveats
+While this plugin is functional, it may not work in *every* scenario. For the above example, "keep" and "replace" fail because the parsed field produces a metric with no fields. This leads to errors when serializing the output.

plugins/processors/parser/parser.go

@@ -0,0 +1,129 @@
+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 `toml:"config"`
+	Original    string         `toml:"original"` // merge, replace, or keep (default)
+	ParseFields []string       `toml:"parse_fields"`
+	Parser      parsers.Parser `toml:"parser"`
+}
+
+// holds a default sample config
+var SampleConfig = `
+  ## specify the name of the field[s] whose value will be parsed
+  parse_fields = []
+
+  ## specify what to do with the original message. [merge|replace|keep] default=keep
+  original = "keep"
+
+  [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
+		}
+	}
+
+	rMetrics := []telegraf.Metric{}
+	if p.Original != "replace" {
+		rMetrics = metrics
+	}
+
+	name := ""
+
+	for _, metric := range metrics {
+		if n := metric.Name(); n != "" {
+			name = n
+		}
+		sMetrics := []telegraf.Metric{}
+		for _, key := range p.ParseFields {
+			if value, ok := metric.Fields()[key]; ok {
+				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)
+					switch p.Original {
+					case "keep":
+						return metrics
+					case "merge":
+						nMetrics = metrics
+					}
+				}
+				sMetrics = append(sMetrics, nMetrics...)
+			} else {
+				fmt.Println("key not found", key)
+			}
+		}
+		rMetrics = append(rMetrics, p.mergeTagsFields(sMetrics...)...)
+	}
+	if p.Original == "merge" {
+		rMetrics = p.mergeTagsFields(rMetrics...)
+	}
+
+	return p.setName(name, rMetrics...)
+
+}
+
+func (p Parser) setName(name string, metrics ...telegraf.Metric) []telegraf.Metric {
+	if len(metrics) == 0 {
+		return nil
+	}
+
+	for i := range metrics {
+		metrics[i].SetName(name)
+	}
+
+	return metrics
+}
+
+func (p Parser) mergeTagsFields(metrics ...telegraf.Metric) []telegraf.Metric {
+	if len(metrics) == 0 {
+		return nil
+	}
+
+	rMetric := metrics[0]
+	for _, metric := range metrics {
+		for key, field := range metric.Fields() {
+			rMetric.AddField(key, field)
+		}
+		for key, tag := range metric.Tags() {
+			rMetric.AddTag(key, tag)
+		}
+	}
+	return []telegraf.Metric{rMetric}
+}
+
+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{Original: "keep"}
+	})
+}