diff --git a/plugins/inputs/influxdb/README.md b/plugins/inputs/influxdb/README.md index 2bab123f8..711503245 100644 --- a/plugins/inputs/influxdb/README.md +++ b/plugins/inputs/influxdb/README.md @@ -20,6 +20,10 @@ InfluxDB-formatted endpoints. See below for more information. "http://localhost:8086/debug/vars" ] + ## Username and password to send using HTTP Basic Authentication. + # username = "" + # password = "" + ## Optional TLS Config # tls_ca = "/etc/telegraf/ca.pem" # tls_cert = "/etc/telegraf/cert.pem" diff --git a/plugins/inputs/influxdb/influxdb.go b/plugins/inputs/influxdb/influxdb.go index 0bb3ead5e..96389a013 100644 --- a/plugins/inputs/influxdb/influxdb.go +++ b/plugins/inputs/influxdb/influxdb.go @@ -1,9 +1,10 @@ package influxdb import ( + "bytes" "encoding/json" "errors" - "fmt" + "io" "net/http" "sync" "time" @@ -14,9 +15,28 @@ import ( "github.com/influxdata/telegraf/plugins/inputs" ) +const ( + maxErrorResponseBodyLength = 1024 +) + +type APIError struct { + StatusCode int + Reason string + Description string `json:"error"` +} + +func (e *APIError) Error() string { + if e.Description != "" { + return e.Reason + ": " + e.Description + } + return e.Reason +} + type InfluxDB struct { - URLs []string `toml:"urls"` - Timeout internal.Duration + URLs []string `toml:"urls"` + Username string `toml:"username"` + Password string `toml:"password"` + Timeout internal.Duration `toml:"timeout"` tls.ClientConfig client *http.Client @@ -38,6 +58,10 @@ func (*InfluxDB) SampleConfig() string { "http://localhost:8086/debug/vars" ] + ## Username and password to send using HTTP Basic Authentication. + # username = "" + # password = "" + ## Optional TLS Config # tls_ca = "/etc/telegraf/ca.pem" # tls_cert = "/etc/telegraf/cert.pem" @@ -75,7 +99,7 @@ func (i *InfluxDB) Gather(acc telegraf.Accumulator) error { go func(url string) { defer wg.Done() if err := i.gatherURL(acc, url); err != nil { - acc.AddError(fmt.Errorf("[url=%s]: %s", url, err)) + acc.AddError(err) } }(u) } @@ -135,12 +159,27 @@ func (i *InfluxDB) gatherURL( shardCounter := 0 now := time.Now() - resp, err := i.client.Get(url) + req, err := http.NewRequest("GET", url, nil) + if err != nil { + return err + } + + if i.Username != "" || i.Password != "" { + req.SetBasicAuth(i.Username, i.Password) + } + + req.Header.Set("User-Agent", "Telegraf/"+internal.Version()) + + resp, err := i.client.Do(req) if err != nil { return err } defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return readResponseError(resp) + } + // It would be nice to be able to decode into a map[string]point, but // we'll get a decoder error like: // `json: cannot unmarshal array into Go value of type influxdb.point` @@ -255,6 +294,27 @@ func (i *InfluxDB) gatherURL( return nil } +func readResponseError(resp *http.Response) error { + apiError := &APIError{ + StatusCode: resp.StatusCode, + Reason: resp.Status, + } + + var buf bytes.Buffer + r := io.LimitReader(resp.Body, maxErrorResponseBodyLength) + _, err := buf.ReadFrom(r) + if err != nil { + return apiError + } + + err = json.Unmarshal(buf.Bytes(), apiError) + if err != nil { + return apiError + } + + return apiError +} + func init() { inputs.Add("influxdb", func() telegraf.Input { return &InfluxDB{ diff --git a/plugins/inputs/influxdb/influxdb_test.go b/plugins/inputs/influxdb/influxdb_test.go index f24ecc24c..9225c45b0 100644 --- a/plugins/inputs/influxdb/influxdb_test.go +++ b/plugins/inputs/influxdb/influxdb_test.go @@ -1,6 +1,7 @@ package influxdb_test import ( + "fmt" "net/http" "net/http/httptest" "testing" @@ -178,6 +179,31 @@ func TestErrorHandling404(t *testing.T) { require.Error(t, acc.GatherError(plugin.Gather)) } +func TestErrorResponse(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusUnauthorized) + w.Write([]byte(`{"error": "unable to parse authentication credentials"}`)) + })) + defer ts.Close() + + plugin := &influxdb.InfluxDB{ + URLs: []string{ts.URL}, + } + + var acc testutil.Accumulator + err := plugin.Gather(&acc) + require.NoError(t, err) + + expected := []error{ + &influxdb.APIError{ + StatusCode: http.StatusUnauthorized, + Reason: fmt.Sprintf("%d %s", http.StatusUnauthorized, http.StatusText(http.StatusUnauthorized)), + Description: "unable to parse authentication credentials", + }, + } + require.Equal(t, expected, acc.Errors) +} + const basicJSON = ` { "_1": {