Merge pull request #30 from rcrowley/librato

@mihasya's updates to the Librato client.

librato/client.go

@@ -0,0 +1,103 @@
+package librato
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+)
+
+const Operations = "operations"
+const OperationsShort = "ops"
+
+type LibratoClient struct {
+	Email, Token string
+}
+
+// property strings
+const (
+	// display attributes
+	Color             = "color"
+	DisplayMax        = "display_max"
+	DisplayMin        = "display_min"
+	DisplayUnitsLong  = "display_units_long"
+	DisplayUnitsShort = "display_units_short"
+	DisplayStacked    = "display_stacked"
+	DisplayTransform  = "display_transform"
+	// special gauge display attributes
+	SummarizeFunction = "summarize_function"
+	Aggregate         = "aggregate"
+
+	// metric keys
+	Name        = "name"
+	Period      = "period"
+	Description = "description"
+	DisplayName = "display_name"
+	Attributes  = "attributes"
+
+	// measurement keys
+	MeasureTime = "measure_time"
+	Source      = "source"
+	Value       = "value"
+
+	// special gauge keys
+	Count      = "count"
+	Sum        = "sum"
+	Max        = "max"
+	Min        = "min"
+	SumSquares = "sum_squares"
+
+	// batch keys
+	Counters = "counters"
+	Gauges   = "gauges"
+
+	MetricsPostUrl = "https://metrics-api.librato.com/v1/metrics"
+)
+
+type Measurement map[string]interface{}
+type Metric map[string]interface{}
+
+type Batch struct {
+	Gauges      []Measurement `json:"gauges,omitempty"`
+	Counters    []Measurement `json:"counters,omitempty"`
+	MeasureTime int64         `json:"measure_time"`
+	Source      string        `json:"source"`
+}
+
+func (self *LibratoClient) PostMetrics(batch Batch) (err error) {
+	var (
+		js   []byte
+		req  *http.Request
+		resp *http.Response
+	)
+
+	if len(batch.Counters) == 0 && len(batch.Gauges) == 0 {
+		return nil
+	}
+
+	if js, err = json.Marshal(batch); err != nil {
+		return
+	}
+
+	if req, err = http.NewRequest("POST", MetricsPostUrl, bytes.NewBuffer(js)); err != nil {
+		return
+	}
+
+	req.Header.Set("Content-Type", "application/json")
+	req.SetBasicAuth(self.Email, self.Token)
+
+	if resp, err = http.DefaultClient.Do(req); err != nil {
+		return
+	}
+
+	if resp.StatusCode != http.StatusOK {
+		var body []byte
+		var err error
+		if body, err = ioutil.ReadAll(resp.Body); err != nil {
+			body = []byte(fmt.Sprintf("(could not fetch response body for error: %s)", err))
+		}
+		err = fmt.Errorf("Unable to post to Librato: %d %s %s", resp.StatusCode, resp.Status, string(body))
+	}
+	return
+}

librato/librato.go

@@ -3,35 +3,50 @@ package librato
 import (
 	"fmt"
 	"github.com/rcrowley/go-metrics"
-	"github.com/samuel/go-librato/librato"
 	"log"
 	"math"
+	"regexp"
 	"time"
 )
 
