Feature/cloudwatch

Gopkg.toml

@@ -94,3 +94,7 @@
   [[prune.project]]
     name = "github.com/spf13/cobra"
     unused-packages = false
+
+[[constraint]]
+  name = "github.com/aws/aws-sdk-go"
+  version = "1.19.38"

cmd/collectors.go

@@ -24,6 +24,8 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/loadimpact/k6/stats/awscloudwatch"
+
 	"gopkg.in/guregu/null.v3"
 
 	"github.com/kelseyhightower/envconfig"
@@ -40,12 +42,13 @@ import (
 )
 
 const (
-	collectorInfluxDB = "influxdb"
-	collectorJSON     = "json"
-	collectorKafka    = "kafka"
-	collectorCloud    = "cloud"
-	collectorStatsD   = "statsd"
-	collectorDatadog  = "datadog"
+	collectorInfluxDB   = "influxdb"
+	collectorJSON       = "json"
+	collectorKafka      = "kafka"
+	collectorCloud      = "cloud"
+	collectorStatsD     = "statsd"
+	collectorDatadog    = "datadog"
+	collectorCloudWatch = "cloudwatch"
 )
 
 func parseCollector(s string) (t, arg string) {
@@ -110,6 +113,8 @@ func newCollector(collectorName, arg string, src *lib.SourceData, conf Config) (
 				return nil, err
 			}
 			return datadog.New(config)
+		case collectorCloudWatch:
+			return awscloudwatch.New(awscloudwatch.ClientFactory(arg))
 		default:
 			return nil, errors.Errorf("unknown output type: %s", collectorName)
 		}

release notes/upcoming.md

@@ -6,6 +6,14 @@ TODO: Intro
 
 Now all http methods have an additional param called `compression` that will make k6 compress the body before sending it. It will also correctly set both `Content-Encoding` and `Content-Length`, unless they were manually set in the request `headers` by the user. The current supported algorithms are `deflate` and `gzip` and any combination of the two separated by a comma (`,`).
 
+### New result outputs: CloudWatch (#1032)
+
+You can now output any metrics k6 collects to CloudWatch by running `k6 run --out cloudwatch=metrics_namespace`.
+Every tag associated with a metric will be sent as a [dimension](https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/cloudwatch_concepts.html#Dimension).
+CloudWatch **only accepts up to 10 dimensions**, the rest of the tags will be ignored. You may want to use **system-tags** option
+to reduce produced tags.
+
+
 ## Bugs fixed!
 
 * JS: Many fixes for `open()`: (#965)
@@ -19,4 +27,4 @@ Now all http methods have an additional param called `compression` that will mak
 ## Internals
 
 * JS: VU initialization time and memory usage has been significantly decreased by caching the parsed version of the core-js library. Thanks, @matlockx! (#1038)
-* CI: removed the gometalinter check in CircleCI, since that project [was deprecated](https://github.com/alecthomas/gometalinter/issues/590) and now exclusively rely on [golangci-lint](https://github.com/golangci/golangci-lint). (#1039)
+* CI: removed the gometalinter check in CircleCI, since that project [was deprecated](https://github.com/alecthomas/gometalinter/issues/590) and now exclusively rely on [golangci-lint](https://github.com/golangci/golangci-lint). (#1039)

stats/awscloudwatch/cloudwatch.go

@@ -0,0 +1,115 @@
+package awscloudwatch
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/aws/aws-sdk-go/service/cloudwatch"
+	"github.com/aws/aws-sdk-go/service/cloudwatch/cloudwatchiface"
+	"github.com/pkg/errors"
+	"github.com/sirupsen/logrus"
+)
+
+type client struct {
+	cloudwatchiface.CloudWatchAPI
+	namespace string
+	endpoint  string
+}
+
+// ClientFactory returns a function that creates the AWS CloudWatch client
+func ClientFactory(namespace string) func() (cloudWatchClient, error) {
+	return func() (cloudWatchClient, error) {
+		cw, err := newCloudWatchClient()
+		if err != nil {
+			return nil, err
+		}
+
+		return &client{
+			CloudWatchAPI: cw,
+			namespace:     namespace,
+			endpoint:      cw.Endpoint,
+		}, nil
+	}
+}
+
+// newCloudWatchClient creates a new CloudWatch client
+func newCloudWatchClient() (*cloudwatch.CloudWatch, error) {
+	sess, err := session.NewSession()
+	if err != nil {
+		return nil, err
+	}
+
+	return cloudwatch.New(sess), nil
+}
+
+const maxMetricsPerCall = 20
+
+// reportSamples reports samples to CloudWatch.
+// It sends samples in batches of max 20, which is the upper limit of metrics
+// accepted per request by CloudWatch
+func (c *client) reportSamples(samples []*sample) error {
+	samplesSent := 0
+	var lastError error
+
+	for samplesSent < len(samples) {
+		input := &cloudwatch.PutMetricDataInput{Namespace: &c.namespace}
+		upperLimit := samplesSent + maxMetricsPerCall
+		if len(samples) < upperLimit {
+			upperLimit = len(samples)
+		}
+
+		for _, s := range samples[samplesSent:upperLimit] {
+			input.MetricData = append(input.MetricData, toMetricDatum(s))
+			samplesSent++
+		}
+
+		_, err := c.PutMetricData(input)
+		if err != nil {
+			logrus.WithError(err).Warn("Error sending metrics to CloudWatch")
+			lastError = err
+		}
+	}
+
+	if lastError != nil {
+		return errors.Wrap(lastError, "Error sending metrics")
+	}
+
+	return nil
+}
+
+func (c *client) address() string {
+	return c.endpoint
+}
+
+const maxNumberOfDimensions = 10
+
+func toMetricDatum(s *sample) *cloudwatch.MetricDatum {
+	datum := &cloudwatch.MetricDatum{
+		Value:      &s.Value,
+		MetricName: &s.Metric,
+		Timestamp:  &s.Time,
+	}
+
+	var dims []*cloudwatch.Dimension
+
+	for name, value := range s.Tags {
+		if len(dims) == maxNumberOfDimensions {
+			logrus.WithField("tags", s.Tags).
+				WithField("dimensions_included", dims).
+				Warnf("More than 10 tags, just 10 will be reported to CloudWatch")
+			break
+		}
+
+		if value != "" {
+			dims = append(dims, &cloudwatch.Dimension{
+				Name:  aws.String(name),
+				Value: aws.String(value),
+			})
+		}
+	}
+
+	if len(dims) > 0 {
+		datum.Dimensions = dims
+	}
+
+	return datum
+}