Add datadog/statsd fix #196 (#915)

cmd/collectors.go

@@ -29,9 +29,12 @@ import (
 	"github.com/kelseyhightower/envconfig"
 	"github.com/loadimpact/k6/lib"
 	"github.com/loadimpact/k6/stats/cloud"
+	"github.com/loadimpact/k6/stats/datadog"
 	"github.com/loadimpact/k6/stats/influxdb"
 	jsonc "github.com/loadimpact/k6/stats/json"
 	"github.com/loadimpact/k6/stats/kafka"
+	"github.com/loadimpact/k6/stats/statsd"
+	"github.com/loadimpact/k6/stats/statsd/common"
 	"github.com/pkg/errors"
 	"github.com/spf13/afero"
 )
@@ -41,6 +44,8 @@ const (
 	collectorJSON     = "json"
 	collectorKafka    = "kafka"
 	collectorCloud    = "cloud"
+	collectorStatsD   = "statsd"
+	collectorDatadog  = "datadog"
 )
 
 func parseCollector(s string) (t, arg string) {
@@ -93,6 +98,18 @@ func newCollector(collectorName, arg string, src *lib.SourceData, conf Config) (
 				config = config.Apply(cmdConfig)
 			}
 			return kafka.New(config)
+		case collectorStatsD:
+			config := common.NewConfig().Apply(conf.Collectors.StatsD)
+			if err := envconfig.Process("k6_statsd", &config); err != nil {
+				return nil, err
+			}
+			return statsd.New(config)
+		case collectorDatadog:
+			config := datadog.NewConfig().Apply(conf.Collectors.Datadog)
+			if err := envconfig.Process("k6_datadog", &config); err != nil {
+				return nil, err
+			}
+			return datadog.New(config)
 		default:
 			return nil, errors.Errorf("unknown output type: %s", collectorName)
 		}

cmd/config.go

@@ -28,8 +28,10 @@ import (
 	"github.com/kelseyhightower/envconfig"
 	"github.com/loadimpact/k6/lib"
 	"github.com/loadimpact/k6/stats/cloud"
+	"github.com/loadimpact/k6/stats/datadog"
 	"github.com/loadimpact/k6/stats/influxdb"
 	"github.com/loadimpact/k6/stats/kafka"
+	"github.com/loadimpact/k6/stats/statsd/common"
 	"github.com/shibukawa/configdir"
 	"github.com/spf13/afero"
 	"github.com/spf13/pflag"
@@ -74,6 +76,8 @@ type Config struct {
 		InfluxDB influxdb.Config `json:"influxdb"`
 		Kafka    kafka.Config    `json:"kafka"`
 		Cloud    cloud.Config    `json:"cloud"`
+		StatsD   common.Config   `json:"statsd"`
+		Datadog  datadog.Config  `json:"datadog"`
 	} `json:"collectors"`
 }
 
@@ -97,6 +101,8 @@ func (c Config) Apply(cfg Config) Config {
 	c.Collectors.InfluxDB = c.Collectors.InfluxDB.Apply(cfg.Collectors.InfluxDB)
 	c.Collectors.Cloud = c.Collectors.Cloud.Apply(cfg.Collectors.Cloud)
 	c.Collectors.Kafka = c.Collectors.Kafka.Apply(cfg.Collectors.Kafka)
+	c.Collectors.StatsD = c.Collectors.StatsD.Apply(cfg.Collectors.StatsD)
+	c.Collectors.Datadog = c.Collectors.Datadog.Apply(cfg.Collectors.Datadog)
 	return c
 }
 

lib/options.go

