feat: add influxv1 auth.

Author: kavinli

bulk_load/load.go

@@ -40,6 +40,9 @@ type BulkLoad interface {
 
 type LoadRunner struct {
 	DbName                 string
+	User                   string
+	Password               string
+	BasicAuthentication    string
 	Workers                int
 	ItemLimit              int64
 	BatchSize              int
@@ -110,6 +113,9 @@ func (r *LoadRunner) notifyHandler(arg int) (int, error) {
 
 func (r *LoadRunner) Init(defaultBatchSize int) {
 	flag.StringVar(&r.DbName, "db", "benchmark_db", "Database name.")
+	flag.StringVar(&r.User, "user", "", "User name, credentials as query parameters.")
+	flag.StringVar(&r.Password, "password", "", "User password, credentials as query parameters.")
+	flag.StringVar(&r.BasicAuthentication, "basic-authentication", "", "Authenticate with basic authentication. format [user:password].")
 	flag.IntVar(&r.BatchSize, "batch-size", defaultBatchSize, "Batch size (1 line of input = 1 item).")
 	flag.IntVar(&r.Workers, "workers", 1, "Number of parallel requests to make.")
 	flag.Int64Var(&r.ItemLimit, "item-limit", -1, "Number of items to read from stdin before quitting. (1 item per 1 line of input.)")

cmd/bulk_load_influx/http_writer.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"log"
 	"time"
+	"encoding/base64"
 
 	"github.com/valyala/fasthttp"
 )
@@ -32,6 +33,10 @@ type HTTPWriterConfig struct {
 	// Name of the target database into which points will be written.
 	Database string
 
+	User string
+	Password string
+	BasicAuthentication string 
+
 	// Id of the target bucket into which points will be written. (InfluxDB v2)
 	BucketId string
 
@@ -86,6 +91,9 @@ func (w *HTTPWriter) WriteLineProtocol(body []byte, isGzip bool) (int64, error)
 	if w.c.AuthToken != "" {
 		req.Header.Add("Authorization", fmt.Sprintf("Token %s", w.c.AuthToken))
 	}
+	if w.c.BasicAuthentication != "" {
+		req.Header.Add("Authorization", fmt.Sprintf("Basic %s", base64.StdEncoding.EncodeToString([]byte(w.c.BasicAuthentication))))
+	}
 	req.SetBody(body)
 
 	resp := fasthttp.AcquireResponse()

cmd/bulk_load_influx/main.go

@@ -290,10 +290,16 @@ func (l *InfluxBulkLoad) PrepareProcess(i int) {
 			DebugInfo:      fmt.Sprintf("worker #%d, dest url: %s", i, l.configs[i].url),
 			Host:           l.configs[i].url,
 			Database:       bulk_load.Runner.DbName,
+			User:           bulk_load.Runner.User,
+			Password:       bulk_load.Runner.Password,
+			BasicAuthentication: bulk_load.Runner.BasicAuthentication,
 			BackingOffChan: l.configs[i].backingOffChan,
 			BackingOffDone: l.configs[i].backingOffDone,
 		}
 		url = c.Host + "/write?consistency=" + l.consistency + "&db=" + neturl.QueryEscape(c.Database)
+		if len(c.User) != 0 && len(c.Password) != 0 {
+			url = url + "&u=" + neturl.QueryEscape(c.User) + "&p=" + neturl.QueryEscape(c.Password)
+		}
 	}
 	l.configs[i].writer = NewHTTPWriter(*c, url)
 }