InfluxDB: Fix returning InfluxDB error messages (#89973)

* Revert "Chore: Return influxdb query error early before parsing the result (#88549)"

This reverts commit a87c155c06.

* Handle error in buffered parser

* handle error message in streaming parser

pkg/tsdb/influxdb/influxql/buffered/response_parser.go

@@ -23,6 +23,14 @@ func ResponseParse(buf io.ReadCloser, statusCode int, query *models.Query) *back
 func parse(buf io.Reader, statusCode int, query *models.Query) *backend.DataResponse {
 	response, jsonErr := parseJSON(buf)
 
+	if statusCode/100 != 2 {
+		errorStr := response.Error
+		if errorStr == "" {
+			errorStr = response.Message
+		}
+		return &backend.DataResponse{Error: fmt.Errorf("InfluxDB returned error: %s", errorStr)}
+	}
+
 	if jsonErr != nil {
 		return &backend.DataResponse{Error: jsonErr}
 	}

pkg/tsdb/influxdb/influxql/buffered/response_parser_test.go

@@ -338,9 +338,15 @@ func TestInfluxdbResponseParser(t *testing.T) {
 	})
 
 	t.Run("Influxdb response parser with top-level error", func(t *testing.T) {
-		result := ResponseParse(readJsonFile("error_on_top_level_response"), 200, generateQuery("Test raw query", "time_series", ""))
+		result := ResponseParse(readJsonFile("error_on_top_level_response"), 400, generateQuery("Test raw query", "time_series", ""))
 		require.Nil(t, result.Frames)
-		require.EqualError(t, result.Error, "error parsing query: found THING")
+		require.EqualError(t, result.Error, "InfluxDB returned error: error parsing query: found THING")
+	})
+
+	t.Run("Influxdb response parser with error message", func(t *testing.T) {
+		result := ResponseParse(readJsonFile("invalid_response"), 400, generateQuery("Test raw query", "time_series", ""))
+		require.Nil(t, result.Frames)
+		require.EqualError(t, result.Error, "InfluxDB returned error: failed to parse query: found WERE, expected ; at line 1, char 38")
 	})
 
 	t.Run("Influxdb response parser parseNumber nil", func(t *testing.T) {

pkg/tsdb/influxdb/influxql/converter/converter.go

@@ -34,6 +34,26 @@ l1Fields:
 			if rsp.Error != nil {
 				return rsp
 			}
+		case "error":
+			v, err := iter.ReadString()
+			if err != nil {
+				rsp.Error = err
+			} else {
+				rsp.Error = fmt.Errorf(v)
+			}
+			return rsp
+		case "code":
+			// we only care of the message
+			_, err := iter.Read()
+			if err != nil {
+				return rspErr(err)
+			}
+		case "message":
+			v, err := iter.Read()
+			if err != nil {
+				return rspErr(err)
+			}
+			return rspErr(fmt.Errorf("%s", v))
 		case "":
 			if err != nil {
 				return rspErr(err)
@@ -41,11 +61,15 @@ l1Fields:
 			break l1Fields
 		default:
 			v, err := iter.Read()
-			if err != nil {
-				rsp.Error = err
-				return rsp
-			}
 			fmt.Printf("[ROOT] unsupported key: %s / %v\n\n", l1Field, v)
+			if err != nil {
+				if rsp != nil {
+					rsp.Error = err
+					return rsp
+				} else {
+					return rspErr(err)
+				}
+			}
 		}
 	}
 

pkg/tsdb/influxdb/influxql/influxql.go

@@ -177,11 +177,6 @@ func execute(ctx context.Context, tracer trace.Tracer, dsInfo *models.Datasource
 	if err != nil {
 		return backend.DataResponse{}, err
 	}
-
-	if res.StatusCode/100 != 2 {
-		return backend.DataResponse{Error: fmt.Errorf("InfluxDB returned error: %v", res.Body)}, nil
-	}
-
 	defer func() {
 		if err := res.Body.Close(); err != nil {
 			logger.Warn("Failed to close response body", "err", err)

pkg/tsdb/influxdb/influxql/querydata/stream_parser.go

@@ -23,6 +23,10 @@ func ResponseParse(buf io.ReadCloser, statusCode int, query *models.Query) *back
 	iter := jsoniter.Parse(jsoniter.ConfigDefault, buf, 1024)
 	r := converter.ReadInfluxQLStyleResult(iter, query)
 
+	if statusCode/100 != 2 {
+		return &backend.DataResponse{Error: fmt.Errorf("InfluxDB returned error: %s", r.Error)}
+	}
+
 	// The ExecutedQueryString can be viewed in QueryInspector in UI
 	for i, frame := range r.Frames {
 		if i == 0 {

pkg/tsdb/influxdb/influxql/querydata/stream_parser_test.go

@@ -105,3 +105,11 @@ func TestParsingAsTimeSeriesWithoutTimeColumn(t *testing.T) {
 		runQuery(t, f, "cardinality", "time_series", query)
 	})
 }
+
+func TestInfluxDBStreamingParser(t *testing.T) {
+	t.Run("Influxdb response parser with error message", func(t *testing.T) {
+		result := ResponseParse(readJsonFile("invalid_response"), 400, generateQuery("Test raw query", "time_series", ""))
+		require.Nil(t, result.Frames)
+		require.EqualError(t, result.Error, "InfluxDB returned error: failed to parse query: found WERE, expected ; at line 1, char 38")
+	})
+}

pkg/tsdb/influxdb/influxql/testdata/invalid_response.json

@@ -0,0 +1,4 @@
+{
+  "code": "invalid",
+  "message": "failed to parse query: found WERE, expected ; at line 1, char 38"
+}

pkg/tsdb/influxdb/models/models.go

@@ -32,6 +32,8 @@ type Select []QueryPart
 type Response struct {
 	Results []Result
 	Error   string
+	Code    string
+	Message string
 }
 
 type Result struct {