Make client errors more helpful on downstream errs

When a downstream server such as a proxy or loadbalancer between
influxdb and the client produces an error, the client currently does
not make this very obvious.

This change introduces checks on both the content type and the
influx version header to identify whether a request was served by
influxdb itself and returns a more appropriate error in the cases
where it can be determined a downstream issue is at play.

CHANGELOG.md

@@ -40,7 +40,8 @@
 - [#8690](https://github.com/influxdata/influxdb/issues/8690): Implicitly decide on a lower limit for fill queries when none is present.
 - [#8947](https://github.com/influxdata/influxdb/pull/8947): Add `EXPLAIN ANALYZE` command, which produces a detailed execution plan of a `SELECT` statement.
 - [#8963](https://github.com/influxdata/influxdb/pull/8963): Streaming inmem2tsi conversion.
-- [#8995](https://github.com/influxdata/influxdb/pull/8995): Sort & validate TSI key value insertion. 
+- [#8995](https://github.com/influxdata/influxdb/pull/8995): Sort & validate TSI key value insertion.
+- [#8968](https://github.com/influxdata/influxdb/issues/8968): Make client errors more helpful on downstream errs
 
 ### Bugfixes
 
@@ -153,7 +154,7 @@ The following new configuration options are available.
 * `max-body-size` was added with a default of 25,000,000, but can be disabled by setting it to 0.
   Specifies the maximum size (in bytes) of a client request body. When a client sends data that exceeds
   the configured maximum size, a `413 Request Entity Too Large` HTTP response is returned.
-  
+
 #### `[continuous_queries]` Section
 
 * `query-stats-enabled` was added with a default of `false`. When set to `true`, continuous query execution statistics are written to the default monitor store.

client/v2/client.go

@@ -9,6 +9,7 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"mime"
 	"net/http"
 	"net/url"
 	"strconv"
@@ -514,6 +515,31 @@ func (c *client) Query(q Query) (*Response, error) {
 	}
 	defer resp.Body.Close()
 
+	// If we lack a X-Influxdb-Version header, then we didn't get a response from influxdb
+	// but instead some other service. If the error code is also a 500+ code, then some
+	// downstream loadbalancer/proxy/etc had an issue and we should report that.
+	if resp.Header.Get("X-Influxdb-Version") == "" && resp.StatusCode >= http.StatusInternalServerError {
+		body, err := ioutil.ReadAll(resp.Body)
+		if err != nil || len(body) == 0 {
+			return nil, fmt.Errorf("received status code %d from downstream server", resp.StatusCode)
+		}
+
+		return nil, fmt.Errorf("received status code %d from downstream server, with response body: %q", resp.StatusCode, body)
+	}
+
+	// If we get an unexpected content type, then it is also not from influx direct and therefore
+	// we want to know what we received and what status code was returned for debugging purposes.
+	if cType, _, _ := mime.ParseMediaType(resp.Header.Get("Content-Type")); cType != "application/json" {
+		// Read up to 1kb of the body to help identify downstream errors and limit the impact of things
+		// like downstream serving a large file
+		body, err := ioutil.ReadAll(io.LimitReader(resp.Body, 1024))
+		if err != nil || len(body) == 0 {
+			return nil, fmt.Errorf("expected json response, got %q, with status: %v", cType, resp.StatusCode)
+		}
+
+		return nil, fmt.Errorf("expected json response, got %q, with status: %v and response body: %q", cType, resp.StatusCode, body)
+	}
+
 	var response Response
 	if q.Chunked {
 		cr := NewChunkedResponse(resp.Body)
@@ -548,11 +574,11 @@ func (c *client) Query(q Query) (*Response, error) {
 			return nil, fmt.Errorf("unable to decode json: received status code %d err: %s", resp.StatusCode, decErr)
 		}
 	}
+
 	// If we don't have an error in our json response, and didn't get statusOK
 	// then send back an error
 	if resp.StatusCode != http.StatusOK && response.Error() == nil {
-		return &response, fmt.Errorf("received status code %d from server",
-			resp.StatusCode)
+		return &response, fmt.Errorf("received status code %d from server", resp.StatusCode)
 	}
 	return &response, nil
 }

client/v2/client_test.go

@@ -139,6 +139,7 @@ func (w *writeLogger) Close() error { return nil }
 func TestClient_Query(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		var data Response
+		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		_ = json.NewEncoder(w).Encode(data)
 	}))
@@ -155,9 +156,126 @@ func TestClient_Query(t *testing.T) {
 	}
 }
 
+func TestClientDownstream500WithBody_Query(t *testing.T) {
+	const err500page = `<html>
+	<head>
+		<title>500 Internal Server Error</title>
+	</head>
+	<body>Internal Server Error</body>
+</html>`
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte(err500page))
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{}
+	_, err := c.Query(query)
+
+	expected := fmt.Sprintf("received status code 500 from downstream server, with response body: %q", err500page)
+	if err.Error() != expected {
+		t.Errorf("unexpected error.  expected %v, actual %v", expected, err)
+	}
+}
+
+func TestClientDownstream500_Query(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{}
+	_, err := c.Query(query)
+
+	expected := "received status code 500 from downstream server"
+	if err.Error() != expected {
+		t.Errorf("unexpected error.  expected %v, actual %v", expected, err)
+	}
+}
+
+func TestClientDownstream400WithBody_Query(t *testing.T) {
+	const err403page = `<html>
+	<head>
+		<title>403 Forbidden</title>
+	</head>
+	<body>Forbidden</body>
+</html>`
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+		w.Write([]byte(err403page))
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{}
+	_, err := c.Query(query)
+
+	expected := fmt.Sprintf(`expected json response, got "text/html", with status: %v and response body: %q`, http.StatusForbidden, err403page)
+	if err.Error() != expected {
+		t.Errorf("unexpected error.  expected %v, actual %v", expected, err)
+	}
+}
+
+func TestClientDownstream400_Query(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{}
+	_, err := c.Query(query)
+
+	expected := fmt.Sprintf(`expected json response, got "text/plain", with status: %v`, http.StatusForbidden)
+	if err.Error() != expected {
+		t.Errorf("unexpected error.  expected %v, actual %v", expected, err)
+	}
+}
+
+func TestClient500_Query(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.Header().Set("X-Influxdb-Version", "1.3.1")
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte(`{"error":"test"}`))
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{}
+	resp, err := c.Query(query)
+
+	if err != nil {
+		t.Errorf("unexpected error.  expected nothing, actual %v", err)
+	}
+
+	if resp.Err != "test" {
+		t.Errorf(`unexpected response error.  expected "test", actual %v`, resp.Err)
+	}
+}
+
 func TestClient_ChunkedQuery(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		var data Response
+		w.Header().Set("Content-Type", "application/json")
+		w.Header().Set("X-Influxdb-Version", "1.3.1")
 		w.WriteHeader(http.StatusOK)
 		enc := json.NewEncoder(w)
 		_ = enc.Encode(data)
@@ -178,12 +296,130 @@ func TestClient_ChunkedQuery(t *testing.T) {
 	}
 }
 
