storage_dogstatsd.go

package main

import (
	"context"
	"fmt"
	"log"
	"sync"

	"github.com/DataDog/datadog-go/statsd"
)

// DogstatsdConfig describes the YAML-provided configuration for a Datadog
// DogstatsD storage backend
type DogstatsdConfig struct {
	Host      string `yaml:"host"`
	Port      int    `yaml:"port"`
	Namespace string `yaml:"metric-namespace"`
}

// DogstatsdStorage holds the configuration for a DogstatsD storage backend
type DogstatsdStorage struct {
	DogstatsdConn *statsd.Client
	Namespace     string
}

// StartStorageEngine creates a goroutine loop to receive metrics and send
// them off to dogstatsd
func (d DogstatsdStorage) StartStorageEngine(ctx context.Context, wg *sync.WaitGroup) (chan<- Metric, chan<- Event) {
	// Datadog storage supports both metrics and events, so we'll initialize both channels
	metricChan := make(chan Metric, 10)
	eventChan := make(chan Event, 10)

	go d.processMetricsAndEvents(ctx, wg, metricChan, eventChan)

	return metricChan, eventChan
}

func (d DogstatsdStorage) processMetricsAndEvents(ctx context.Context, wg *sync.WaitGroup, mchan <-chan Metric, echan <-chan Event) {
	wg.Add(1)
	defer wg.Done()

	for {
		select {
		case m := <-mchan:
			err := d.sendMetric(m)
			if err != nil {
				log.Println(err)
			}
		case e := <-echan:
			err := d.sendEvent(e)
			if err != nil {
				log.Println(err)
			}
		case <-ctx.Done():
			log.Println("Cancellation request recieved.  Cancelling metrics processor.")
			return
		}
	}
}

// sendMetric sends a metric value to dogstatsd
func (d DogstatsdStorage) sendMetric(m Metric) error {
	var metricName string

	if d.Namespace == "" {
		metricName = fmt.Sprintf("crabby.%v.%v", m.Job, m.Timing)
	} else {
		metricName = fmt.Sprintf("%v.%v.%v", d.Namespace, m.Job, m.Timing)
	}

	dogTags := makeDogstatsdTagsFromTagsMap(m.Tags)

	err := d.DogstatsdConn.TimeInMilliseconds(metricName, m.Value, dogTags, 1)
	if err != nil {
		log.Printf("Could not send metric %v: %v\n", metricName, err)
		return err
	}

	return nil
}

// sendEvent sends an event (as a service check) to the Datadog API endpoint
func (d DogstatsdStorage) sendEvent(e Event) error {

	var eventName string

	if d.Namespace == "" {
		eventName = fmt.Sprintf("crabby.%v", e.Name)
	} else {
		eventName = fmt.Sprintf("%v.%v", d.Namespace, e.Name)
	}

	// While Crabby calls this an "event", it's really a "service check" in
	// Datadog parlance.  Datadog does have the concept of "events" but it's
	// more difficult to set up monitoring for events than it is service checks.
	// With service checks, we can send the status with every check.  We just
	// set the Status field to indicate whether things are OK (response code 1xx/2xx/3xx)
	// or are failing (response code 4xx/5xx)
	sc := &statsd.ServiceCheck{
		Name:    eventName,
		Message: fmt.Sprintf("%v is returning a HTTP status code of %v", e.Name, e.ServerStatus),
	}

	if (e.ServerStatus < 400) && (e.ServerStatus > 0) {
		sc.Status = statsd.Ok
	} else {
		sc.Status = statsd.Critical
	}

	sc.Tags = makeDogstatsdTagsFromTagsMap(e.Tags)

	err := d.DogstatsdConn.ServiceCheck(sc)

	return err
}

// NewDogstatsdStorage sets up a new Dogstatsd storage backend
func NewDogstatsdStorage(c ServiceConfig) DogstatsdStorage {
	var err error
	d := DogstatsdStorage{}

	d.Namespace = c.Storage.Dogstatsd.Namespace

	d.DogstatsdConn, err = statsd.New(fmt.Sprint(c.Storage.Dogstatsd.Host, ":", c.Storage.Dogstatsd.Port))
	if err != nil {
		log.Println("Warning: could not create dogstatsd connection", err)
	}

	return d
}

func makeDogstatsdTagsFromTagsMap(tags map[string]string) []string {
	var dogTags []string

	for k, v := range tags {
		dogTags = append(dogTags, fmt.Sprintf("%v=%v", k, v))
	}

	return dogTags
}