feat(instrumentation): influxdb is working, now need to find a way to better support tags, #4696

pkg/metrics/publishers/graphite.go

@@ -22,12 +22,12 @@ func CreateGraphitePublisher() (*GraphitePublisher, error) {
 		return nil, nil
 		return nil, nil
 	}
 	}
 
 
-	graphiteReceiver := &GraphitePublisher{}
-	graphiteReceiver.Protocol = "tcp"
-	graphiteReceiver.Address = graphiteSection.Key("address").MustString("localhost:2003")
-	graphiteReceiver.Prefix = graphiteSection.Key("prefix").MustString("service.grafana.%(instance_name)s")
+	publisher := &GraphitePublisher{}
+	publisher.Protocol = "tcp"
+	publisher.Address = graphiteSection.Key("address").MustString("localhost:2003")
+	publisher.Prefix = graphiteSection.Key("prefix").MustString("service.grafana.%(instance_name)s")
 
 
-	return graphiteReceiver, nil
+	return publisher, nil
 }
 }
 
 
 func (this *GraphitePublisher) Publish(metrics map[string]interface{}) {
 func (this *GraphitePublisher) Publish(metrics map[string]interface{}) {

pkg/metrics/publishers/influxdb.go

+package publishers
+
+import (
+	"net/url"
+	"time"
+
+	"github.com/grafana/grafana/pkg/log"
+	"github.com/grafana/grafana/pkg/setting"
+	"github.com/influxdata/influxdb/client"
+)
+
+type InfluxPublisher struct {
+	database string
+	tags     map[string]string
+	client   *client.Client
+}
+
+func CreateInfluxPublisher() (*InfluxPublisher, error) {
+	influxSection, err := setting.Cfg.GetSection("metrics.influxdb")
+	if err != nil {
+		return nil, nil
+	}
+
+	publisher := &InfluxPublisher{
+		tags: make(map[string]string),
+	}
+
+	urlStr := influxSection.Key("url").MustString("localhost:2003")
+	urlParsed, err := url.Parse(urlStr)
+
+	if err != nil {
+		log.Error(3, "Metics: InfluxPublisher: failed to init influxdb publisher", err)
+		return nil, nil
+	}
+
+	publisher.database = influxSection.Key("database").MustString("grafana_metrics")
+	username := influxSection.Key("User").MustString("grafana")
+	password := influxSection.Key("Password").MustString("grafana")
+
+	publisher.client, err = client.NewClient(client.Config{
+		URL:      *urlParsed,
+		Username: username,
+		Password: password,
+	})
+
+	tagsSec, err := setting.Cfg.GetSection("metrics.influxdb.tags")
+	if err != nil {
+		log.Error(3, "Metics: InfluxPublisher: failed to init influxdb settings no metrics.influxdb.tags section")
+		return nil, nil
+	}
+
+	for _, key := range tagsSec.Keys() {
+		publisher.tags[key.Name()] = key.String()
+	}
+
+	if err != nil {
+		log.Error(3, "Metics: InfluxPublisher: failed to init influxdb publisher", err)
+	}
+
+	return publisher, nil
+}
+
+func (this *InfluxPublisher) Publish(metrics map[string]interface{}) {
+	bp := client.BatchPoints{
+		Time:     time.Now(),
+		Database: this.database,
+		Tags:     map[string]string{},
+	}
+
+	for key, value := range this.tags {
+		bp.Tags[key] = value
+	}
+
+	for key, value := range metrics {
+		bp.Points = append(bp.Points, client.Point{
+			Measurement: key,
+			Fields: map[string]interface{}{
+				"value": value,
+			},
+		})
+	}
+
+	_, err := this.client.Write(bp)
+	if err != nil {
+		log.Error(3, "Metrics: InfluxPublisher: publish error", err)
+	}
+}

pkg/metrics/settings.go

@@ -43,5 +43,12 @@ func readSettings() *MetricSettings {
 		settings.Publishers = append(settings.Publishers, graphitePublisher)
 		settings.Publishers = append(settings.Publishers, graphitePublisher)
 	}
 	}
 
 
+	if influxPublisher, err := publishers.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
 	return settings
 }
 }