Kinesis compressed metrics

Gopkg.toml

@@ -265,3 +265,7 @@
 [[constraint]]
   name = "github.com/go-logfmt/logfmt"
   version = "0.4.0"
+
+[[constraint]]
+  name = "github.com/golang/snappy"
+  revision = "2a8bb927dd31d8daada140a5d09578521ce5c36a"

plugins/outputs/kinesis/README.md

@@ -1,4 +1,4 @@
-## Amazon Kinesis Output for Telegraf
+# Amazon Kinesis Output for Telegraf
 
 This is an experimental plugin that is still in the early stages of development. It will batch up all of the Points
 in one Put request to Kinesis. This should save the number of API requests by a considerable level.
@@ -13,16 +13,42 @@ maybe useful for users to review Amazons official documentation which is availab
 
 This plugin uses a credential chain for Authentication with the Kinesis API endpoint. In the following order the plugin
 will attempt to authenticate.
+
 1. Assumed credentials via STS if `role_arn` attribute is specified (source credentials are evaluated from subsequent rules)
 2. Explicit credentials from `access_key`, `secret_key`, and `token` attributes
 3. Shared profile from `profile` attribute
 4. [Environment Variables](https://github.com/aws/aws-sdk-go/wiki/configuring-sdk#environment-variables)
 5. [Shared Credentials](https://github.com/aws/aws-sdk-go/wiki/configuring-sdk#shared-credentials-file)
 6. [EC2 Instance Profile](http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html)
 
+## Example Configuration
+
+```toml
+  access_key = "AWSKEYVALUE"
+  secret_key = "AWSSecretKeyValue"
+  region = "eu-west-1"
+  streamname = "KinesisStreamName"
+  aggregate_metrics = true
+  # Either "gzip", "snappy"
+  compress_metrics_with = gzip
+  partition = { method = "random" }
+  debug = true
+```
 
 ## Config
 
+## AWS Configration
+
+The following AWS configuration variables are available and map directly to the normal AWS settings. If you don't know what they are then you most likely don't need to touch them.
+
+* access_key
+* secret_key
+* role_arn
+* profile
+* shared_credential_file
+* token
+* endpoint_url
+
 For this output plugin to function correctly the following variables must be configured.
 
 * region
@@ -31,6 +57,7 @@ For this output plugin to function correctly the following variables must be con
 ### region
 
 The region is the Amazon region that you wish to connect to. Examples include but are not limited to
+
 * us-west-1
 * us-west-2
 * us-east-1
@@ -89,3 +116,35 @@ String is defined using the default Point.String() value and translated to []byt
 #### custom
 
 Custom is a string defined by a number of values in the FormatMetric() function.
+
+### aggregate_metrics
+
+This will make the plugin gather the metrics and send them as blocks of metrics in Kinesis records. The number of put requests depends on a few factors.
+
+1. If a random key is in use then a block for each shard in the stream will be created unless there isn't enough metrics then as many blocks as metrics.
+1. Each record will be 1020kb in size + partition key
+
+### compress_metrics_with
+
+`compress_metrics_with` has the following values. If no value is set then compression is skipped.
+
+* gzip
+* snappy
+
+They are explained below.
+
+#### gzip
+
+This will make the plugin compress the data using GZip before the data is shipped to Kinesis.
+GZip is slower than snappy but generally fast enough and gives much better compression. Use GZip in most cases.
+
+If both gzip and snappy are true. GZip wins.
+
+#### snappy
+
+This will make the plugin compress the data using Google's Snappy compression before the data is shipped to Kinesis.
+Snappy is much quicker and would be used if you are taking too long to compress and write before the next flush interval.
+
+### debug
+
+Prints debugging data into the logs.

plugins/outputs/kinesis/compression.go

@@ -0,0 +1,37 @@
+package kinesis
+
+import (
+	"bytes"
+	"compress/gzip"
+	"fmt"
+
+	"github.com/golang/snappy"
+)
+
+var (
+	// gzipCompressionLevel sets the compression level. Tests indicate that 7 gives the best trade off
+	// between speed and compression.
+	gzipCompressionLevel = 7
+)
+
+func gzipMetrics(metrics []byte) ([]byte, error) {
+	var buffer bytes.Buffer
+
+	gzw, err := gzip.NewWriterLevel(&buffer, gzipCompressionLevel)
+	if err != nil {
+		return []byte{}, fmt.Errorf("Compression level is incorrect for gzip")
+	}
+	_, err = gzw.Write(metrics)
+	if err != nil {
+		return []byte{}, fmt.Errorf("There was an error in writing to the gzip writer")
+	}
+	if err := gzw.Close(); err != nil {
+		return []byte{}, fmt.Errorf("There was an error in closing the gzip writer")
+	}
+
+	return buffer.Bytes(), nil
+}
+
+func snappyMetrics(metrics []byte) ([]byte, error) {
+	return snappy.Encode(nil, metrics), nil
+}

plugins/outputs/kinesis/compression_test.go

@@ -0,0 +1,38 @@
+package kinesis
+
+import (
+	"testing"
+	"time"
+)
+
+func TestGoodCompression(t *testing.T) {
+	tests := []string{
+		"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
+		time.Now().String(),
+		`abcdefghijklmnopqrstuvwzyz1234567890@~|\/?><@~#+=!"£$%^&_*(){}[]`,
+	}
+
+	for _, test := range tests {
+		_, err := gzipMetrics([]byte(test))
+		if err != nil {
+			t.Logf("Failed to gzip test data")
+			t.Fail()
+		}
+
+		// Snappy doesn't error, so we can only look for panics
+		snappyMetrics([]byte(test))
+	}
+}
+
+func TestBadGzipCompressionLevel(t *testing.T) {
+	oldlevel := gzipCompressionLevel
+	gzipCompressionLevel = 11
+	defer func() { gzipCompressionLevel = oldlevel }()
+
+	_, err := gzipMetrics([]byte(time.Now().String()))
+	if err == nil {
+		t.Logf("Expect gzip to fail because of a bad compression level")
+		t.Fail()
+	}
+
+}