Create db through relay

Author: Lucas SANTONI

README.md

@@ -112,6 +112,7 @@ ssl-combined-pem = "/path/to/influxdb-relay.pem"
 output = [
     # name: name of the backend, used for display purposes only.
     # location: full URL of the /write endpoint of the backend
+    # admin: full URL of the /query endpoint of the backend. OPTIONAL.
     # timeout: Go-parseable time duration. Fail writes if incomplete in this time.
     # skip-tls-verification: skip verification for HTTPS location. WARNING: it's insecure. Don't use in production.
     # type: type of input source. OPTIONAL: see below for more information.
@@ -155,6 +156,43 @@ InfluxDB Relay is able to forward from a variety of input sources, including:
 
 The `type` parameter in the configuration file defaults to `influxdb`.
 
+### Administrative tasks
+
+Whereas data manipulation relies on the `/write` endpoint, some other features
+such as database or user management are based on the `/query` endpoint. As
+InfluxDB Relay does not send back a response body to the client(s), we are not
+able to forward all of the features this endpoint provides. Still, we decided
+to expose it through the `/admin` route.
+
+In order to use it, one shall define the optional `admin` parameter of the
+target backend(s) in the configuration file, as follows:
+
+```toml
+[[http]]
+# Name of the HTTP server, used for display purposes only.
+name = "expose-database-creation"
+
+# TCP address to bind to, for HTTP server.
+bind-addr = "127.0.0.1:9096"
+
+# Array of InfluxDB instances to use as backends for Relay.
+output = [
+    # InfluxDB
+    { name="local-influxdb01", location="http://127.0.0.1:8086/write", admin="http://127.0.0.1:8086/query", type="influxdb" },
+    { name="local-influxdb02", location="http://127.0.0.1:7086/write", admin="http://127.0.0.1:7086/query", type="influxdb" },
+]
+```
+
+It is now possible to query the `/admin` endpoint. Its usage is the same as the
+standard `/query` Influx DB enpoint:
+
+```
+curl -X POST "http://127.0.0.1:9096/admin" --data-urlencode 'q=CREATE DATABASE some_database'
+```
+
+Errors will be logged just like regular `/write` queries. The HTTP response
+bodies will not be forwarded back to the clients.
+
 ## Limitations
 
 So far, this is compatible with Debian, RedHat, and other derivatives.

config/config.go

@@ -1,6 +1,8 @@
 package config
 
 import (
+	"errors"
+	"net/url"
 	"os"
 
 	"github.com/naoina/toml"
@@ -43,6 +45,10 @@ type HTTPOutputConfig struct {
 	// Location should be set to the URL of the backend server's write endpoint
 	Location string `toml:"location"`
 
+	// Query should be set to the URL of the backend server's query endpoint
+	// This is used for database creation through relay
+	Admin string `toml:"admin"`
+
 	// Timeout sets a per-backend timeout for write requests (default: 10s)
 	// The format used is the same seen in time.ParseDuration
 	Timeout string `toml:"timeout"`
@@ -113,6 +119,11 @@ func LoadConfigFile(filename string) (Config, error) {
 				if backend.InputType == "" {
 					cfg.HTTPRelays[index].Outputs[indexB].InputType = TypeInfluxdb
 				}
+
+				_, err := url.Parse(backend.Admin)
+				if err != nil {
+					return cfg, errors.New("invalid query parameter for backend " + backend.Name)
+				}
 			}
 		}
 	}

dev/prepare_databases.sh

@@ -1,7 +1,7 @@
 #! /usr/bin/env bash
 
-curl -X POST "http://127.0.0.1:8086/query" --data-urlencode 'q=CREATE DATABASE NOT_prometheus'
-curl -X POST "http://127.0.0.1:8086/query" --data-urlencode 'q=CREATE DATABASE prometheus'
+curl -v -X POST "http://127.0.0.1:9096/admin" --data-urlencode 'q=CREATE DATABASE NOT_prometheus'
+curl -v -X POST "http://127.0.0.1:9096/admin" --data-urlencode 'q=CREATE DATABASE prometheus'
 
 # curl -X POST "http://127.0.0.1:8086/query" --data-urlencode 'db=prometheus' --data-urlencode 'q=SHOW SERIES'
 # curl -X POST "http://127.0.0.1:8086/query" --data-urlencode 'db=NOT_prometheus' --data-urlencode 'q=SHOW SERIES'

dev/prometheus.yml

@@ -3,7 +3,7 @@ remote_write:
 
 scrape_configs:
   - job_name: 'AUTOPARSE'
-    scrape_interval: 1ms
+    scrape_interval: 10s
     static_configs:
       - targets: ['localhost:9090']
         labels:

dev/test.toml

@@ -2,6 +2,6 @@
 name = "test-config"
 bind-addr = "0.0.0.0:9096"
 output = [
-    { name="from_influx", location = "http://influxdb:8086/write" },
+    { name="from_influx", location = "http://influxdb:8086/write", admin = "http://influxdb:8086/query" },
     { name="from_prometheus", location = "http://influxdb:8086/api/v1/prom/write", type="prometheus" }
 ]

relay/http.go

@@ -60,6 +60,7 @@ var (
 		"/api/v1/prom/write": (*HTTP).handleProm,
 		"/ping":              (*HTTP).handlePing,
 		"/status":            (*HTTP).handleStatus,
+		"/admin":             (*HTTP).handleAdmin,
 	}
 
 	middlewares = []relayMiddleware{
@@ -278,6 +279,7 @@ type httpBackend struct {
 	poster
 	name      string
 	inputType config.Input
+	admin     string
 }
 
 func newHTTPBackend(cfg *config.HTTPOutputConfig) (*httpBackend, error) {
@@ -321,6 +323,7 @@ func newHTTPBackend(cfg *config.HTTPOutputConfig) (*httpBackend, error) {
 		poster:    p,
 		name:      cfg.Name,
 		inputType: cfg.InputType,
+		admin:     cfg.Admin,
 	}, nil
 }
 

relay/http_handlers.go

@@ -40,6 +40,100 @@ func (h *HTTP) handlePing(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
+func (h *HTTP) handleAdmin(w http.ResponseWriter, r *http.Request) {
+	// Client to perform the raw queries
+	client := http.Client{}
+
+	if r.Method == http.MethodPost {
+		// Responses
+		var responses = make(chan *http.Response, len(h.backends))
+
+		// Associated waitgroup
+		var wg sync.WaitGroup
+		wg.Add(len(h.backends))
+
+		// Iterate over all backends
+		for _, b := range h.backends {
+			b := b
+
+			if b.admin == "" {
+				// Empty query, skip backend
+				wg.Done()
+				continue
+			}
+
+			go func() {
+				defer wg.Done()
+
+				// Create new request
+				// Update location according to backend
+				// Forward body
+				req, err := http.NewRequest("POST", b.admin, r.Body)
+				if err != nil {
+					jsonResponse(w, response{http.StatusServiceUnavailable, "could not prepare request: " + err.Error()})
+				}
+
+				// Forward headers
+				req.Header = r.Header
+
+				// Forward the request
+				resp, err := client.Do(req)
+				if err != nil {
+					// Internal error
+					log.Printf("Problem posting to relay %q backend %q: %v", h.Name(), b.name, err)
+
+					// So empty response
+					responses <- &http.Response{}
+				} else {
+					if resp.StatusCode / 100 == 5 {
+						// HTTP error
+						log.Printf("5xx response for relay %q backend %q: %v", h.Name(), b.name, resp.StatusCode)
+					}
+
+					// Get response
+					responses <- resp
+				}
+			}()
+		}
+
+		// Wait for requests
+		go func() {
+			wg.Wait()
+			close(responses)
+		}()
+
+		var errResponse *responseData
+		for resp := range responses {
+			switch resp.StatusCode / 100 {
+			case 2:
+				w.WriteHeader(http.StatusNoContent)
+				return
+
+			case 4:
+				// User error
+				resp.Write(w)
+				return
+
+			default:
+				// Hold on to one of the responses to return back to the client
+				errResponse = nil
+			}
+		}
+
+		// No successful writes
+		if errResponse == nil {
+			// Failed to make any valid request...
+			jsonResponse(w, response{http.StatusServiceUnavailable, "unable to forward query"})
+			return
+		}
+
+		errResponse.Write(w)
+	} else { // Bad method
+		jsonResponse(w, response{http.StatusMethodNotAllowed, http.StatusText(http.StatusMethodNotAllowed)})
+		return
+	}
+}
+
 func (h *HTTP) handleStandard(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodPost {
 		w.Header().Set("Allow", http.MethodPost)