Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions relay/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -52,6 +52,12 @@ type HTTPOutputConfig struct {
// Skip TLS verification in order to use self signed certificate.
// WARNING: It's insecure. Use it only for developing and don't use in production.
SkipTLSVerification bool `toml:"skip-tls-verification"`

// If specified, overrides any client Authorization headers sent
HTTPUser string `toml:"http-user"`

// If specified, overrides any client AUthorization headers sent
HTTPPass string `toml:"http-pass"`
}

type UDPConfig struct {
Expand Down
25 changes: 20 additions & 5 deletions relay/http.go
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ import (
"bytes"
"compress/gzip"
"crypto/tls"
"encoding/base64"
"errors"
"fmt"
"io/ioutil"
Expand Down Expand Up @@ -114,16 +115,15 @@ func (h *HTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) {
start := time.Now()

if r.URL.Path == "/ping" && (r.Method == "GET" || r.Method == "HEAD") {
w.Header().Add("X-InfluxDB-Version", "relay")
w.WriteHeader(http.StatusNoContent)
return
w.Header().Add("X-InfluxDB-Version", "relay")
w.WriteHeader(http.StatusNoContent)
return
}

if r.URL.Path != "/write" {
jsonError(w, http.StatusNotFound, "invalid write endpoint")
return
}

if r.Method != "POST" {
w.Header().Set("Allow", "POST")
if r.Method == "OPTIONS" {
Expand Down Expand Up @@ -209,7 +209,7 @@ func (h *HTTP) ServeHTTP(w http.ResponseWriter, r *http.Request) {
b := b
go func() {
defer wg.Done()
resp, err := b.post(outBytes, query, authHeader)
resp, err := b.post(outBytes, query, b.getAuthHeader(authHeader))
if err != nil {
log.Printf("Problem posting to relay %q backend %q: %v", h.Name(), b.name, err)
} else {
Expand Down Expand Up @@ -321,6 +321,7 @@ func (b *simplePoster) post(buf []byte, query string, auth string) (*responseDat
req.URL.RawQuery = query
req.Header.Set("Content-Type", "text/plain")
req.Header.Set("Content-Length", strconv.Itoa(len(buf)))

if auth != "" {
req.Header.Set("Authorization", auth)
}
Expand Down Expand Up @@ -350,6 +351,14 @@ func (b *simplePoster) post(buf []byte, query string, auth string) (*responseDat
type httpBackend struct {
poster
name string
auth string
}

func (h *httpBackend) getAuthHeader(passthru string) string {
if h.auth != "" {
return h.auth
}
return passthru
}

func newHTTPBackend(cfg *HTTPOutputConfig) (*httpBackend, error) {
Expand Down Expand Up @@ -388,9 +397,15 @@ func newHTTPBackend(cfg *HTTPOutputConfig) (*httpBackend, error) {
p = newRetryBuffer(cfg.BufferSizeMB*MB, batch, max, p)
}

auth := ""
if cfg.HTTPUser != "" && cfg.HTTPPass != "" {
auth = fmt.Sprintf("Basic %s", base64.StdEncoding.EncodeToString([]byte(fmt.Sprintf("%s:%s", cfg.HTTPUser, cfg.HTTPPass))))
}

return &httpBackend{
poster: p,
name: cfg.Name,
auth: auth,
}, nil
}

Expand Down