Metrics: Add support for InfluxDB Database / RetentionPolicy and HTTP client

metrics/influxdb.go

@@ -2,6 +2,9 @@ package metrics
 
 import (
 	"bytes"
+	"fmt"
+	"net/url"
+	"regexp"
 	"time"
 
 	"github.com/containous/traefik/log"
@@ -12,10 +15,7 @@ import (
 	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
-}))
+var influxDBClient *influx.Influx
 
 type influxDBWriter struct {
 	buf    bytes.Buffer
@@ -41,6 +41,9 @@ const (
 
 // RegisterInfluxDB registers the metrics pusher if this didn't happen yet and creates a InfluxDB Registry instance.
 func RegisterInfluxDB(config *types.InfluxDB) Registry {
+	if influxDBClient == nil {
+		influxDBClient = initInfluxDBClient(config)
+	}
 	if influxDBTicker == nil {
 		influxDBTicker = initInfluxDBTicker(config)
 	}
@@ -62,7 +65,48 @@ func RegisterInfluxDB(config *types.InfluxDB) Registry {
 	}
 }
 
-// initInfluxDBTicker initializes metrics pusher and creates a influxDBClient if not created already
+// initInfluxDBTicker creates a influxDBClient
+func initInfluxDBClient(config *types.InfluxDB) *influx.Influx {
+	// TODO deprecated: move this switch into configuration.SetEffectiveConfiguration when web provider will be removed.
+	switch config.Protocol {
+	case "udp":
+		if len(config.Database) > 0 || len(config.RetentionPolicy) > 0 {
+			log.Warn("Database and RetentionPolicy are only used when protocol is http.")
+			config.Database = ""
+			config.RetentionPolicy = ""
+		}
+	case "http":
+		if u, err := url.Parse(config.Address); err == nil {
+			if u.Scheme != "http" && u.Scheme != "https" {
+				log.Warnf("InfluxDB address %s should specify a scheme of http or https, defaulting to http.", config.Address)
+				config.Address = "http://" + config.Address
+			}
+		} else {
+			log.Errorf("Unable to parse influxdb address: %v, defaulting to udp.", err)
+			config.Protocol = "udp"
+			config.Database = ""
+			config.RetentionPolicy = ""
+		}
+	default:
+		log.Warnf("Unsupported protocol: %s, defaulting to udp.", config.Protocol)
+		config.Protocol = "udp"
+		config.Database = ""
+		config.RetentionPolicy = ""
+	}
+
+	return influx.New(
+		map[string]string{},
+		influxdb.BatchPointsConfig{
+			Database:        config.Database,
+			RetentionPolicy: config.RetentionPolicy,
+		},
+		kitlog.LoggerFunc(func(keyvals ...interface{}) error {
+			log.Info(keyvals)
+			return nil
+		}))
+}
+
+// initInfluxDBTicker initializes metrics pusher
 func initInfluxDBTicker(config *types.InfluxDB) *time.Ticker {
 	pushInterval, err := time.ParseDuration(config.PushInterval)
 	if err != nil {
@@ -88,16 +132,69 @@ func StopInfluxDB() {
 	influxDBTicker = nil
 }
 
+// Write creates a http or udp client and attempts to write BatchPoints.
+// If a "database not found" error is encountered, a CREATE DATABASE
+// query is attempted when using protocol http.
 func (w *influxDBWriter) Write(bp influxdb.BatchPoints) error {
-	c, err := influxdb.NewUDPClient(influxdb.UDPConfig{
-		Addr: w.config.Address,
-	})
-
+	c, err := w.initWriteClient()
 	if err != nil {
 		return err
 	}
 
 	defer c.Close()
 
-	return c.Write(bp)
+	if writeErr := c.Write(bp); writeErr != nil {
+		log.Errorf("Error writing to influx: %s", writeErr.Error())
+		if handleErr := w.handleWriteError(c, writeErr); handleErr != nil {
+			return handleErr
+		}
+		// Retry write after successful handling of writeErr
+		return c.Write(bp)
+	}
+	return nil
+}
+
+func (w *influxDBWriter) initWriteClient() (c influxdb.Client, err error) {
+	if w.config.Protocol == "http" {
+		c, err = influxdb.NewHTTPClient(influxdb.HTTPConfig{
+			Addr: w.config.Address,
+		})
+	} else {
+		c, err = influxdb.NewUDPClient(influxdb.UDPConfig{
+			Addr: w.config.Address,
+		})
+	}
+	return
+}
+
+func (w *influxDBWriter) handleWriteError(c influxdb.Client, writeErr error) error {
+	if w.config.Protocol != "http" {
+		return writeErr
+	}
+
+	match, matchErr := regexp.MatchString("database not found", writeErr.Error())
+
+	if matchErr != nil || !match {
+		return writeErr
+	}
+
+	qStr := fmt.Sprintf("CREATE DATABASE \"%s\"", w.config.Database)
+	if w.config.RetentionPolicy != "" {
+		qStr = fmt.Sprintf("%s WITH NAME \"%s\"", qStr, w.config.RetentionPolicy)
+	}
+
+	log.Debugf("Influx database does not exist, attempting to create with query: %s", qStr)
+
+	q := influxdb.NewQuery(qStr, "", "")
+	response, queryErr := c.Query(q)
+	if queryErr == nil && response.Error() != nil {
+		queryErr = response.Error()
+	}
+	if queryErr != nil {
+		log.Errorf("Error creating InfluxDB database: %s", queryErr)
+		return queryErr
+	}
+
+	log.Debugf("Successfully created influx database: %s", w.config.Database)
+	return nil
 }