-type LibratoReporter struct {
-	Email, Token string
-	Source       string
-	Interval     time.Duration
-	Registry     metrics.Registry
-	Percentiles  []float64 // percentiles to report on histogram metrics
+// a regexp for extracting the unit from time.Duration.String
+var unitRegexp = regexp.MustCompile("[^\\d]+$")
+
+// a helper that turns a time.Duration into librato display attributes for timer metrics
+func translateTimerAttributes(d time.Duration) (attrs map[string]interface{}) {
+	attrs = make(map[string]interface{})
+	attrs[DisplayTransform] = fmt.Sprintf("x/%d", int64(d))
+	attrs[DisplayUnitsShort] = string(unitRegexp.Find([]byte(d.String())))
+	return
+}
+
+type Reporter struct {
+	Email, Token    string
+	Source          string
+	Interval        time.Duration
+	Registry        metrics.Registry
+	Percentiles     []float64              // percentiles to report on histogram metrics
+	TimerAttributes map[string]interface{} // units in which timers will be displayed
 }
 
-func Librato(r metrics.Registry, d time.Duration, e string, t string, s string, p []float64) {
-	reporter := &LibratoReporter{e, t, s, d, r, p}
-	reporter.Run()
+func NewReporter(r metrics.Registry, d time.Duration, e string, t string, s string, p []float64, u time.Duration) *Reporter {
+	return &Reporter{e, t, s, d, r, p, translateTimerAttributes(u)}
 }
 
-func (self *LibratoReporter) Run() {
+func Librato(r metrics.Registry, d time.Duration, e string, t string, s string, p []float64, u time.Duration) {
+	NewReporter(r, d, e, t, s, p, u).Run()
+}
+
+func (self *Reporter) Run() {
 	ticker := time.Tick(self.Interval)
-	metricsApi := &librato.Metrics{self.Email, self.Token}
+	metricsApi := &LibratoClient{self.Email, self.Token}
 	for now := range ticker {
-		var metrics *librato.MetricsFormat
+		var metrics Batch
 		var err error
 		if metrics, err = self.BuildRequest(now, self.Registry); err != nil {
 			log.Printf("ERROR constructing librato request body %s", err)
 		}
-		if err := metricsApi.SendMetrics(metrics); err != nil {
+		if err := metricsApi.PostMetrics(metrics); err != nil {
 			log.Printf("ERROR sending metrics to librato %s", err)
 		}
 	}
@@ -60,81 +75,123 @@ func sumSquaresTimer(m metrics.Timer) float64 {
 	return sumSquares
 }
 
-func (self *LibratoReporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot *librato.MetricsFormat, err error) {
-	snapshot = &librato.MetricsFormat{}
+func (self *Reporter) BuildRequest(now time.Time, r metrics.Registry) (snapshot Batch, err error) {
+	snapshot = Batch{
+		MeasureTime: now.Unix(),
+		Source:      self.Source,
+	}
 	snapshot.MeasureTime = now.Unix()
-	snapshot.Source = self.Source
-	snapshot.Gauges = make([]interface{}, 0)
-	snapshot.Counters = make([]librato.Metric, 0)
+	snapshot.Gauges = make([]Measurement, 0)
+	snapshot.Counters = make([]Measurement, 0)
 	histogramGaugeCount := 1 + len(self.Percentiles)
 	r.Each(func(name string, metric interface{}) {
+		measurement := Measurement{}
+		measurement[Period] = self.Interval.Seconds()
 		switch m := metric.(type) {
 		case metrics.Counter:
-			libratoName := fmt.Sprintf("%s.%s", name, "count")
-			snapshot.Counters = append(snapshot.Counters, librato.Metric{Name: libratoName, Value: float64(m.Count())})
+			measurement[Name] = fmt.Sprintf("%s.%s", name, "count")
+			measurement[Value] = float64(m.Count())
+			snapshot.Counters = append(snapshot.Counters, measurement)
 		case metrics.Gauge:
-			snapshot.Gauges = append(snapshot.Gauges, librato.Metric{Name: name, Value: float64(m.Value())})
+			measurement[Name] = name
+			measurement[Value] = float64(m.Value())
+			snapshot.Gauges = append(snapshot.Gauges, measurement)
 		case metrics.Histogram:
 			if m.Count() > 0 {
-				libratoName := fmt.Sprintf("%s.%s", name, "hist")
-				gauges := make([]interface{}, histogramGaugeCount, histogramGaugeCount)
-				gauges[0] = librato.Gauge{
-					Name:       libratoName,
-					Count:      uint64(m.Count()),
-					Sum:        m.Mean() * float64(m.Count()),
-					Max:        float64(m.Max()),
-					Min:        float64(m.Min()),
-					SumSquares: sumSquares(m),
-				}
+				gauges := make([]Measurement, histogramGaugeCount, histogramGaugeCount)
+				measurement[Name] = fmt.Sprintf("%s.%s", name, "hist")
+				measurement[Count] = uint64(m.Count())
+				measurement[Sum] = m.Mean() * float64(m.Count())
+				measurement[Max] = float64(m.Max())
+				measurement[Min] = float64(m.Min())
+				measurement[SumSquares] = sumSquares(m)
+				gauges[0] = measurement
 				for i, p := range self.Percentiles {
-					gauges[i+1] = librato.Metric{Name: fmt.Sprintf("%s.%.2f", libratoName, p), Value: m.Percentile(p)}
+					gauges[i+1] = Measurement{
+						Name:   fmt.Sprintf("%s.%.2f", measurement[Name], p),
+						Value:  m.Percentile(p),
+						Period: measurement[Period],
+					}
 				}
 				snapshot.Gauges = append(snapshot.Gauges, gauges...)
 			}
 		case metrics.Meter:
-			snapshot.Counters = append(snapshot.Counters, librato.Metric{Name: name, Value: float64(m.Count())})
+			measurement[Name] = name
+			measurement[Value] = float64(m.Count())
+			snapshot.Counters = append(snapshot.Counters, measurement)
 			snapshot.Gauges = append(snapshot.Gauges,
-				librato.Metric{
-					Name:  fmt.Sprintf("%s.%s", name, "1min"),
-					Value: m.Rate1(),
+				Measurement{
+					Name:   fmt.Sprintf("%s.%s", name, "1min"),
+					Value:  m.Rate1(),
+					Period: int64(self.Interval.Seconds()),
 				},
-				librato.Metric{
-					Name:  fmt.Sprintf("%s.%s", name, "5min"),
-					Value: m.Rate5(),
+				Measurement{
+					Name:   fmt.Sprintf("%s.%s", name, "5min"),
+					Value:  m.Rate5(),
+					Period: int64(self.Interval.Seconds()),
 				},
-				librato.Metric{
-					Name:  fmt.Sprintf("%s.%s", name, "15min"),
-					Value: m.Rate15(),
+				Measurement{
+					Name:   fmt.Sprintf("%s.%s", name, "15min"),
+					Value:  m.Rate15(),
+					Period: int64(self.Interval.Seconds()),
 				},
 			)
 		case metrics.Timer:
+			measurement[Name] = name
+			measurement[Value] = float64(m.Count())
+			snapshot.Counters = append(snapshot.Counters, measurement)
 			if m.Count() > 0 {
-				libratoName := fmt.Sprintf("%s.%s", name, "timer")
-				gauges := make([]interface{}, histogramGaugeCount, histogramGaugeCount)
-				gauges[0] = librato.Gauge{
+				libratoName := fmt.Sprintf("%s.%s", name, "timer.mean")
+				gauges := make([]Measurement, histogramGaugeCount, histogramGaugeCount)
+				gauges[0] = Measurement{
 					Name:       libratoName,
 					Count:      uint64(m.Count()),
 					Sum:        m.Mean() * float64(m.Count()),
 					Max:        float64(m.Max()),
 					Min:        float64(m.Min()),
 					SumSquares: sumSquaresTimer(m),
+					Period:     int64(self.Interval.Seconds()),
+					Attributes: self.TimerAttributes,
 				}
 				for i, p := range self.Percentiles {
-					gauges[i+1] = librato.Metric{Name: fmt.Sprintf("%s.%2.0f", libratoName, p*100), Value: m.Percentile(p)}
+					gauges[i+1] = Measurement{
+						Name:       fmt.Sprintf("%s.timer.%2.0f", name, p*100),
+						Value:      m.Percentile(p),
+						Period:     int64(self.Interval.Seconds()),
+						Attributes: self.TimerAttributes,
+					}
 				}
 				snapshot.Gauges = append(snapshot.Gauges, gauges...)
 				snapshot.Gauges = append(snapshot.Gauges,
-					librato.Metric{
-						Name:  fmt.Sprintf("%s.%s", name, "1min"),
-						Value: m.Rate1(),
+					Measurement{
+						Name:   fmt.Sprintf("%s.%s", name, "rate.1min"),
+						Value:  m.Rate1(),
+						Period: int64(self.Interval.Seconds()),
+						Attributes: map[string]interface{}{
+							DisplayUnitsLong:  Operations,
+							DisplayUnitsShort: OperationsShort,
+							DisplayMin:        "0",
+						},
 					},
-					librato.Metric{
-						Name:  fmt.Sprintf("%s.%s", name, "5min"),
-						Value: m.Rate5(),
+					Measurement{
+						Name:   fmt.Sprintf("%s.%s", name, "rate.5min"),
+						Value:  m.Rate5(),
+						Period: int64(self.Interval.Seconds()),
+						Attributes: map[string]interface{}{
+							DisplayUnitsLong:  Operations,
+							DisplayUnitsShort: OperationsShort,
+							DisplayMin:        "0",
+						},
 					},
-					librato.Metric{
-						Name:  fmt.Sprintf("%s.%s", name, "15min"),
-						Value: m.Rate15(),
+					Measurement{
+						Name:   fmt.Sprintf("%s.%s", name, "rate.15min"),
+						Value:  m.Rate15(),
+						Period: int64(self.Interval.Seconds()),
+						Attributes: map[string]interface{}{
+							DisplayUnitsLong:  Operations,
+							DisplayUnitsShort: OperationsShort,
+							DisplayMin:        "0",
+						},
 					},
 				)
 			}