+func TestClientDownstream500WithBody_ChunkedQuery(t *testing.T) {
+	const err500page = `<html>
+	<head>
+		<title>500 Internal Server Error</title>
+	</head>
+	<body>Internal Server Error</body>
+</html>`
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte(err500page))
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, err := NewHTTPClient(config)
+	if err != nil {
+		t.Fatalf("unexpected error.  expected %v, actual %v", nil, err)
+	}
+
+	query := Query{Chunked: true}
+	_, err = c.Query(query)
+
+	expected := fmt.Sprintf("received status code 500 from downstream server, with response body: %q", err500page)
+	if err.Error() != expected {
+		t.Errorf("unexpected error.  expected %v, actual %v", expected, err)
+	}
+}
+
+func TestClientDownstream500_ChunkedQuery(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{Chunked: true}
+	_, err := c.Query(query)
+
+	expected := "received status code 500 from downstream server"
+	if err.Error() != expected {
+		t.Errorf("unexpected error.  expected %v, actual %v", expected, err)
+	}
+}
+
+func TestClient500_ChunkedQuery(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		w.Header().Set("X-Influxdb-Version", "1.3.1")
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte(`{"error":"test"}`))
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{Chunked: true}
+	resp, err := c.Query(query)
+
+	if err != nil {
+		t.Errorf("unexpected error.  expected nothing, actual %v", err)
+	}
+
+	if resp.Err != "test" {
+		t.Errorf(`unexpected response error.  expected "test", actual %v`, resp.Err)
+	}
+}
+
+func TestClientDownstream400WithBody_ChunkedQuery(t *testing.T) {
+	const err403page = `<html>
+	<head>
+		<title>403 Forbidden</title>
+	</head>
+	<body>Forbidden</body>
+</html>`
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+		w.Write([]byte(err403page))
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{Chunked: true}
+	_, err := c.Query(query)
+
+	expected := fmt.Sprintf(`expected json response, got "text/html", with status: %v and response body: %q`, http.StatusForbidden, err403page)
+	if err.Error() != expected {
+		t.Errorf("unexpected error.  expected %v, actual %v", expected, err)
+	}
+}
+
+func TestClientDownstream400_ChunkedQuery(t *testing.T) {
+	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusForbidden)
+	}))
+	defer ts.Close()
+
+	config := HTTPConfig{Addr: ts.URL}
+	c, _ := NewHTTPClient(config)
+	defer c.Close()
+
+	query := Query{Chunked: true}
+	_, err := c.Query(query)
+
+	expected := fmt.Sprintf(`expected json response, got "text/plain", with status: %v`, http.StatusForbidden)
+	if err.Error() != expected {
+		t.Errorf("unexpected error.  expected %v, actual %v", expected, err)
+	}
+}
+
 func TestClient_BoundParameters(t *testing.T) {
 	var parameterString string
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		var data Response
 		r.ParseForm()
 		parameterString = r.FormValue("params")
+		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		_ = json.NewEncoder(w).Encode(data)
 	}))
