Read an invalid JSON response as an error in the influx client

CHANGELOG.md

@@ -6,6 +6,7 @@
 - [#7270](https://github.com/influxdata/influxdb/issues/7270): Implement time math for lazy time literals.
 - [#7272](https://github.com/influxdata/influxdb/issues/7272): Report cmdline and memstats in /debug/vars.
 - [#7299](https://github.com/influxdata/influxdb/ssues/7299): Ensure fieldsCreated stat available in shard measurement.
+- [#6846](https://github.com/influxdata/influxdb/issues/6846): Read an invalid JSON response as an error in the influx client.
 
 ## v1.0.0 [2016-09-07]
 

client/influxdb.go

@@ -498,17 +498,36 @@ func (r *Response) Error() error {
 	return nil
 }
 
+// duplexReader reads responses and writes it to another writer while
+// satisfying the reader interface.
+type duplexReader struct {
+	r io.Reader
+	w io.Writer
+}
+
+func (r *duplexReader) Read(p []byte) (n int, err error) {
+	n, err = r.r.Read(p)
+	if err == nil {
+		r.w.Write(p[:n])
+	}
+	return n, err
+}
+
 // ChunkedResponse represents a response from the server that
 // uses chunking to stream the output.
 type ChunkedResponse struct {
-	dec *json.Decoder
+	dec    *json.Decoder
+	duplex *duplexReader
+	buf    bytes.Buffer
 }
 
 // NewChunkedResponse reads a stream and produces responses from the stream.
 func NewChunkedResponse(r io.Reader) *ChunkedResponse {
-	dec := json.NewDecoder(r)
+	resp := &ChunkedResponse{}
-	dec.UseNumber()
+	resp.duplex = &duplexReader{r: r, w: &resp.buf}
-	return &ChunkedResponse{dec: dec}
+	resp.dec = json.NewDecoder(resp.duplex)
+	resp.dec.UseNumber()
+	return resp
 }
 
 // NextResponse reads the next line of the stream and returns a response.
@@ -518,8 +537,13 @@ func (r *ChunkedResponse) NextResponse() (*Response, error) {
 		if err == io.EOF {
 			return nil, nil
 		}
-		return nil, err
+		// A decoding error happened. This probably means the server crashed
+		// and sent a last-ditch error message to us. Ensure we have read the
+		// entirety of the connection to get any remaining error text.
+		io.Copy(ioutil.Discard, r.duplex)
+		return nil, errors.New(strings.TrimSpace(r.buf.String()))
 	}
+	r.buf.Reset()
 	return &response, nil
 }