Moved to using the inbuilt serializer.

Author: wolfeidau

plugins/outputs/kinesis/kinesis.go

@@ -1,7 +1,6 @@
 package kinesis
 
 import (
-	"fmt"
 	"log"
 	"os"
 	"sync/atomic"
@@ -13,6 +12,7 @@ import (
 	"github.com/influxdata/telegraf"
 	internalaws "github.com/influxdata/telegraf/internal/config/aws"
 	"github.com/influxdata/telegraf/plugins/outputs"
+	"github.com/influxdata/telegraf/plugins/serializers"
 )
 
 type KinesisOutput struct {
@@ -26,9 +26,10 @@ type KinesisOutput struct {
 
 	StreamName   string `toml:"streamname"`
 	PartitionKey string `toml:"partitionkey"`
-	Format       string `toml:"format"`
 	Debug        bool   `toml:"debug"`
 	svc          *kinesis.Kinesis
+
+	serializer serializers.Serializer
 }
 
 var sampleConfig = `
@@ -54,9 +55,13 @@ var sampleConfig = `
   streamname = "StreamName"
   ## PartitionKey as used for sharding data.
   partitionkey = "PartitionKey"
-  ## format of the Data payload in the kinesis PutRecord, supported
-  ## String and Custom.
-  format = "string"
+
+  ## Data format to output.
+  ## Each data format has it's own unique set of configuration options, read
+  ## more about them here:
+  ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+  data_format = "influx"
+
   ## debug will show upstream aws messages.
   debug = false
 `
@@ -125,16 +130,8 @@ func (k *KinesisOutput) Close() error {
 	return nil
 }
 
-func FormatMetric(k *KinesisOutput, point telegraf.Metric) (string, error) {
-	if k.Format == "string" {
-		return point.String(), nil
-	} else {
-		m := fmt.Sprintf("%+v,%+v,%+v",
-			point.Name(),
-			point.Tags(),
-			point.String())
-		return m, nil
-	}
+func (k *KinesisOutput) SetSerializer(serializer serializers.Serializer) {
+	k.serializer = serializer
 }
 
 func writekinesis(k *KinesisOutput, r []*kinesis.PutRecordsRequestEntry) time.Duration {
@@ -169,23 +166,30 @@ func (k *KinesisOutput) Write(metrics []telegraf.Metric) error {
 
 	r := []*kinesis.PutRecordsRequestEntry{}
 
-	for _, p := range metrics {
+	for _, metric := range metrics {
 		atomic.AddUint32(&sz, 1)
 
-		metric, _ := FormatMetric(k, p)
-		d := kinesis.PutRecordsRequestEntry{
-			Data:         []byte(metric),
-			PartitionKey: aws.String(k.PartitionKey),
+		values, err := k.serializer.Serialize(metric)
+		if err != nil {
+			return err
 		}
-		r = append(r, &d)
-
-		if sz == 500 {
-			// Max Messages Per PutRecordRequest is 500
-			elapsed := writekinesis(k, r)
-			log.Printf("E! Wrote a %+v point batch to Kinesis in %+v.\n", sz, elapsed)
-			atomic.StoreUint32(&sz, 0)
-			r = nil
+
+		for _, metric := range values {
+			d := kinesis.PutRecordsRequestEntry{
+				Data:         []byte(metric),
+				PartitionKey: aws.String(k.PartitionKey),
+			}
+			r = append(r, &d)
+
+			if sz == 500 {
+				// Max Messages Per PutRecordRequest is 500
+				elapsed := writekinesis(k, r)
+				log.Printf("E! Wrote a %+v point batch to Kinesis in %+v.\n", sz, elapsed)
+				atomic.StoreUint32(&sz, 0)
+				r = nil
+			}
 		}
+
 	}
 
 	writekinesis(k, r)