@@ -233,6 +469,7 @@ func TestClient_BasicAuth(t *testing.T) {
 			t.Errorf("unexpected password, expected %q, actual %q", "password", p)
 		}
 		var data Response
+		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		_ = json.NewEncoder(w).Encode(data)
 	}))
@@ -252,6 +489,7 @@ func TestClient_BasicAuth(t *testing.T) {
 func TestClient_Ping(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		var data Response
+		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusNoContent)
 		_ = json.NewEncoder(w).Encode(data)
 	}))
@@ -269,6 +507,7 @@ func TestClient_Ping(t *testing.T) {
 
 func TestClient_Concurrent_Use(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		w.Write([]byte(`{}`))
 	}))
@@ -381,6 +620,7 @@ func TestClient_UserAgent(t *testing.T) {
 		receivedUserAgent = r.UserAgent()
 
 		var data Response
+		w.Header().Set("Content-Type", "application/json")
 		w.WriteHeader(http.StatusOK)
 		_ = json.NewEncoder(w).Encode(data)
 	}))

stress/stress_test.go

@@ -389,6 +389,7 @@ var basicQC = &BasicQueryClient{
 func TestBasicQueryClient_Query(t *testing.T) {
 	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		time.Sleep(50 * time.Millisecond)
+		w.Header().Set("Content-Type", "application/json; charset=utf-8")
 		w.Header().Set("X-Influxdb-Version", "x.x")
 		w.Header().Set("X-Influxdb-Build", "OSS")
 		var data client.Response