AWS CloudWatch Output Plugin

plugins/outputs/all/all.go

@@ -3,6 +3,7 @@ package all
 import (
 	_ "github.com/influxdb/telegraf/plugins/outputs/amon"
 	_ "github.com/influxdb/telegraf/plugins/outputs/amqp"
+	_ "github.com/influxdb/telegraf/plugins/outputs/cloudwatch"
 	_ "github.com/influxdb/telegraf/plugins/outputs/datadog"
 	_ "github.com/influxdb/telegraf/plugins/outputs/graphite"
 	_ "github.com/influxdb/telegraf/plugins/outputs/influxdb"

plugins/outputs/cloudwatch/README.md

@@ -0,0 +1,31 @@
+## Amazon CloudWatch Output for Telegraf
+
+This plugin will send points to Amazon CloudWatch.
+
+## Amazon Authentication
+
+This plugin uses a credential chain for Authentication with the CloudWatch API endpoint. In the following order the plugin
+will attempt to authenticate.
+1. [IAMS Role](http://docs.aws.amazon.com/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html)
+2. [Environment Variables](https://github.com/aws/aws-sdk-go/wiki/configuring-sdk)
+3. [Shared Credentials](https://github.com/aws/aws-sdk-go/wiki/configuring-sdk)
+
+## Config
+
+For this output plugin to function correctly the following variables must be configured.
+
+* region
+* namespace
+
+### 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
+* ap-southeast-1
+* ap-southeast-2
+
+### namespace
+
+The namespace used for AWS CloudWatch metrics.

plugins/outputs/cloudwatch/cloudwatch.go

@@ -0,0 +1,187 @@
+package cloudwatch
+
+import (
+	"log"
+	"strings"
+	"time"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/credentials"
+	"github.com/aws/aws-sdk-go/aws/credentials/ec2rolecreds"
+	"github.com/aws/aws-sdk-go/aws/ec2metadata"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/cloudwatch"
+
+	"github.com/influxdb/influxdb/client/v2"
+	"github.com/influxdb/telegraf/plugins/outputs"
+
+	"github.com/meirf/gopart"
+)
+
+type CloudWatch struct {
+	Region    string // AWS Region
+	Namespace string // CloudWatch Metrics Namespace
+	svc       *cloudwatch.CloudWatch
+}
+
+var sampleConfig = `
+  # Amazon REGION
+  region = 'us-east-1'
+
+  # Namespace for the CloudWatch MetricDatums
+  namespace = 'InfluxData/Telegraf'
+`
+
+func (c *CloudWatch) SampleConfig() string {
+	return sampleConfig
+}
+
+func (c *CloudWatch) Description() string {
+	return "Configuration for AWS CloudWatch output."
+}
+
+func (c *CloudWatch) Connect() error {
+	Config := &aws.Config{
+		Region: aws.String(c.Region),
+		Credentials: credentials.NewChainCredentials(
+			[]credentials.Provider{
+				&ec2rolecreds.EC2RoleProvider{Client: ec2metadata.New(session.New())},
+				&credentials.EnvProvider{},
+				&credentials.SharedCredentialsProvider{},
+			}),
+	}
+
+	svc := cloudwatch.New(session.New(Config))
+
+	params := &cloudwatch.ListMetricsInput{
+		Namespace: aws.String(c.Namespace),
+	}
+
+	_, err := svc.ListMetrics(params) // Try a read-only call to test connection.
+
+	if err != nil {
+		log.Printf("cloudwatch: Error in ListMetrics API call : %+v \n", err.Error())
+	}
+
+	c.svc = svc
+
+	return err
+}
+
+func (c *CloudWatch) Close() error {
+	return nil
+}
+
+func (c *CloudWatch) Write(points []*client.Point) error {
+	for _, pt := range points {
+		err := c.WriteSinglePoint(pt)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// Write data for a single point. A point can have many fields and one field
+// is equal to one MetricDatum. There is a limit on how many MetricDatums a
+// request can have so we process one Point at a time.
+func (c *CloudWatch) WriteSinglePoint(point *client.Point) error {
+	datums := buildMetricDatum(point)
+
+	const maxDatumsPerCall = 20 // PutMetricData only supports up to 20 data points per call
+
+	for idxRange := range gopart.Partition(len(datums), maxDatumsPerCall) {
+		err := c.WriteToCloudWatch(datums[idxRange.Low:idxRange.High])
+
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *CloudWatch) WriteToCloudWatch(datums []*cloudwatch.MetricDatum) error {
+	params := &cloudwatch.PutMetricDataInput{
+		MetricData: datums,
+		Namespace:  aws.String(c.Namespace),
+	}
+
+	_, err := c.svc.PutMetricData(params)
+
+	if err != nil {
+		log.Printf("CloudWatch: Unable to write to CloudWatch : %+v \n", err.Error())
+	}
+
+	return err
+}
+
+// Make a MetricDatum for each field in a Point. Only fields with values that can be
+// converted to float64 are supported. Non-supported fields are skipped.
+func buildMetricDatum(point *client.Point) []*cloudwatch.MetricDatum {
+	datums := make([]*cloudwatch.MetricDatum, len(point.Fields()))
+	i := 0
+
+	var value float64
+
+	for k, v := range point.Fields() {
+		switch t := v.(type) {
+		case int:
+			value = float64(t)
+		case int32:
+			value = float64(t)
+		case int64:
+			value = float64(t)
+		case float64:
+			value = t
+		case bool:
+			if t {
+				value = 1
+			} else {
+				value = 0
+			}
+		case time.Time:
+			value = float64(t.Unix())
+		default:
+			// Skip unsupported type.
+			datums = datums[:len(datums)-1]
+			continue
+		}
+
+		datums[i] = &cloudwatch.MetricDatum{
+			MetricName: aws.String(strings.Join([]string{point.Name(), k}, "_")),
+			Value:      aws.Float64(value),
+			Dimensions: buildDimensions(point.Tags()),
+			Timestamp:  aws.Time(point.Time()),
+		}
+
+		i += 1
+	}
+
+	return datums
+}
+
+// Make a list of Dimensions by using a Point's tags.
+func buildDimensions(ptTags map[string]string) []*cloudwatch.Dimension {
+
+	dimensions := make([]*cloudwatch.Dimension, len(ptTags))
+	i := 0
+
+	for k, v := range ptTags {
+		dimensions[i] = &cloudwatch.Dimension{
+			Name:  aws.String(k),
+			Value: aws.String(v),
+		}
+
+		i += 1
+	}
+
+	return dimensions
+}
+
+func init() {
+	outputs.Add("cloudwatch", func() outputs.Output {
+		return &CloudWatch{}
+	})
+}