tsdbrelay: Relaying metadata to secondary relays. Also making relay that sets host header properly.

cmd/bosun/main.go

@@ -7,7 +7,6 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
-	"net/http/httputil"
 	_ "net/http/pprof"
 	"net/url"
 	"os"
@@ -106,7 +105,7 @@ func main() {
 	if c.RelayListen != "" {
 		go func() {
 			mux := http.NewServeMux()
-			mux.Handle("/api/", httputil.NewSingleHostReverseProxy(httpListen))
+			mux.Handle("/api/", util.NewSingleHostProxy(httpListen))
 			s := &http.Server{
 				Addr:    c.RelayListen,
 				Handler: mux,
@@ -123,7 +122,7 @@ func main() {
 			Host:   c.TSDBHost,
 		}
 		if *flagReadonly {
-			rp := httputil.NewSingleHostReverseProxy(tsdbHost)
+			rp := util.NewSingleHostProxy(tsdbHost)
 			ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 				if r.URL.Path == "/api/put" {
 					w.WriteHeader(204)

cmd/bosun/web/web.go

@@ -24,6 +24,7 @@ import (
 	"bosun.org/metadata"
 	"bosun.org/opentsdb"
 	"bosun.org/slog"
+	"bosun.org/util"
 	"bosun.org/version"
 )
 
@@ -163,7 +164,7 @@ func (rp *relayProxy) ServeHTTP(responseWriter http.ResponseWriter, r *http.Requ
 }
 
 func Relay(dest string) http.Handler {
-	return &relayProxy{ReverseProxy: httputil.NewSingleHostReverseProxy(&url.URL{
+	return &relayProxy{ReverseProxy: util.NewSingleHostProxy(&url.URL{
 		Scheme: "http",
 		Host:   dest,
 	})}

cmd/tsdbrelay/main.go

@@ -17,6 +17,7 @@ import (
 
 	"bosun.org/cmd/tsdbrelay/denormalize"
 	"bosun.org/opentsdb"
+	"bosun.org/util"
 )
 
 var (
@@ -81,8 +82,6 @@ func main() {
 		Path:   "/api/index",
 	}
 	bosunIndexURL = u.String()
-	tsdbProxy := httputil.NewSingleHostReverseProxy(tsdbURL)
-
 	if *secondaryRelays != "" {
 		for _, rUrl := range strings.Split(*secondaryRelays, ",") {
 			u = url.URL{
@@ -94,10 +93,18 @@ func main() {
 		}
 	}
 
-	http.Handle("/api/put", &relayProxy{
-		ReverseProxy: tsdbProxy,
+	tsdbProxy := util.NewSingleHostProxy(tsdbURL)
+	bosunProxy := util.NewSingleHostProxy(bosunURL)
+	rp := &relayProxy{
+		TSDBProxy:  tsdbProxy,
+		BosunProxy: bosunProxy,
+	}
+	http.HandleFunc("/api/put", func(w http.ResponseWriter, r *http.Request) {
+		rp.relayPut(w, r, true)
+	})
+	http.HandleFunc("/api/metadata/put", func(w http.ResponseWriter, r *http.Request) {
+		rp.relayMetadata(w, r)
 	})
-	http.Handle("/api/metadata/put", httputil.NewSingleHostReverseProxy(bosunURL))
 	http.Handle("/", tsdbProxy)
 	log.Fatal(http.ListenAndServe(*listenAddr, nil))
 }
@@ -109,7 +116,8 @@ func verbose(format string, a ...interface{}) {
 }
 
 type relayProxy struct {
-	*httputil.ReverseProxy
+	TSDBProxy  *httputil.ReverseProxy
+	BosunProxy *httputil.ReverseProxy
 }
 
 type passthru struct {
@@ -133,21 +141,17 @@ func (rw *relayWriter) WriteHeader(code int) {
 	rw.ResponseWriter.WriteHeader(code)
 }
 
-func (rp *relayProxy) ServeHTTP(responseWriter http.ResponseWriter, r *http.Request) {
-	rp.relayRequest(responseWriter, r, true)
-}
-
 var (
 	relayHeader = "X-Relayed-From"
 	myHost      string
 )
 
-func (rp *relayProxy) relayRequest(responseWriter http.ResponseWriter, r *http.Request, parse bool) {
+func (rp *relayProxy) relayPut(responseWriter http.ResponseWriter, r *http.Request, parse bool) {
 	isRelayed := r.Header.Get(relayHeader) != ""
 	reader := &passthru{ReadCloser: r.Body}
 	r.Body = reader
 	w := &relayWriter{ResponseWriter: responseWriter}
-	rp.ReverseProxy.ServeHTTP(w, r)
+	rp.TSDBProxy.ServeHTTP(w, r)
 	if w.code != 204 {
 		verbose("got status", w.code)
 		return
@@ -179,12 +183,12 @@ func (rp *relayProxy) relayRequest(responseWriter http.ResponseWriter, r *http.R
 			for _, relayUrl := range relayPutUrls {
 				body := bytes.NewBuffer(reader.buf.Bytes())
 				req, err := http.NewRequest(r.Method, relayUrl, body)
-				req.Header.Set("Content-Type", "application/json")
-				req.Header.Set("Content-Encoding", "gzip")
 				if err != nil {
 					verbose("%v", err)
 					return
 				}
+				req.Header.Set("Content-Type", "application/json")
+				req.Header.Set("Content-Encoding", "gzip")
 				req.Header.Add(relayHeader, myHost)
 				resp, err := http.DefaultClient.Do(req)
 				if err != nil {
@@ -245,7 +249,46 @@ func (rp *relayProxy) denormalize(body io.Reader) {
 	req.Header.Set("Content-Encoding", "gzip")
 
 	responseWriter := httptest.NewRecorder()
-	rp.relayRequest(responseWriter, req, false)
+	rp.relayPut(responseWriter, req, false)
 
 	verbose("relayed %d denormalized data points. Tsdb response: %d", len(relayDps), responseWriter.Code)
 }
+
+func (rp *relayProxy) relayMetadata(responseWriter http.ResponseWriter, r *http.Request) {
+
+	reader := &passthru{ReadCloser: r.Body}
+	r.Body = reader
+	w := &relayWriter{ResponseWriter: responseWriter}
+	rp.BosunProxy.ServeHTTP(w, r)
+	if w.code != 204 {
+		verbose("got status", w.code)
+		return
+	}
+	verbose("relayed metadata to bosun")
+	if r.Header.Get(relayHeader) != "" {
+		return
+	}
+	if len(relayPutUrls) != 0 {
+		go func() {
+			for _, relayUrl := range relayPutUrls {
+				relayUrl = strings.Replace(relayUrl, "/put", "/metadata/put", 1)
+				body := bytes.NewBuffer(reader.buf.Bytes())
+				req, err := http.NewRequest(r.Method, relayUrl, body)
+				if err != nil {
+					verbose("%v", err)
+					return
+				}
+				req.Header.Set("Content-Type", "application/json")
+				req.Header.Set("Content-Encoding", "gzip")
+				req.Header.Add(relayHeader, myHost)
+				resp, err := http.DefaultClient.Do(req)
+				if err != nil {
+					verbose("secondary relay error: %v", err)
+					return
+				}
+				resp.Body.Close()
+				verbose("secondary relay success")
+			}
+		}()
+	}
+}

util/proxy.go

@@ -0,0 +1,18 @@
+package util
+
+import (
+	"net/http"
+	"net/http/httputil"
+	"net/url"
+)
+
+// Creates a new http Proxy that forwards requests to the specified url.
+// Differs from httputil.NewSingleHostReverseProxy only in that it properly sets the host header.
+func NewSingleHostProxy(target *url.URL) *httputil.ReverseProxy {
+	proxy := httputil.NewSingleHostReverseProxy(target)
+	director := func(req *http.Request) {
+		proxy.Director(req)
+		req.Host = target.Host
+	}
+	return &httputil.ReverseProxy{Director: director}
+}