feat(instrumentation): lots of refactoring to support tag based backend, #4696

conf/defaults.ini

@@ -338,3 +338,24 @@ global_api_key = -1
 
 # global limit on number of logged in users.
 global_session = -1
+
+#################################### Internal Grafana Metrics ##########################
+[metrics]
+enabled           = false
+interval_seconds  = 10
+
+[metrics.graphite]
+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/dashboard.go

@@ -77,8 +77,6 @@ func GetDashboard(c *middleware.Context) {
 	}
 
 	c.JSON(200, dto)
-
-	metrics.M_Api_Dashboard_Get_Timer.AddTiming(123333)
 }
 
 func getUserLogin(userId int64) string {

pkg/metrics/common.go

@@ -0,0 +1,53 @@
+package metrics
+
+import "github.com/grafana/grafana/pkg/log"
+
+type MetricMeta struct {
+	tags map[string]string
+	name string
+}
+
+func NewMetricMeta(name string, tagStrings []string) *MetricMeta {
+	if len(tagStrings)%2 != 0 {
+		log.Fatal(3, "Metrics: tags array is missing value for key, %v", tagStrings)
+	}
+
+	tags := make(map[string]string)
+	for i := 0; i < len(tagStrings); i += 2 {
+		tags[tagStrings[i]] = tagStrings[i+1]
+	}
+
+	return &MetricMeta{
+		tags: tags,
+		name: name,
+	}
+}
+
+func (m *MetricMeta) Name() string {
+	return m.name
+}
+
+func (m *MetricMeta) Tags() map[string]string {
+	return m.tags
+}
+
+func (m *MetricMeta) StringifyTags() string {
+	if len(m.tags) == 0 {
+		return ""
+	}
+
+	str := ""
+	for key, value := range m.tags {
+		str += "." + key + "_" + value
+	}
+
+	return str
+}
+
+type Metric interface {
+	Name() string
+	Tags() map[string]string
+	StringifyTags() string
+	Snapshot() Metric
+	Clear()
+}

pkg/metrics/counter.go

@@ -4,45 +4,26 @@ import "sync/atomic"
 
 // Counters hold an int64 value that can be incremented and decremented.
 type Counter interface {
-	Clear()
+	Metric
+
 	Count() int64
 	Dec(int64)
 	Inc(int64)
-	Snapshot() Counter
 }
 
 // NewCounter constructs a new StandardCounter.
-func NewCounter() Counter {
-	return &StandardCounter{0}
-}
-
-// CounterSnapshot is a read-only copy of another Counter.
-type CounterSnapshot int64
-
-// Clear panics.
-func (CounterSnapshot) Clear() {
-	panic("Clear called on a CounterSnapshot")
+func NewCounter(meta *MetricMeta) Counter {
+	return &StandardCounter{
+		MetricMeta: meta,
+		count:      0,
+	}
 }
 
-// Count returns the count at the time the snapshot was taken.
-func (c CounterSnapshot) Count() int64 { return int64(c) }
-
-// Dec panics.
-func (CounterSnapshot) Dec(int64) {
-	panic("Dec called on a CounterSnapshot")
-}
-
-// Inc panics.
-func (CounterSnapshot) Inc(int64) {
-	panic("Inc called on a CounterSnapshot")
-}
-
-// Snapshot returns the snapshot.
-func (c CounterSnapshot) Snapshot() Counter { return c }
-
 // StandardCounter is the standard implementation of a Counter and uses the
 // sync/atomic package to manage a single int64 value.
 type StandardCounter struct {
+	*MetricMeta
+
 	count int64
 }
 
@@ -66,7 +47,9 @@ func (c *StandardCounter) Inc(i int64) {
 	atomic.AddInt64(&c.count, i)
 }
 
-// Snapshot returns a read-only copy of the counter.
-func (c *StandardCounter) Snapshot() Counter {
-	return CounterSnapshot(c.Count())
+func (c *StandardCounter) Snapshot() Metric {
+	return &StandardCounter{
+		MetricMeta: c.MetricMeta,
+		count:      c.count,
+	}
 }

pkg/metrics/publishers/graphite.go → pkg/metrics/graphite.go

@@ -1,4 +1,4 @@
-package publishers
+package metrics
 
 import (
 	"bytes"
@@ -30,7 +30,7 @@ func CreateGraphitePublisher() (*GraphitePublisher, error) {
 	return publisher, nil
 }
 
-func (this *GraphitePublisher) Publish(metrics map[string]interface{}) {
+func (this *GraphitePublisher) Publish(metrics []Metric) {
 	conn, err := net.DialTimeout(this.Protocol, this.Address, time.Second*5)
 
 	if err != nil {
@@ -40,10 +40,17 @@ func (this *GraphitePublisher) Publish(metrics map[string]interface{}) {
 
 	buf := bytes.NewBufferString("")
 	now := time.Now().Unix()
-	for key, value := range metrics {
-		metricName := this.Prefix + key
-		line := fmt.Sprintf("%s %d %d\n", metricName, value, now)
-		buf.WriteString(line)
+	for _, m := range metrics {
+		metricName := this.Prefix + m.Name() + m.StringifyTags()
+
+		switch metric := m.(type) {
+		case Counter:
+			if metric.Count() > 0 {
+				line := fmt.Sprintf("%s %d %d\n", metricName, metric.Count(), now)
+				buf.WriteString(line)
+			}
+		}
+
 	}
 
 	log.Trace("Metrics: GraphitePublisher.Publish() \n%s", buf)

pkg/metrics/publishers/influxdb.go → pkg/metrics/influxdb.go

@@ -1,4 +1,4 @@
-package publishers
+package metrics
 
 import (
 	"net/url"
@@ -12,6 +12,7 @@ import (
 type InfluxPublisher struct {
 	database string
 	tags     map[string]string
+	prefix   string
 	client   *client.Client
 }
 
@@ -34,6 +35,8 @@ func CreateInfluxPublisher() (*InfluxPublisher, error) {
 	}
 
 	publisher.database = influxSection.Key("database").MustString("grafana_metrics")
+	publisher.prefix = influxSection.Key("prefix").MustString("prefix")
+
 	username := influxSection.Key("User").MustString("grafana")
 	password := influxSection.Key("Password").MustString("grafana")
 
@@ -60,7 +63,7 @@ func CreateInfluxPublisher() (*InfluxPublisher, error) {
 	return publisher, nil
 }
 
-func (this *InfluxPublisher) Publish(metrics map[string]interface{}) {
+func (this *InfluxPublisher) Publish(metrics []Metric) {
 	bp := client.BatchPoints{
 		Time:     time.Now(),
 		Database: this.database,
@@ -71,13 +74,19 @@ func (this *InfluxPublisher) Publish(metrics map[string]interface{}) {
 		bp.Tags[key] = value
 	}
 
-	for key, value := range metrics {
-		bp.Points = append(bp.Points, client.Point{
-			Measurement: key,
-			Fields: map[string]interface{}{
-				"value": value,
-			},
-		})
+	for _, m := range metrics {
+		point := client.Point{
+			Measurement: this.prefix + m.Name(),
+			Tags:        m.Tags(),
+		}
+
+		switch metric := m.(type) {
+		case Counter:
+			if metric.Count() > 0 {
+				point.Fields = map[string]interface{}{"value": metric.Count()}
+				bp.Points = append(bp.Points, point)
+			}
+		}
 	}
 
 	_, err := this.client.Write(bp)

pkg/metrics/metric_ref.go

@@ -1,28 +1,38 @@
 package metrics
 
 type comboCounterRef struct {
+	*MetricMeta
 	usageCounter  Counter
 	metricCounter Counter
 }
 
 type comboTimerRef struct {
+	*MetricMeta
 	usageTimer  Timer
 	metricTimer Timer
 }
 
-func NewComboCounterRef(name string) Counter {
-	cr := &comboCounterRef{}
-	cr.usageCounter = UsageStats.GetOrRegister(name, NewCounter).(Counter)
-	cr.metricCounter = MetricStats.GetOrRegister(name, NewCounter).(Counter)
+func RegComboCounter(name string, tagStrings ...string) Counter {
+	meta := NewMetricMeta(name, tagStrings)
+	cr := &comboCounterRef{
+		MetricMeta:    meta,
+		usageCounter:  NewCounter(meta),
+		metricCounter: NewCounter(meta),
+	}
+
+	UsageStats.Register(cr.usageCounter)
+	MetricStats.Register(cr.metricCounter)
+
 	return cr
 }
 
-func NewComboTimerRef(name string) Timer {
-	tr := &comboTimerRef{}
-	tr.usageTimer = UsageStats.GetOrRegister(name, NewTimer).(Timer)
-	tr.metricTimer = MetricStats.GetOrRegister(name, NewTimer).(Timer)
-	return tr
-}
+// func NewComboTimerRef(name string, tagStrings ...string) Timer {
+// 	meta := NewMetricMeta(name, tagStrings)
+// 	tr := &comboTimerRef{}
+// 	tr.usageTimer = UsageStats.GetOrRegister(NewTimer).(Timer)
+// 	tr.metricTimer = MetricStats.GetOrRegister(NewTimer).(Timer)
+// 	return tr
+// }
 
 func (t comboTimerRef) Clear() {
 	t.metricTimer.Clear()
@@ -71,7 +81,6 @@ func (c comboCounterRef) Inc(i int64) {
 	c.metricCounter.Inc(i)
 }
 
-// Snapshot returns the snapshot.
-func (c comboCounterRef) Snapshot() Counter {
-	panic("snapshot called on a combocounter ref")
+func (c comboCounterRef) Snapshot() Metric {
+	return c.metricCounter.Snapshot()
 }

pkg/metrics/metrics.go

@@ -4,31 +4,31 @@ var UsageStats = NewRegistry()
 var MetricStats = NewRegistry()
 
 var (
-	M_Instance_Start = NewComboCounterRef("instance.start")
+	M_Instance_Start = RegComboCounter("instance_start")
 
-	M_Page_Status_200 = NewComboCounterRef("page.status.200")
-	M_Page_Status_500 = NewComboCounterRef("page.status.500")
-	M_Page_Status_404 = NewComboCounterRef("page.status.404")
+	M_Page_Status_200 = RegComboCounter("page_resp_status", "code", "200")
+	M_Page_Status_500 = RegComboCounter("page_resp_status", "code", "500")
+	M_Page_Status_404 = RegComboCounter("page_resp_status", "code", "404")
 
-	M_Api_Status_500 = NewComboCounterRef("api.status.500")
-	M_Api_Status_404 = NewComboCounterRef("api.status.404")
+	M_Api_Status_500 = RegComboCounter("api_resp_status", "code", "500")
+	M_Api_Status_404 = RegComboCounter("api_resp_status", "code", "404")
 
-	M_Api_User_SignUpStarted   = NewComboCounterRef("api.user.signup_started")
-	M_Api_User_SignUpCompleted = NewComboCounterRef("api.user.signup_completed")
-	M_Api_User_SignUpInvite    = NewComboCounterRef("api.user.signup_invite")
-	M_Api_Dashboard_Get        = NewComboCounterRef("api.dashboard.get")
+	M_Api_User_SignUpStarted   = RegComboCounter("api.user.signup_started")
+	M_Api_User_SignUpCompleted = RegComboCounter("api.user.signup_completed")
+	M_Api_User_SignUpInvite    = RegComboCounter("api.user.signup_invite")
+	M_Api_Dashboard_Get        = RegComboCounter("api.dashboard.get")
 
-	M_Api_Dashboard_Get_Timer = NewComboTimerRef("api.dashboard_load")
+	// M_Api_Dashboard_Get_Timer = NewComboTimerRef("api.dashboard_load")
 
-	M_Api_Dashboard_Post    = NewComboCounterRef("api.dashboard.post")
-	M_Api_Admin_User_Create = NewComboCounterRef("api.admin.user_create")
-	M_Api_Login_Post        = NewComboCounterRef("api.login.post")
-	M_Api_Login_OAuth       = NewComboCounterRef("api.login.oauth")
-	M_Api_Org_Create        = NewComboCounterRef("api.org.create")
+	M_Api_Dashboard_Post    = RegComboCounter("api.dashboard.post")
+	M_Api_Admin_User_Create = RegComboCounter("api.admin.user_create")
+	M_Api_Login_Post        = RegComboCounter("api.login.post")
+	M_Api_Login_OAuth       = RegComboCounter("api.login.oauth")
+	M_Api_Org_Create        = RegComboCounter("api.org.create")
 
-	M_Api_Dashboard_Snapshot_Create   = NewComboCounterRef("api.dashboard_snapshot.create")
-	M_Api_Dashboard_Snapshot_External = NewComboCounterRef("api.dashboard_snapshot.external")
-	M_Api_Dashboard_Snapshot_Get      = NewComboCounterRef("api.dashboard_snapshot.get")
+	M_Api_Dashboard_Snapshot_Create   = RegComboCounter("api.dashboard_snapshot.create")
+	M_Api_Dashboard_Snapshot_External = RegComboCounter("api.dashboard_snapshot.external")
+	M_Api_Dashboard_Snapshot_Get      = RegComboCounter("api.dashboard_snapshot.get")
 
-	M_Models_Dashboard_Insert = NewComboCounterRef("models.dashboard.insert")
+	M_Models_Dashboard_Insert = RegComboCounter("models.dashboard.insert")
 )

pkg/metrics/publish.go

@@ -39,25 +39,7 @@ func instrumentationLoop() chan struct{} {
 }
 
 func sendMetrics(settings *MetricSettings) {
-	metrics := map[string]interface{}{}
-
-	MetricStats.Each(func(name string, i interface{}) {
-		switch metric := i.(type) {
-		case Counter:
-			if metric.Count() > 0 {
-				metrics[name+".count"] = metric.Count()
-				metric.Clear()
-			}
-		case Timer:
-			if metric.Total() > 0 {
-				metrics[name+".avg"] = metric.Avg()
-				metrics[name+".min"] = metric.Min()
-				metrics[name+".max"] = metric.Max()
-				metrics[name+".total"] = metric.Total()
-				metric.Clear()
-			}
-		}
-	})
+	metrics := MetricStats.GetSnapshots()
 
 	for _, publisher := range settings.Publishers {
 		publisher.Publish(metrics)
@@ -79,15 +61,15 @@ func sendUsageStats() {
 		"metrics": metrics,
 	}
 
-	UsageStats.Each(func(name string, i interface{}) {
-		switch metric := i.(type) {
+	snapshots := UsageStats.GetSnapshots()
+	for _, m := range snapshots {
+		switch metric := m.(type) {
 		case Counter:
 			if metric.Count() > 0 {
-				metrics[name+".count"] = metric.Count()
-				metric.Clear()
+				metrics[metric.Name()+".count"] = metric.Count()
 			}
 		}
-	})
+	}
 
 	statsQuery := m.GetSystemStatsQuery{}
 	if err := bus.Dispatch(&statsQuery); err != nil {

pkg/metrics/registry.go

@@ -1,102 +1,38 @@
 package metrics
 
-import (
-	"fmt"
-	"reflect"
-	"sync"
-)
-
-// DuplicateMetric is the error returned by Registry.Register when a metric
-// already exists.  If you mean to Register that metric you must first
-// Unregister the existing metric.
-type DuplicateMetric string
-
-func (err DuplicateMetric) Error() string {
-	return fmt.Sprintf("duplicate metric: %s", string(err))
-}
+import "sync"
 
 type Registry interface {
-	// Call the given function for each registered metric.
-	Each(func(string, interface{}))
-
-	// Get the metric by the given name or nil if none is registered.
-	Get(string) interface{}
-
-	// Gets an existing metric or registers the given one.
-	// The interface can be the metric to register if not found in registry,
-	// or a function returning the metric for lazy instantiation.
-	GetOrRegister(string, interface{}) interface{}
-
-	// Register the given metric under the given name.
-	Register(string, interface{}) error
+	GetSnapshots() []Metric
+	Register(metric Metric)
 }
 
 // The standard implementation of a Registry is a mutex-protected map
 // of names to metrics.
 type StandardRegistry struct {
-	metrics map[string]interface{}
+	metrics []Metric
 	mutex   sync.Mutex
 }
 
 // Create a new registry.
 func NewRegistry() Registry {
-	return &StandardRegistry{metrics: make(map[string]interface{})}
-}
-
-// Call the given function for each registered metric.
-func (r *StandardRegistry) Each(f func(string, interface{})) {
-	for name, i := range r.registered() {
-		f(name, i)
-	}
-}
-
-// Get the metric by the given name or nil if none is registered.
-func (r *StandardRegistry) Get(name string) interface{} {
-	r.mutex.Lock()
-	defer r.mutex.Unlock()
-	return r.metrics[name]
-}
-
-// Gets an existing metric or creates and registers a new one. Threadsafe
-// alternative to calling Get and Register on failure.
-// The interface can be the metric to register if not found in registry,
-// or a function returning the metric for lazy instantiation.
-func (r *StandardRegistry) GetOrRegister(name string, i interface{}) interface{} {
-	r.mutex.Lock()
-	defer r.mutex.Unlock()
-	if metric, ok := r.metrics[name]; ok {
-		return metric
-	}
-	if v := reflect.ValueOf(i); v.Kind() == reflect.Func {
-		i = v.Call(nil)[0].Interface()
+	return &StandardRegistry{
+		metrics: make([]Metric, 0),
 	}
-	r.register(name, i)
-	return i
 }
 
-// Register the given metric under the given name.  Returns a DuplicateMetric
-// if a metric by the given name is already registered.
-func (r *StandardRegistry) Register(name string, i interface{}) error {
+func (r *StandardRegistry) Register(metric Metric) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
-	return r.register(name, i)
+	r.metrics = append(r.metrics, metric)
 }
 
-func (r *StandardRegistry) register(name string, i interface{}) error {
-	if _, ok := r.metrics[name]; ok {
-		return DuplicateMetric(name)
-	}
-
-	r.metrics[name] = i
-	return nil
-}
-
-func (r *StandardRegistry) registered() map[string]interface{} {
-	metrics := make(map[string]interface{}, len(r.metrics))
-	r.mutex.Lock()
-	defer r.mutex.Unlock()
-	for name, i := range r.metrics {
-		metrics[name] = i
+// Call the given function for each registered metric.
+func (r *StandardRegistry) GetSnapshots() []Metric {
+	metrics := make([]Metric, len(r.metrics))
+	for i, metric := range r.metrics {
+		metrics[i] = metric.Snapshot()
+		metric.Clear()
 	}
 	return metrics
 }

pkg/metrics/settings.go

@@ -2,12 +2,11 @@ package metrics
 
 import (
 	"github.com/grafana/grafana/pkg/log"
-	"github.com/grafana/grafana/pkg/metrics/publishers"
 	"github.com/grafana/grafana/pkg/setting"
 )
 
 type MetricPublisher interface {
-	Publish(metrics map[string]interface{})
+	Publish(metrics []Metric)
 }
 
 type MetricSettings struct {
@@ -36,14 +35,14 @@ func readSettings() *MetricSettings {
 		return settings
 	}
 
-	if graphitePublisher, err := publishers.CreateGraphitePublisher(); err != nil {
+	if graphitePublisher, err := CreateGraphitePublisher(); err != nil {
 		log.Error(3, "Metrics: Failed to init Graphite metric publisher", err)
 	} else if graphitePublisher != nil {
 		log.Info("Metrics: Internal metrics publisher Graphite initialized")
 		settings.Publishers = append(settings.Publishers, graphitePublisher)
 	}
 
-	if influxPublisher, err := publishers.CreateInfluxPublisher(); err != nil {
+	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")

pkg/metrics/timer.go

@@ -3,21 +3,33 @@ package metrics
 //import "sync/atomic"
 
 type Timer interface {
+	Metric
+
 	AddTiming(int64)
-	Clear()
 	Avg() int64
 	Min() int64
 	Max() int64
-	Total() int64
+	Count() int64
+}
+
+type StandardTimer struct {
+	*MetricMeta
+
+	total int64
+	count int64
+	avg   int64
+	min   int64
+	max   int64
 }
 
-func NewTimer() Timer {
+func NewTimer(meta *MetricMeta) Timer {
 	return &StandardTimer{
-		avg:   0,
-		min:   0,
-		max:   0,
-		total: 0,
-		count: 0,
+		MetricMeta: meta,
+		avg:        0,
+		min:        0,
+		max:        0,
+		total:      0,
+		count:      0,
 	}
 }
 
@@ -56,14 +68,17 @@ func (this *StandardTimer) Max() int64 {
 	return this.max
 }
 
-func (this *StandardTimer) Total() int64 {
-	return this.total
+func (this *StandardTimer) Count() int64 {
+	return this.count
 }
 
-type StandardTimer struct {
-	total int64
-	count int64
-	avg   int64
-	min   int64
-	max   int64
+func (this *StandardTimer) Snapshot() Metric {
+	return &StandardTimer{
+		MetricMeta: this.MetricMeta,
+		avg:        this.avg,
+		min:        this.min,
+		max:        this.max,
+		total:      this.total,
+		count:      this.count,
+	}
 }