@@ -21,11 +21,13 @@
 package lib
 
 import (
+	"bytes"
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
 	"net"
 	"reflect"
+	"strings"
 
 	"github.com/loadimpact/k6/lib/types"
 	"github.com/loadimpact/k6/stats"
@@ -73,6 +75,20 @@ func (t *TagSet) UnmarshalJSON(data []byte) error {
 	return nil
 }
 
+// UnmarshalText converts the tag list to tagset.
+func (t *TagSet) UnmarshalText(data []byte) error {
+	var list = bytes.Split(data, []byte(","))
+	*t = make(map[string]bool, len(list))
+	for _, key := range list {
+		key := strings.TrimSpace(string(key))
+		if key == "" {
+			continue
+		}
+		(*t)[key] = true
+	}
+	return nil
+}
+
 // Describes a TLS version. Serialised to/from JSON as a string, eg. "tls1.2".
 type TLSVersion int
 

lib/options_test.go

@@ -33,7 +33,8 @@ import (
 	"github.com/loadimpact/k6/lib/types"
 	"github.com/loadimpact/k6/stats"
 	"github.com/stretchr/testify/assert"
-	"gopkg.in/guregu/null.v3"
+	"github.com/stretchr/testify/require"
+	null "gopkg.in/guregu/null.v3"
 )
 
 func TestOptions(t *testing.T) {
@@ -480,3 +481,23 @@ func TestOptionsEnv(t *testing.T) {
 		})
 	}
 }
+
+func TestTagSetTextUnmarshal(t *testing.T) {
+
+	var testMatrix = map[string]map[string]bool{
+		"":                         {},
+		"test":                     {"test": true},
+		"test1,test2":              {"test1": true, "test2": true},
+		"   test1  ,  test2  ":     {"test1": true, "test2": true},
+		"   test1  ,   ,  test2  ": {"test1": true, "test2": true},
+		"   test1  ,,  test2  ,,":  {"test1": true, "test2": true},
+	}
+
+	for input, expected := range testMatrix {
+		var set = new(TagSet)
+		err := set.UnmarshalText([]byte(input))
+		require.NoError(t, err)
+
+		require.Equal(t, (map[string]bool)(*set), expected)
+	}
+}

release notes/upcoming.md

@@ -8,6 +8,19 @@ You can now specify a file for all things logged by `console.log` to get written
 
 Thanks to @cheesedosa for both proposing and implementing this!
 
+### New result outputs: StatsD and Datadog (#915)
+
+You can now output any metrics k6 collects to StatsD or Datadog by running `k6 run --out statsd script.js` or `k6 run --out datadog script.js` respectively. Both are very similar, but Datadog has a concept of metric tags, the key-value metadata pairs that will allow you to distinguish between requests for different URLs, response statuses, different groups, etc.
+
+Some details:
+- By default both outputs send metrics to a local agent listening on `localhost:8125` (currently only UDP is supported as a transport). You can change this address via the `K6_DATADOG_ADDR` or `K6_STATSD_ADDR` environment variables, by setting their values in the format of `address:port`.
+- The new outputs also support adding a `namespace` - a prefix before all the metric names. You can set it via the `K6_DATADOG_NAMESPACE` or `K6_STATSD_NAMESPACE` environment variables respectively. Its default value is `k6.` - notice the dot at the end.
+- You can configure how often data batches are sent via the  `K6_STATSD_PUSH_INTERVAL` / `K6_DATADOG_PUSH_INTEVAL` environment variables. The default value is `1s`.
+- Another performance tweak can be done by changing the default buffer size of 20 through `K6_STATSD_BUFFER_SIZE` / `K6_DATADOG_BUFFER_SIZE`.
+- In the case of Datadog, there is an additional configuration `K6_DATADOG_TAG_BLACKLIST`, which by default is equal to `` (nothing). This is a comma separated list of tags that should *NOT* be sent to Datadog. All other metric tags that k6 emits will be sent.
+
+Thanks to @ivoreis for their work on this!
+
 ### k6/crypto: Random bytes method (#922)
 This feature adds a method to return an array with a number of cryptographically random bytes. It will either return exactly the amount of bytes requested or will throw an exception if something went wrong.
 

stats/datadog/collector.go

