Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Moved to using the inbuilt serializer. #1942

Merged
merged 3 commits into from
Dec 20, 2016
Merged

Moved to using the inbuilt serializer. #1942

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
976dcc9
Moved to using the inbuilt serializer.
wolfeidau Oct 24, 2016
81f9a7a
Remove Atomic variable as it is not required.
wolfeidau Oct 26, 2016
2fbca33
Adjusted metric type in line with latest changes.
wolfeidau Dec 20, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 24 additions & 22 deletions plugins/outputs/kinesis/kinesis.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,10 +1,8 @@
package kinesis

import (
"fmt"
"log"
"os"
"sync/atomic"
"time"

"github.com/aws/aws-sdk-go/aws"
Expand All @@ -13,6 +11,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 {
Expand All @@ -26,9 +25,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 = `
Expand All @@ -54,9 +54,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
`
Expand Down Expand Up @@ -125,16 +129,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
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Won't this break existing users that are using a custom format?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am not sure how the custom format came to be but it seemed to me to be a placeholder, it actually contains duplicate data which makes it harder to parse..

That said I would not introduce this without calling out the change in format, I myself had to trim off the swap,map[host:rufus] (see sample below) to parse it using the influx format.

So in summary that this change goes from influx format with prefix containing duplicate data, to just influx format so I think it is justified.

swap,map[host:rufus],swap,host=rufus free=104853504i,in=0i,out=0i,total=104853504i,used=0i,used_percent=0 1476434910000000000
mem,map[host:rufus],mem,host=rufus active=143302656i,available=898482176i,available_percent=92.57988410421336,buffered=41435136i,cached=149790720i,free=729427968i,inactive=65368064i,total=970493952i,used=72011776i,used_percent=7.420115895786644 1476434910000000000

}

func writekinesis(k *KinesisOutput, r []*kinesis.PutRecordsRequestEntry) time.Duration {
Expand All @@ -161,31 +157,37 @@ func writekinesis(k *KinesisOutput, r []*kinesis.PutRecordsRequestEntry) time.Du
}

func (k *KinesisOutput) Write(metrics []telegraf.Metric) error {
var sz uint32 = 0
var sz uint32

if len(metrics) == 0 {
return nil
}

r := []*kinesis.PutRecordsRequestEntry{}

for _, p := range metrics {
atomic.AddUint32(&sz, 1)
for _, metric := range metrics {
sz++

values, err := k.serializer.Serialize(metric)
if err != nil {
return err
}

metric, _ := FormatMetric(k, p)
d := kinesis.PutRecordsRequestEntry{
Data: []byte(metric),
Data: values,
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)
sz = 0
r = nil
}

}

writekinesis(k, r)
Expand Down
39 changes: 0 additions & 39 deletions plugins/outputs/kinesis/kinesis_test.go
View file
Open in desktop

This file was deleted.