Add InfluxDB support for traefik metrics

metrics/influxdb.go

@@ -0,0 +1,90 @@
+package metrics
+
+import (
+	"bytes"
+	"time"
+
+	"github.com/containous/traefik/log"
+	"github.com/containous/traefik/safe"
+	"github.com/containous/traefik/types"
+	kitlog "github.com/go-kit/kit/log"
+	"github.com/go-kit/kit/metrics/influx"
+	influxdb "github.com/influxdata/influxdb/client/v2"
+)
+
+var influxDBClient = influx.New(map[string]string{}, influxdb.BatchPointsConfig{}, kitlog.LoggerFunc(func(keyvals ...interface{}) error {
+	log.Info(keyvals)
+	return nil
+}))
+
+type influxDBWriter struct {
+	buf    bytes.Buffer
+	config *types.InfluxDB
+}
+
+var influxDBTicker *time.Ticker
+
+const (
+	influxDBMetricsReqsName    = "traefik.requests.total"
+	influxDBMetricsLatencyName = "traefik.request.duration"
+	influxDBRetriesTotalName   = "traefik.backend.retries.total"
+)
+
+// RegisterInfluxDB registers the metrics pusher if this didn't happen yet and creates a InfluxDB Registry instance.
+func RegisterInfluxDB(config *types.InfluxDB) Registry {
+	if influxDBTicker == nil {
+		influxDBTicker = initInfluxDBTicker(config)
+	}
+
+	return &standardRegistry{
+		enabled:              true,
+		reqsCounter:          influxDBClient.NewCounter(influxDBMetricsReqsName),
+		reqDurationHistogram: influxDBClient.NewHistogram(influxDBMetricsLatencyName),
+		retriesCounter:       influxDBClient.NewCounter(influxDBRetriesTotalName),
+	}
+}
+
+// initInfluxDBTicker initializes metrics pusher and creates a influxDBClient if not created already
+func initInfluxDBTicker(config *types.InfluxDB) *time.Ticker {
+	address := config.Address
+	if len(address) == 0 {
+		address = "localhost:8089"
+	}
+
+	pushInterval, err := time.ParseDuration(config.PushInterval)
+	if err != nil {
+		log.Warnf("Unable to parse %s into pushInterval, using 10s as default value", config.PushInterval)
+		pushInterval = 10 * time.Second
+	}
+
+	report := time.NewTicker(pushInterval)
+
+	safe.Go(func() {
+		var buf bytes.Buffer
+		influxDBClient.WriteLoop(report.C, &influxDBWriter{buf: buf, config: config})
+	})
+
+	return report
+}
+
+// StopInfluxDB stops internal influxDBTicker which controls the pushing of metrics to InfluxDB Agent and resets it to `nil`
+func StopInfluxDB() {
+	if influxDBTicker != nil {
+		influxDBTicker.Stop()
+	}
+	influxDBTicker = nil
+}
+
+func (w *influxDBWriter) Write(bp influxdb.BatchPoints) error {
+	c, err := influxdb.NewUDPClient(influxdb.UDPConfig{
+		Addr: w.config.Address,
+	})
+
+	if err != nil {
+		return err
+	}
+
+	defer c.Close()
+
+	return c.Write(bp)
+}

metrics/influxdb_test.go

@@ -0,0 +1,53 @@
+package metrics
+
+import (
+	"net/http"
+	"regexp"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/containous/traefik/types"
+	"github.com/stvp/go-udp-testing"
+)
+
+func TestInfluxDB(t *testing.T) {
+	udp.SetAddr(":8089")
+	// This is needed to make sure that UDP Listener listens for data a bit longer, otherwise it will quit after a millisecond
+	udp.Timeout = 5 * time.Second
+
+	influxDBRegistry := RegisterInfluxDB(&types.InfluxDB{Address: ":8089", PushInterval: "1s"})
+	defer StopInfluxDB()
+
+	if !influxDBRegistry.IsEnabled() {
+		t.Fatalf("InfluxDB registry must be enabled")
+	}
+
+	expected := []string{
+		"(traefik.requests.total,code=200,method=GET,service=test count=1) [0-9]{19}",
+		"(traefik.requests.total,code=404,method=GET,service=test count=1) [0-9]{19}",
+		"(traefik.request.duration,code=200,method=GET,service=test p50=10000,p90=10000,p95=10000,p99=10000) [0-9]{19}",
+		"(traefik.backend.retries.total,code=404,method=GET,service=test count=2) [0-9]{19}",
+	}
+
+	msg := udp.ReceiveString(t, func() {
+		influxDBRegistry.ReqsCounter().With("service", "test", "code", strconv.Itoa(http.StatusOK), "method", http.MethodGet).Add(1)
+		influxDBRegistry.ReqsCounter().With("service", "test", "code", strconv.Itoa(http.StatusNotFound), "method", http.MethodGet).Add(1)
+		influxDBRegistry.RetriesCounter().With("service", "test").Add(1)
+		influxDBRegistry.RetriesCounter().With("service", "test").Add(1)
+		influxDBRegistry.ReqDurationHistogram().With("service", "test", "code", strconv.Itoa(http.StatusOK)).Observe(10000)
+	})
+
+	extractAndMatchMessage(t, expected, msg)
+}
+
+func extractAndMatchMessage(t *testing.T, patterns []string, msg string) {
+	t.Helper()
+	for _, pattern := range patterns {
+		re := regexp.MustCompile(pattern)
+		match := re.FindStringSubmatch(msg)
+		if len(match) != 2 {
+			t.Errorf("Got %q %v, want %q", msg, match, pattern)
+		}
+	}
+}

metrics/statsd.go

@@ -17,6 +17,12 @@ var statsdClient = statsd.New("traefik.", kitlog.LoggerFunc(func(keyvals ...inte
 
 var statsdTicker *time.Ticker
 
+const (
+	statsdMetricsReqsName    = "requests.total"
+	statsdMetricsLatencyName = "request.duration"
+	statsdRetriesTotalName   = "backend.retries.total"
+)
+
 // RegisterStatsd registers the metrics pusher if this didn't happen yet and creates a statsd Registry instance.
 func RegisterStatsd(config *types.Statsd) Registry {
 	if statsdTicker == nil {
@@ -25,9 +31,9 @@ func RegisterStatsd(config *types.Statsd) Registry {
 
 	return &standardRegistry{
 		enabled:              true,
-		reqsCounter:          statsdClient.NewCounter(ddMetricsReqsName, 1.0),
-		reqDurationHistogram: statsdClient.NewTiming(ddMetricsLatencyName, 1.0),
-		retriesCounter:       statsdClient.NewCounter(ddRetriesTotalName, 1.0),
+		reqsCounter:          statsdClient.NewCounter(statsdMetricsReqsName, 1.0),
+		reqDurationHistogram: statsdClient.NewTiming(statsdMetricsLatencyName, 1.0),
+		retriesCounter:       statsdClient.NewCounter(statsdRetriesTotalName, 1.0),
 	}
 }
 

metrics/statsd_test.go

@@ -18,7 +18,7 @@ func TestStatsD(t *testing.T) {
 	defer StopStatsd()
 
 	if !statsdRegistry.IsEnabled() {
-		t.Errorf("PrometheusRegistry should return true for IsEnabled()")
+		t.Errorf("Statsd registry should return true for IsEnabled()")
 	}
 
 	expected := []string{