Implementing generic parser plugins and documentation

This constitutes a large change in how we will parse different data
formats going forward (for the plugins that support it)

This is working off @henrypfhu's changes.

plugins/inputs/kafka_consumer/kafka_consumer.go

@@ -1,12 +1,14 @@
 package kafka_consumer
 
 import (
+	"fmt"
 	"log"
 	"strings"
 	"sync"
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/inputs"
+	"github.com/influxdata/telegraf/plugins/parsers"
 
 	"github.com/Shopify/sarama"
 	"github.com/wvanbergen/kafka/consumergroup"
@@ -20,6 +22,8 @@ type Kafka struct {
 	PointBuffer    int
 	Offset         string
 
+	parser parsers.Parser
+
 	sync.Mutex
 
 	// channel for all incoming kafka messages
@@ -36,16 +40,22 @@ type Kafka struct {
 }
 
 var sampleConfig = `
-  # topic(s) to consume
+  ### topic(s) to consume
   topics = ["telegraf"]
-  # an array of Zookeeper connection strings
+  ### an array of Zookeeper connection strings
   zookeeper_peers = ["localhost:2181"]
-  # the name of the consumer group
+  ### the name of the consumer group
   consumer_group = "telegraf_metrics_consumers"
-  # Maximum number of points to buffer between collection intervals
+  ### Maximum number of points to buffer between collection intervals
   point_buffer = 100000
-  # Offset (must be either "oldest" or "newest")
+  ### Offset (must be either "oldest" or "newest")
   offset = "oldest"
+
+  ### Data format to consume. This can be "json", "influx" or "graphite"
+  ### Each data format has it's own unique set of configuration options, read
+  ### more about them here:
+  ### https://github.com/influxdata/telegraf/blob/master/DATA_FORMATS.md
+  data_format = "influx"
 `
 
 func (k *Kafka) SampleConfig() string {
@@ -53,7 +63,11 @@ func (k *Kafka) SampleConfig() string {
 }
 
 func (k *Kafka) Description() string {
-	return "Read line-protocol metrics from Kafka topic(s)"
+	return "Read metrics from Kafka topic(s)"
+}
+
+func (k *Kafka) SetParser(parser parsers.Parser) {
+	k.parser = parser
 }
 
 func (k *Kafka) Start() error {
@@ -96,15 +110,15 @@ func (k *Kafka) Start() error {
 	k.metricC = make(chan telegraf.Metric, k.PointBuffer)
 
 	// Start the kafka message reader
-	go k.parser()
+	go k.receiver()
 	log.Printf("Started the kafka consumer service, peers: %v, topics: %v\n",
 		k.ZookeeperPeers, k.Topics)
 	return nil
 }
 
-// parser() reads all incoming messages from the consumer, and parses them into
+// receiver() reads all incoming messages from the consumer, and parses them into
 // influxdb metric points.
-func (k *Kafka) parser() {
+func (k *Kafka) receiver() {
 	for {
 		select {
 		case <-k.done:
@@ -112,13 +126,14 @@ func (k *Kafka) parser() {
 		case err := <-k.errs:
 			log.Printf("Kafka Consumer Error: %s\n", err.Error())
 		case msg := <-k.in:
-			metrics, err := telegraf.ParseMetrics(msg.Value)
+			metrics, err := k.parser.Parse(msg.Value)
 			if err != nil {
 				log.Printf("Could not parse kafka message: %s, error: %s",
 					string(msg.Value), err.Error())
 			}
 
 			for _, metric := range metrics {
+				fmt.Println(string(metric.Name()))
 				select {
 				case k.metricC <- metric:
 					continue

plugins/inputs/kafka_consumer/kafka_consumer_integration_test.go

@@ -9,6 +9,8 @@ import (
 	"github.com/influxdata/telegraf/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/influxdata/telegraf/plugins/parsers"
 )
 
 func TestReadsMetricsFromKafka(t *testing.T) {
@@ -40,6 +42,8 @@ func TestReadsMetricsFromKafka(t *testing.T) {
 		PointBuffer:    100000,
 		Offset:         "oldest",
 	}
+	p, _ := parsers.NewInfluxParser()
+	k.SetParser(p)
 	if err := k.Start(); err != nil {
 		t.Fatal(err.Error())
 	} else {

plugins/inputs/kafka_consumer/kafka_consumer_test.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/parsers"
 	"github.com/influxdata/telegraf/testutil"
 
 	"github.com/Shopify/sarama"
@@ -12,9 +13,11 @@ import (
 )
 
 const (
-	testMsg     = "cpu_load_short,host=server01 value=23422.0 1422568543702900257"
-	invalidMsg  = "cpu_load_short,host=server01 1422568543702900257"
-	pointBuffer = 5
+	testMsg         = "cpu_load_short,host=server01 value=23422.0 1422568543702900257"
+	testMsgGraphite = "cpu.load.short.graphite 23422 1454780029"
+	testMsgJSON     = "{\"a\": 5, \"b\": {\"c\": 6}}\n"
+	invalidMsg      = "cpu_load_short,host=server01 1422568543702900257"
+	pointBuffer     = 5
 )
 
 func NewTestKafka() (*Kafka, chan *sarama.ConsumerMessage) {
@@ -39,7 +42,8 @@ func TestRunParser(t *testing.T) {
 	k, in := NewTestKafka()
 	defer close(k.done)
 
-	go k.parser()
+	k.parser, _ = parsers.NewInfluxParser()
+	go k.receiver()
 	in <- saramaMsg(testMsg)
 	time.Sleep(time.Millisecond)
 
@@ -51,7 +55,8 @@ func TestRunParserInvalidMsg(t *testing.T) {
 	k, in := NewTestKafka()
 	defer close(k.done)
 
-	go k.parser()
+	k.parser, _ = parsers.NewInfluxParser()
+	go k.receiver()
 	in <- saramaMsg(invalidMsg)
 	time.Sleep(time.Millisecond)
 
@@ -63,7 +68,8 @@ func TestRunParserRespectsBuffer(t *testing.T) {
 	k, in := NewTestKafka()
 	defer close(k.done)
 
-	go k.parser()
+	k.parser, _ = parsers.NewInfluxParser()
+	go k.receiver()
 	for i := 0; i < pointBuffer+1; i++ {
 		in <- saramaMsg(testMsg)
 	}
@@ -77,7 +83,8 @@ func TestRunParserAndGather(t *testing.T) {
 	k, in := NewTestKafka()
 	defer close(k.done)
 
-	go k.parser()
+	k.parser, _ = parsers.NewInfluxParser()
+	go k.receiver()
 	in <- saramaMsg(testMsg)
 	time.Sleep(time.Millisecond)
 
@@ -89,6 +96,45 @@ func TestRunParserAndGather(t *testing.T) {
 		map[string]interface{}{"value": float64(23422)})
 }
 
+// Test that the parser parses kafka messages into points
+func TestRunParserAndGatherGraphite(t *testing.T) {
+	k, in := NewTestKafka()
+	defer close(k.done)
+
+	k.parser, _ = parsers.NewGraphiteParser("_", []string{}, nil)
+	go k.receiver()
+	in <- saramaMsg(testMsgGraphite)
+	time.Sleep(time.Millisecond)
+
+	acc := testutil.Accumulator{}
+	k.Gather(&acc)
+
+	assert.Equal(t, len(acc.Metrics), 1)
+	acc.AssertContainsFields(t, "cpu_load_short_graphite",
+		map[string]interface{}{"value": float64(23422)})
+}
+
+// Test that the parser parses kafka messages into points
+func TestRunParserAndGatherJSON(t *testing.T) {
+	k, in := NewTestKafka()
+	defer close(k.done)
+
+	k.parser, _ = parsers.NewJSONParser("kafka_json_test", []string{}, nil)
+	go k.receiver()
+	in <- saramaMsg(testMsgJSON)
+	time.Sleep(time.Millisecond)
+
+	acc := testutil.Accumulator{}
+	k.Gather(&acc)
+
+	assert.Equal(t, len(acc.Metrics), 1)
+	acc.AssertContainsFields(t, "kafka_json_test",
+		map[string]interface{}{
+			"a":   float64(5),
+			"b_c": float64(6),
+		})
+}
+
 func saramaMsg(val string) *sarama.ConsumerMessage {
 	return &sarama.ConsumerMessage{
 		Key:       nil,