Elasticsearch: Add error source for DataQuery (#77386)

* WIP

* Refactor, plus update source of error in response_parser

* Adjust test

* Use methods and httpclient from errorsource

* Update pkg/tsdb/elasticsearch/data_query.go

Co-authored-by: Scott Lepper <scott.lepper@gmail.com>

* Return nil error

* Fix test

* Fix integration test

---------

Co-authored-by: Scott Lepper <scott.lepper@gmail.com>

pkg/tests/api/elasticsearch/elasticsearch_test.go

@@ -95,7 +95,7 @@ func TestIntegrationElasticsearch(t *testing.T) {
 		resp, err := http.Post(u, "application/json", buf1)
 		require.NoError(t, err)
 
-		require.Equal(t, http.StatusInternalServerError, resp.StatusCode)
+		require.Equal(t, http.StatusBadRequest, resp.StatusCode)
 		t.Cleanup(func() {
 			err := resp.Body.Close()
 			require.NoError(t, err)

pkg/tsdb/elasticsearch/data_query.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana-plugin-sdk-go/experimental/errorsource"
 
 	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/infra/log"
@@ -41,12 +42,13 @@ var newElasticsearchDataQuery = func(ctx context.Context, client es.Client, data
 
 func (e *elasticsearchDataQuery) execute() (*backend.QueryDataResponse, error) {
 	start := time.Now()
+	response := backend.NewQueryDataResponse()
 	e.logger.Debug("Parsing queries", "queriesLength", len(e.dataQueries))
 	queries, err := parseQuery(e.dataQueries, e.logger)
 	if err != nil {
 		mq, _ := json.Marshal(e.dataQueries)
 		e.logger.Error("Failed to parse queries", "error", err, "queries", string(mq), "queriesLength", len(queries), "duration", time.Since(start), "stage", es.StagePrepareRequest)
-		return &backend.QueryDataResponse{}, err
+		return errorsource.AddPluginErrorToResponse(e.dataQueries[0].RefID, response, err), nil
 	}
 
 	ms := e.client.MultiSearch()
@@ -57,7 +59,7 @@ func (e *elasticsearchDataQuery) execute() (*backend.QueryDataResponse, error) {
 		if err := e.processQuery(q, ms, from, to); err != nil {
 			mq, _ := json.Marshal(q)
 			e.logger.Error("Failed to process query to multisearch request builder", "error", err, "query", string(mq), "queriesLength", len(queries), "duration", time.Since(start), "stage", es.StagePrepareRequest)
-			return &backend.QueryDataResponse{}, err
+			return errorsource.AddPluginErrorToResponse(q.RefID, response, err), nil
 		}
 	}
 
@@ -65,13 +67,14 @@ func (e *elasticsearchDataQuery) execute() (*backend.QueryDataResponse, error) {
 	if err != nil {
 		mqs, _ := json.Marshal(e.dataQueries)
 		e.logger.Error("Failed to build multisearch request", "error", err, "queriesLength", len(queries), "queries", string(mqs), "duration", time.Since(start), "stage", es.StagePrepareRequest)
-		return &backend.QueryDataResponse{}, err
+		return errorsource.AddPluginErrorToResponse(e.dataQueries[0].RefID, response, err), nil
 	}
 
 	e.logger.Info("Prepared request", "queriesLength", len(queries), "duration", time.Since(start), "stage", es.StagePrepareRequest)
 	res, err := e.client.ExecuteMultisearch(req)
 	if err != nil {
-		return &backend.QueryDataResponse{}, err
+		// We are returning error containing the source that was added trough errorsource.Middleware
+		return errorsource.AddErrorToResponse(e.dataQueries[0].RefID, response, err), nil
 	}
 
 	return parseResponse(e.ctx, res.Responses, queries, e.client.GetConfiguredFields(), e.logger, e.tracer)

pkg/tsdb/elasticsearch/data_query_test.go

@@ -1428,10 +1428,12 @@ func TestExecuteElasticsearchDataQuery(t *testing.T) {
 
 		t.Run("With invalid query should return error", (func(t *testing.T) {
 			c := newFakeClient()
-			_, err := executeElasticsearchDataQuery(c, `{
+			res, err := executeElasticsearchDataQuery(c, `{
 				"query": "foo",
 			}`, from, to)
-			require.Error(t, err)
+			require.NoError(t, err)
+			require.Equal(t, res.Responses["A"].ErrorSource, backend.ErrorSourcePlugin)
+			require.Equal(t, res.Responses["A"].Error.Error(), "invalid character '}' looking for beginning of object key string")
 		}))
 	})
 }
@@ -1856,6 +1858,7 @@ func executeElasticsearchDataQuery(c es.Client, body string, from, to time.Time)
 			{
 				JSON:      json.RawMessage(body),
 				TimeRange: timeRange,
+				RefID:     "A",
 			},
 		},
 	}

pkg/tsdb/elasticsearch/elasticsearch.go

@@ -17,6 +17,7 @@ import (
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/backend/datasource"
 	"github.com/grafana/grafana-plugin-sdk-go/backend/instancemgmt"
+	exphttpclient "github.com/grafana/grafana-plugin-sdk-go/experimental/errorsource/httpclient"
 
 	"github.com/grafana/grafana/pkg/infra/httpclient"
 	"github.com/grafana/grafana/pkg/infra/log"
@@ -87,7 +88,8 @@ func newInstanceSettings(httpClientProvider httpclient.Provider) datasource.Inst
 			httpCliOpts.SigV4.Service = "es"
 		}
 
-		httpCli, err := httpClientProvider.New(httpCliOpts)
+		// enable experimental http client to support errors with source
+		httpCli, err := exphttpclient.New(httpCliOpts)
 		if err != nil {
 			return nil, err
 		}

pkg/tsdb/elasticsearch/error_handling_test.go

@@ -139,11 +139,12 @@ func TestNonElasticError(t *testing.T) {
 	// to access the database for some reason.
 	response := []byte(`Access to the database is forbidden`)
 
-	_, err := queryDataTestWithResponseCode(query, 403, response)
+	res, err := queryDataTestWithResponseCode(query, 403, response)
 	// FIXME: we should return something better.
 	// currently it returns the error-message about being unable to decode JSON
 	// it is not 100% clear what we should return to the browser
 	// (and what to debug-log for example), we could return
 	// at least something like "unknown response, http status code 403"
-	require.ErrorContains(t, err, "invalid character")
+	require.NoError(t, err)
+	require.Contains(t, res.response.Responses["A"].Error.Error(), "invalid character")
 }

pkg/tsdb/elasticsearch/response_parser.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/grafana/grafana-plugin-sdk-go/experimental/errorsource"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/trace"
@@ -73,9 +74,7 @@ func parseResponse(ctx context.Context, responses []*es.SearchResponse, targets
 			resSpan.End()
 			logger.Error("Processing error response from Elasticsearch", "error", string(me), "query", string(mt))
 			errResult := getErrorFromElasticResponse(res)
-			result.Responses[target.RefID] = backend.DataResponse{
-				Error: errors.New(errResult),
-			}
+			result.Responses[target.RefID] = errorsource.Response(errorsource.PluginError(errors.New(errResult), false))
 			continue
 		}