feat(instrumentation): more work

conf/defaults.ini

@@ -348,14 +348,4 @@ interval_seconds  = 10
 address = localhost:2003
 prefix = prod.grafana.%(instance_name)s.
 
-[metrics.influxdb]
-url             = http://localhost:8086
-database        = site
-prefix          =
-username        = grafana
-password        = grafana
-
-[metrics.influxdb.tags]
-hostname = ${HOSTNAME}
-service = Grafana
 

pkg/api/api.go

@@ -237,6 +237,9 @@ func Register(r *macaron.Macaron) {
 		// metrics
 		r.Get("/metrics/test", GetTestMetrics)
 
+		// metrics
+		r.Get("/metrics", wrap(GetInternalMetrics))
+
 	}, reqSignedIn)
 
 	// admin api

pkg/api/dataproxy.go

@@ -17,6 +17,8 @@ import (
 	"github.com/grafana/grafana/pkg/util"
 )
 
+var i int = 0
+
 var dataProxyTransport = &http.Transport{
 	TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
 	Proxy:           http.ProxyFromEnvironment,

pkg/api/metrics.go

 package api
 
 import (
-	"github.com/grafana/grafana/pkg/api/dtos"
-	"github.com/grafana/grafana/pkg/middleware"
 	"math/rand"
 	"strconv"
+
+	"github.com/grafana/grafana/pkg/api/dtos"
+	"github.com/grafana/grafana/pkg/metrics"
+	"github.com/grafana/grafana/pkg/middleware"
 )
 
 func GetTestMetrics(c *middleware.Context) {
@@ -34,3 +36,35 @@ func GetTestMetrics(c *middleware.Context) {
 
 	c.JSON(200, &result)
 }
+
+func GetInternalMetrics(c middleware.Context) Response {
+	snapshots := metrics.MetricStats.GetSnapshots()
+
+	resp := make(map[string]interface{})
+
+	for _, m := range snapshots {
+		metricName := m.Name() + m.StringifyTags()
+
+		switch metric := m.(type) {
+		case metrics.Counter:
+			resp[metricName] = map[string]interface{}{
+				"count": metric.Count(),
+			}
+		case metrics.Timer:
+			percentiles := metric.Percentiles([]float64{0.25, 0.75, 0.90, 0.99})
+			resp[metricName] = map[string]interface{}{
+				"count": metric.Count(),
+				"min":   metric.Min(),
+				"max":   metric.Max(),
+				"mean":  metric.Mean(),
+				"std":   metric.StdDev(),
+				"p25":   percentiles[0],
+				"p75":   percentiles[1],
+				"p90":   percentiles[2],
+				"p99":   percentiles[3],
+			}
+		}
+	}
+
+	return Json(200, resp)
+}

pkg/metrics/delta.go

+package metrics
+
+import "math"
+
+func calculateDelta(oldValue, newValue int64) int64 {
+	if oldValue < newValue {
+		return newValue - oldValue
+	} else {
+		return (math.MaxInt64 - oldValue) + (newValue - math.MinInt64) + 1
+	}
+}

pkg/metrics/graphite.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"fmt"
 	"net"
-	"reflect"
 	"time"
 
 	"github.com/grafana/grafana/pkg/log"
@@ -12,9 +11,10 @@ import (
 )
 
 type GraphitePublisher struct {
-	Address  string
-	Protocol string
-	Prefix   string
+	address    string
+	protocol   string
+	prefix     string
+	prevCounts map[string]int64
 }
 
 func CreateGraphitePublisher() (*GraphitePublisher, error) {
@@ -24,15 +24,16 @@ func CreateGraphitePublisher() (*GraphitePublisher, error) {
 	}
 
 	publisher := &GraphitePublisher{}
-	publisher.Protocol = "tcp"
-	publisher.Address = graphiteSection.Key("address").MustString("localhost:2003")
-	publisher.Prefix = graphiteSection.Key("prefix").MustString("service.grafana.%(instance_name)s")
+	publisher.prevCounts = make(map[string]int64)
+	publisher.protocol = "tcp"
+	publisher.address = graphiteSection.Key("address").MustString("localhost:2003")
+	publisher.prefix = graphiteSection.Key("prefix").MustString("service.grafana.%(instance_name)s")
 
 	return publisher, nil
 }
 
 func (this *GraphitePublisher) Publish(metrics []Metric) {
-	conn, err := net.DialTimeout(this.Protocol, this.Address, time.Second*5)
+	conn, err := net.DialTimeout(this.protocol, this.address, time.Second*5)
 
 	if err != nil {
 		log.Error(3, "Metrics: GraphitePublisher:  Failed to connect to %s!", err)
@@ -41,36 +42,24 @@ func (this *GraphitePublisher) Publish(metrics []Metric) {
 
 	buf := bytes.NewBufferString("")
 	now := time.Now().Unix()
-	addIntToBuf := func(metric string, value int64) {
-		buf.WriteString(fmt.Sprintf("%s %d %d\n", metric, value, now))
-	}
-	addFloatToBuf := func(metric string, value float64) {
-		buf.WriteString(fmt.Sprintf("%s %f %d\n", metric, value, now))
-	}
 
 	for _, m := range metrics {
-		log.Info("metric: %v, %v", m, reflect.TypeOf(m))
-		metricName := this.Prefix + m.Name() + m.StringifyTags()
+		metricName := this.prefix + m.Name() + m.StringifyTags()
 
 		switch metric := m.(type) {
 		case Counter:
-			addIntToBuf(metricName+".count", metric.Count())
-		case SimpleTimer:
-			addIntToBuf(metricName+".count", metric.Count())
-			addIntToBuf(metricName+".max", metric.Max())
-			addIntToBuf(metricName+".min", metric.Min())
-			addFloatToBuf(metricName+".mean", metric.Mean())
+			this.addCount(buf, metricName+".count", metric.Count(), now)
 		case Timer:
 			percentiles := metric.Percentiles([]float64{0.25, 0.75, 0.90, 0.99})
-			addIntToBuf(metricName+".count", metric.Count())
-			addIntToBuf(metricName+".max", metric.Max())
-			addIntToBuf(metricName+".min", metric.Min())
-			addFloatToBuf(metricName+".mean", metric.Mean())
-			addFloatToBuf(metricName+".std", metric.StdDev())
-			addFloatToBuf(metricName+".p25", percentiles[0])
-			addFloatToBuf(metricName+".p75", percentiles[1])
-			addFloatToBuf(metricName+".p90", percentiles[2])
-			addFloatToBuf(metricName+".p99", percentiles[3])
+			this.addCount(buf, metricName+".count", metric.Count(), now)
+			this.addInt(buf, metricName+".max", metric.Max(), now)
+			this.addInt(buf, metricName+".min", metric.Min(), now)
+			this.addFloat(buf, metricName+".mean", metric.Mean(), now)
+			this.addFloat(buf, metricName+".std", metric.StdDev(), now)
+			this.addFloat(buf, metricName+".p25", percentiles[0], now)
+			this.addFloat(buf, metricName+".p75", percentiles[1], now)
+			this.addFloat(buf, metricName+".p90", percentiles[2], now)
+			this.addFloat(buf, metricName+".p99", percentiles[3], now)
 		}
 
 	}
@@ -82,3 +71,22 @@ func (this *GraphitePublisher) Publish(metrics []Metric) {
 		log.Error(3, "Metrics: GraphitePublisher: Failed to send metrics! %s", err)
 	}
 }
+
+func (this *GraphitePublisher) addInt(buf *bytes.Buffer, metric string, value int64, now int64) {
+	buf.WriteString(fmt.Sprintf("%s %d %d\n", metric, value, now))
+}
+
+func (this *GraphitePublisher) addFloat(buf *bytes.Buffer, metric string, value float64, now int64) {
+	buf.WriteString(fmt.Sprintf("%s %f %d\n", metric, value, now))
+}
+
+func (this *GraphitePublisher) addCount(buf *bytes.Buffer, metric string, value int64, now int64) {
+	delta := value
+
+	if last, ok := this.prevCounts[metric]; ok {
+		delta = calculateDelta(last, value)
+	}
+
+	this.prevCounts[metric] = value
+	buf.WriteString(fmt.Sprintf("%s %d %d\n", metric, delta, now))
+}

pkg/metrics/influxdb.go

@@ -10,10 +10,11 @@ import (
 )
 
 type InfluxPublisher struct {
-	database string
-	tags     map[string]string
-	prefix   string
-	client   *client.Client
+	database   string
+	tags       map[string]string
+	prefix     string
+	client     *client.Client
+	prevCounts map[string]int64
 }
 
 func CreateInfluxPublisher() (*InfluxPublisher, error) {
@@ -36,6 +37,7 @@ func CreateInfluxPublisher() (*InfluxPublisher, error) {
 
 	publisher.database = influxSection.Key("database").MustString("grafana_metrics")
 	publisher.prefix = influxSection.Key("prefix").MustString("prefix")
+	publisher.prevCounts = make(map[string]int64)
 
 	username := influxSection.Key("User").MustString("grafana")
 	password := influxSection.Key("Password").MustString("grafana")
@@ -75,23 +77,20 @@ func (this *InfluxPublisher) Publish(metrics []Metric) {
 	}
 
 	for _, m := range metrics {
-		tags := m.GetTagsCopy()
-		addPoint := func(name string, value interface{}) {
-			bp.Points = append(bp.Points, client.Point{
-				Measurement: name,
-				Tags:        tags,
-				Fields:      map[string]interface{}{"value": value},
-			})
-		}
-
 		switch metric := m.(type) {
 		case Counter:
-			addPoint(metric.Name()+".count", metric.Count())
-		case SimpleTimer:
-			addPoint(metric.Name()+".count", metric.Count())
-			addPoint(metric.Name()+".max", metric.Max())
-			addPoint(metric.Name()+".min", metric.Min())
-			addPoint(metric.Name()+".avg", metric.Mean())
+			this.addPoint(&bp, metric, "count", metric.Count())
+		case Timer:
+			percentiles := metric.Percentiles([]float64{0.25, 0.75, 0.90, 0.99})
+			this.addPoint(&bp, metric, "count", metric.Count())
+			this.addPoint(&bp, metric, "min", metric.Min())
+			this.addPoint(&bp, metric, "max", metric.Max())
+			this.addPoint(&bp, metric, "mean", metric.Mean())
+			this.addPoint(&bp, metric, "std", metric.StdDev())
+			this.addPoint(&bp, metric, "p25", percentiles[0])
+			this.addPoint(&bp, metric, "p75", percentiles[1])
+			this.addPoint(&bp, metric, "p90", percentiles[2])
+			this.addPoint(&bp, metric, "p99", percentiles[2])
 		}
 	}
 
@@ -100,3 +99,32 @@ func (this *InfluxPublisher) Publish(metrics []Metric) {
 		log.Error(3, "Metrics: InfluxPublisher: publish error", err)
 	}
 }
+
+func (this *InfluxPublisher) addPoint(bp *client.BatchPoints, metric Metric, metricTag string, value interface{}) {
+	tags := metric.GetTagsCopy()
+	tags["metric"] = metricTag
+
+	bp.Points = append(bp.Points, client.Point{
+		Measurement: metric.Name(),
+		Tags:        tags,
+		Fields:      map[string]interface{}{"value": value},
+	})
+}
+
+func (this *InfluxPublisher) addCountPoint(bp *client.BatchPoints, metric Metric, value int64) {
+	tags := metric.GetTagsCopy()
+	tags["metric"] = "count"
+
+	name := metric.Name()
+	delta := value
+	if last, ok := this.prevCounts[name]; ok {
+		delta = calculateDelta(last, value)
+	}
+	this.prevCounts[name] = value
+
+	bp.Points = append(bp.Points, client.Point{
+		Measurement: name,
+		Tags:        tags,
+		Fields:      map[string]interface{}{"value": delta},
+	})
+}

pkg/metrics/registry.go

@@ -32,12 +32,6 @@ func (r *StandardRegistry) GetSnapshots() []Metric {
 	metrics := make([]Metric, len(r.metrics))
 	for i, metric := range r.metrics {
 		metrics[i] = metric.Snapshot()
-		switch typedMetric := metric.(type) {
-		case Histogram:
-			// do not clear histograms
-		case Counter:
-			typedMetric.Clear()
-		}
 	}
 	return metrics
 }

pkg/metrics/settings.go

@@ -42,12 +42,5 @@ func readSettings() *MetricSettings {
 		settings.Publishers = append(settings.Publishers, graphitePublisher)
 	}
 
-	if influxPublisher, err := CreateInfluxPublisher(); err != nil {
-		log.Error(3, "Metrics: Failed to init InfluxDB metric publisher", err)
-	} else if influxPublisher != nil {
-		log.Info("Metrics: Internal metrics publisher InfluxDB initialized")
-		settings.Publishers = append(settings.Publishers, influxPublisher)
-	}
-
 	return settings
 }

pkg/metrics/simple_timer.go

-package metrics
-
-//import "sync/atomic"
-
-type SimpleTimer interface {
-	Metric
-
-	AddTiming(int64)
-	Mean() float64
-	Min() int64
-	Max() int64
-	Count() int64
-}
-
-type StandardSimpleTimer struct {
-	*MetricMeta
-
-	total int64
-	count int64
-	mean  float64
-	min   int64
-	max   int64
-}
-
-func NewSimpleTimer(meta *MetricMeta) SimpleTimer {
-	return &StandardSimpleTimer{
-		MetricMeta: meta,
-		mean:       0,
-		min:        0,
-		max:        0,
-		total:      0,
-		count:      0,
-	}
-}
-
-func RegSimpleTimer(name string, tagStrings ...string) SimpleTimer {
-	tr := NewSimpleTimer(NewMetricMeta(name, tagStrings))
-	MetricStats.Register(tr)
-	return tr
-}
-
-func (this *StandardSimpleTimer) AddTiming(time int64) {
-	if this.min > time {
-		this.min = time
-	}
-
-	if this.max < time {
-		this.max = time
-	}
-
-	this.total += time
-	this.count++
-	this.mean = float64(this.total) / float64(this.count)
-}
-
-func (this *StandardSimpleTimer) Clear() {
-	this.mean = 0
-	this.min = 0
-	this.max = 0
-	this.total = 0
-	this.count = 0
-}
-
-func (this *StandardSimpleTimer) Mean() float64 {
-	return this.mean
-}
-
-func (this *StandardSimpleTimer) Min() int64 {
-	return this.min
-}
-
-func (this *StandardSimpleTimer) Max() int64 {
-	return this.max
-}
-
-func (this *StandardSimpleTimer) Count() int64 {
-	return this.count
-}
-
-func (this *StandardSimpleTimer) Snapshot() Metric {
-	return &StandardSimpleTimer{
-		MetricMeta: this.MetricMeta,
-		mean:       this.mean,
-		min:        this.min,
-		max:        this.max,
-		total:      this.total,
-		count:      this.count,
-	}
-}