Merge pull request #24 from callumj/cj-statsd

Add StatsD support

README.md

@@ -1,6 +1,6 @@
 # Buildkite Metrics
 
-A command-line tool for collecting [Buildkite](https://buildkite.com/) build/job statistics for external metrics systems. Currently [AWS Cloudwatch](http://aws.amazon.com/cloudwatch/) is supported.
+A command-line tool for collecting [Buildkite](https://buildkite.com/) build/job statistics for external metrics systems. Currently [AWS Cloudwatch](http://aws.amazon.com/cloudwatch/) and [StatsD](https://github.com/etsy/statsd) are supported.
 
 [![Build status](https://badge.buildkite.com/80d04fcde3a306bef44e77aadb1f1ffdc20ebb3c8f1f585a60.svg)](https://buildkite.com/buildkite/buildkite-metrics)
 
@@ -12,6 +12,13 @@ Either download the latest binary from [buildkite-metrics/buildkite-metrics-Linu
 go get github.com/buildkite/buildkite-metrics
 ```
 
+### Backends
+
+By default metrics will be submitted to CloudWatch but the backend can be switched to StatsD using the command-line argument `-backend statsd`. The StatsD backend supports the following arguments
+
+* `-statsd-host HOST`: The StatsD host and port (defaults to `127.0.0.1:8125`).
+* `-statsd-tags`: Some StatsD servers like the agent provided by DataDog support tags. If specified, metrics will be tagged by `queue` and `pipeline` otherwise metrics will include the queue/pipeline name in the metric. Only enable this option if you know your StatsD server supports tags.
+
 ## Development
 
 You can build and run the binary tool locally with golang installed:

backend/backends.go

@@ -0,0 +1,8 @@
+package backend
+
+import "github.com/buildkite/buildkite-metrics/collector"
+
+// Backend is a receiver of metrics
+type Backend interface {
+	Collect(r *collector.Result) error
+}

cloudwatch.go → backend/cloudwatch.go

@@ -1,4 +1,4 @@
-package main
+package backend
 
 import (
 	"log"
@@ -9,7 +9,15 @@ import (
 	"github.com/buildkite/buildkite-metrics/collector"
 )
 
-func cloudwatchSend(r *collector.Result) error {
+// CloudWatchBackend sends metrics to AWS CloudWatch
+type CloudWatchBackend struct {
+}
+
+func NewCloudWatchBackend() *CloudWatchBackend {
+	return &CloudWatchBackend{}
+}
+
+func (cb *CloudWatchBackend) Collect(r *collector.Result) error {
 	svc := cloudwatch.New(session.New())
 
 	metrics := []*cloudwatch.MetricDatum{}

backend/statsd.go

@@ -0,0 +1,67 @@
+package backend
+
+import (
+	"github.com/DataDog/datadog-go/statsd"
+	"github.com/buildkite/buildkite-metrics/collector"
+)
+
+// StatsD sends metrics to StatsD (Datadog spec)
+type StatsD struct {
+	client        *statsd.Client
+	tagsSupported bool
+}
+
+func NewStatsDBackend(host string, tagsSupported bool) (*StatsD, error) {
+	c, err := statsd.NewBuffered(host, 100)
+	if err != nil {
+		return nil, err
+	}
+	// prefix every metric with the app name
+	c.Namespace = "buildkite."
+	return &StatsD{
+		client:        c,
+		tagsSupported: tagsSupported,
+	}, nil
+}
+
+func (cb *StatsD) Collect(r *collector.Result) error {
+	for name, value := range r.Totals {
+		if err := cb.client.Gauge(name, float64(value), []string{}, 1.0); err != nil {
+			return err
+		}
+	}
+
+	for queue, counts := range r.Queues {
+		for name, value := range counts {
+			var finalName string
+			tags := []string{}
+			if cb.tagsSupported {
+				finalName = "queues." + name
+				tags = []string{"queue:" + queue}
+			} else {
+				finalName = "queues." + queue + "." + name
+			}
+			if err := cb.client.Gauge(finalName, float64(value), tags, 1.0); err != nil {
+				return err
+			}
+		}
+	}
+
+	for pipeline, counts := range r.Pipelines {
+		for name, value := range counts {
+			var finalName string
+			tags := []string{}
+			if cb.tagsSupported {
+				finalName = "pipeline." + name
+				tags = []string{"pipeline:" + pipeline}
+			} else {
+				finalName = "pipeline." + pipeline + "." + name
+			}
+			if err := cb.client.Gauge(finalName, float64(value), tags, 1.0); err != nil {
+				return err
+			}
+		}
+	}
+
+	return nil
+}

lambda.go

@@ -5,8 +5,10 @@ import (
 	"encoding/json"
 	"log"
 	"os"
+	"strings"
 	"time"
 
+	"github.com/buildkite/buildkite-metrics/backend"
 	"github.com/buildkite/buildkite-metrics/collector"
 	"github.com/eawsy/aws-lambda-go/service/lambda/runtime"
 	"gopkg.in/buildkite/go-buildkite.v2/buildkite"
@@ -18,6 +20,7 @@ func handle(evt json.RawMessage, ctx *runtime.Context) (interface{}, error) {
 
 	org := os.Getenv("BUILDKITE_ORG")
 	token := os.Getenv("BUILDKITE_TOKEN")
+	backendOpt := os.Getenv("BUILDKITE_BACKEND")
 
 	config, err := buildkite.NewTokenConfig(token, false)
 	if err != nil {
@@ -32,14 +35,24 @@ func handle(evt json.RawMessage, ctx *runtime.Context) (interface{}, error) {
 		Historical: time.Hour * 24,
 	})
 
+	var bk backend.Backend
+	if backendOpt == "statsd" {
+		bk, err = backend.NewStatsDBackend(os.Getenv("STATSD_HOST"), strings.ToLower(os.Getenv("STATSD_TAGS")) == "true")
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		bk = &backend.CloudWatchBackend{}
+	}
+
 	res, err := col.Collect()
 	if err != nil {
 		return nil, err
 	}
 
 	res.Dump()
 
-	err = cloudwatchSend(res)
+	err = bk.Collect(res)
 	if err != nil {
 		return nil, err
 	}

main.go

@@ -6,15 +6,19 @@ import (
 	"io/ioutil"
 	"log"
 	"os"
+	"strings"
 	"time"
 
+	"github.com/buildkite/buildkite-metrics/backend"
 	"github.com/buildkite/buildkite-metrics/collector"
 	"gopkg.in/buildkite/go-buildkite.v2/buildkite"
 )
 
 // Version is passed in via ldflags
 var Version string
 
+var bk backend.Backend
+
 func main() {
 	var (
 		accessToken = flag.String("token", "", "A Buildkite API Access Token")
@@ -26,6 +30,11 @@ func main() {
 		quiet       = flag.Bool("quiet", false, "Only print errors")
 		dryRun      = flag.Bool("dry-run", false, "Whether to only print metrics")
 
+		// backend config
+		backendOpt = flag.String("backend", "cloudwatch", "Specify the backend to send metrics to: cloudwatch, statsd")
+		statsdHost = flag.String("statsd-host", "127.0.0.1:8125", "Specify the StatsD server")
+		statsdTags = flag.Bool("statsd-tags", false, "Whether your StatsD server supports tagging like Datadog")
+
 		// filters
 		queue = flag.String("queue", "", "Only include a specific queue")
 	)
@@ -47,6 +56,21 @@ func main() {
 		os.Exit(1)
 	}
 
+	lowerBackendOpt := strings.ToLower(*backendOpt)
+	if lowerBackendOpt == "cloudwatch" {
+		bk = backend.NewCloudWatchBackend()
+	} else if lowerBackendOpt == "statsd" {
+		var err error
+		bk, err = backend.NewStatsDBackend(*statsdHost, *statsdTags)
+		if err != nil {
+			fmt.Printf("Error starting StatsD, err: %v\n", err)
+			os.Exit(1)
+		}
+	} else {
+		fmt.Println("Must provide a supported backend: cloudwatch, statsd")
+		os.Exit(1)
+	}
+
 	if *quiet {
 		log.SetOutput(ioutil.Discard)
 	}
@@ -82,7 +106,7 @@ func main() {
 		}
 
 		if !*dryRun {
-			err = cloudwatchSend(res)
+			err = bk.Collect(res)
 			if err != nil {
 				return err
 			}