@@ -0,0 +1,74 @@
+/*
+ *
+ * k6 - a next-generation load testing tool
+ * Copyright (C) 2019 Load Impact
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package datadog
+
+import (
+	"github.com/loadimpact/k6/lib"
+	"github.com/loadimpact/k6/stats/statsd/common"
+)
+
+type tagHandler lib.TagSet
+
+func (t tagHandler) processTags(tags map[string]string) []string {
+	var res []string
+
+	for key, value := range tags {
+		if value != "" && !t[key] {
+			res = append(res, key+":"+value)
+		}
+	}
+	return res
+}
+
+// Config defines the datadog configuration
+type Config struct {
+	common.Config
+
+	TagBlacklist lib.TagSet `json:"tagBlacklist,omitempty" envconfig:"TAG_BLACKLIST"`
+}
+
+// Apply saves config non-zero config values from the passed config in the receiver.
+func (c Config) Apply(cfg Config) Config {
+	c.Config = c.Config.Apply(cfg.Config)
+
+	if cfg.TagBlacklist != nil {
+		c.TagBlacklist = cfg.TagBlacklist
+	}
+
+	return c
+}
+
+// NewConfig creates a new Config instance with default values for some fields.
+func NewConfig() Config {
+	return Config{
+		Config:       common.NewConfig(),
+		TagBlacklist: lib.GetTagSet(),
+	}
+}
+
+// New creates a new statsd connector client
+func New(conf Config) (*common.Collector, error) {
+	return &common.Collector{
+		Config:      conf.Config,
+		Type:        "datadog",
+		ProcessTags: tagHandler(conf.TagBlacklist).processTags,
+	}, nil
+}

stats/datadog/collector_test.go

@@ -0,0 +1,45 @@
+package datadog
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/loadimpact/k6/lib"
+	"github.com/loadimpact/k6/stats"
+	"github.com/loadimpact/k6/stats/statsd/common"
+	"github.com/loadimpact/k6/stats/statsd/common/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCollector(t *testing.T) {
+	var tagSet = lib.GetTagSet("tag1", "tag2")
+	var handler = tagHandler(tagSet)
+	testutil.BaseTest(t, func(config common.Config) (*common.Collector, error) {
+		return New(NewConfig().Apply(Config{
+			TagBlacklist: tagSet,
+			Config:       config,
+		}))
+	}, func(t *testing.T, containers []stats.SampleContainer, expectedOutput, output string) {
+		var outputLines = strings.Split(output, "\n")
+		var expectedOutputLines = strings.Split(expectedOutput, "\n")
+		for i, container := range containers {
+			for j, sample := range container.GetSamples() {
+				var (
+					expectedTagList    = handler.processTags(sample.GetTags().CloneTags())
+					expectedOutputLine = expectedOutputLines[i*j+i]
+					outputLine         = outputLines[i*j+i]
+					outputWithoutTags  = outputLine
+					outputTagList      = []string{}
+					tagSplit           = strings.LastIndex(outputLine, "|#")
+				)
+
+				if tagSplit != -1 {
+					outputWithoutTags = outputLine[:tagSplit]
+					outputTagList = strings.Split(outputLine[tagSplit+len("|#"):], ",")
+				}
+				require.Equal(t, expectedOutputLine, outputWithoutTags)
+				require.ElementsMatch(t, expectedTagList, outputTagList)
+			}
+		}
+	})
+}

stats/statsd/collector.go

@@ -0,0 +1,33 @@
+/*
+ *
+ * k6 - a next-generation load testing tool
+ * Copyright (C) 2019 Load Impact
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package statsd
+
+import (
+	"github.com/loadimpact/k6/stats/statsd/common"
+)
+
+// New creates a new statsd connector client
+func New(conf common.Config) (*common.Collector, error) {
+	return &common.Collector{
+		Config: conf,
+		Type:   "statsd",
+	}, nil
+}

stats/statsd/collector_test.go

@@ -0,0 +1,16 @@
+package statsd
+
+import (
+	"testing"
+
+	"github.com/loadimpact/k6/stats"
+	"github.com/loadimpact/k6/stats/statsd/common/testutil"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCollector(t *testing.T) {
+	testutil.BaseTest(t, New,
+		func(t *testing.T, _ []stats.SampleContainer, expectedOutput, output string) {
+			require.Equal(t, expectedOutput, output)
+		})
+}