Refactor Response Parser

pkg/opensearch/lucene_handler.go

@@ -21,15 +21,17 @@ type luceneHandler struct {
 	intervalCalculator tsdb.IntervalCalculator
 	ms                 *es.MultiSearchRequestBuilder
 	queries            []*Query
+	dsSettings         *backend.DataSourceInstanceSettings
 }
 
-func newLuceneHandler(client es.Client, queries []backend.DataQuery, intervalCalculator tsdb.IntervalCalculator) *luceneHandler {
+func newLuceneHandler(client es.Client, queries []backend.DataQuery, intervalCalculator tsdb.IntervalCalculator, dsSettings *backend.DataSourceInstanceSettings) *luceneHandler {
 	return &luceneHandler{
 		client:             client,
 		reqQueries:         queries,
 		intervalCalculator: intervalCalculator,
 		ms:                 client.MultiSearch(),
 		queries:            make([]*Query, 0),
+		dsSettings:         dsSettings,
 	}
 }
 
@@ -278,8 +280,8 @@ func (h *luceneHandler) executeQueries(ctx context.Context) (*backend.QueryDataR
 		return nil, err
 	}
 
-	rp := newResponseParser(res.Responses, h.queries, res.DebugInfo)
-	return rp.getTimeSeries(h.client.GetConfiguredFields())
+	rp := newResponseParser(res.Responses, h.queries, res.DebugInfo, h.client.GetConfiguredFields(), h.dsSettings)
+	return rp.parseResponse()
 }
 
 func addDateHistogramAgg(aggBuilder es.AggBuilder, bucketAgg *BucketAgg, timeFrom, timeTo int64, timeField string) es.AggBuilder {

pkg/opensearch/opensearch.go

@@ -64,7 +64,7 @@ func (ds *OpenSearchDatasource) QueryData(ctx context.Context, req *backend.Quer
 		return nil, err
 	}
 
-	query := newTimeSeriesQuery(osClient, req.Queries, intervalCalculator)
+	query := newQueryRequest(osClient, req.Queries, req.PluginContext.DataSourceInstanceSettings, intervalCalculator)
 	response, err := wrapError(query.execute(ctx))
 	return response, err
 }

pkg/opensearch/time_series_query.go → pkg/opensearch/query_request.go

@@ -12,27 +12,29 @@ import (
 	"github.com/grafana/opensearch-datasource/pkg/utils"
 )
 
-type timeSeriesQuery struct {
+type queryRequest struct {
 	client             es.Client
-	tsdbQueries        []backend.DataQuery
+	queries            []backend.DataQuery
+	dsSettings         *backend.DataSourceInstanceSettings
 	intervalCalculator tsdb.IntervalCalculator
 }
 
-func newTimeSeriesQuery(client es.Client, query []backend.DataQuery, intervalCalculator tsdb.IntervalCalculator) *timeSeriesQuery {
-	return &timeSeriesQuery{
+func newQueryRequest(client es.Client, queries []backend.DataQuery, dsSettings *backend.DataSourceInstanceSettings, intervalCalculator tsdb.IntervalCalculator) *queryRequest {
+	return &queryRequest{
 		client:             client,
-		tsdbQueries:        query,
+		queries:            queries,
+		dsSettings:         dsSettings,
 		intervalCalculator: intervalCalculator,
 	}
 }
 
-func (e *timeSeriesQuery) execute(ctx context.Context) (*backend.QueryDataResponse, error) {
+func (e *queryRequest) execute(ctx context.Context) (*backend.QueryDataResponse, error) {
 	handlers := make(map[string]queryHandler)
 
-	handlers[Lucene] = newLuceneHandler(e.client, e.tsdbQueries, e.intervalCalculator)
-	handlers[PPL] = newPPLHandler(e.client, e.tsdbQueries)
+	handlers[Lucene] = newLuceneHandler(e.client, e.queries, e.intervalCalculator, e.dsSettings)
+	handlers[PPL] = newPPLHandler(e.client, e.queries)
 
-	queries, err := parse(e.tsdbQueries)
+	queries, err := parse(e.queries)
 	if err != nil {
 		return nil, err
 	}

pkg/opensearch/time_series_query_test.go → pkg/opensearch/query_request_test.go

@@ -969,7 +969,9 @@ func executeTsdbQuery(c es.Client, body string, from, to time.Time, minInterval
 			},
 		},
 	}
-	query := newTimeSeriesQuery(c, tsdbQuery, tsdb.NewIntervalCalculator(&tsdb.IntervalOptions{MinInterval: minInterval}))
+
+	dsSettings := backend.DataSourceInstanceSettings{}
+	query := newQueryRequest(c, tsdbQuery, &dsSettings, tsdb.NewIntervalCalculator(&tsdb.IntervalOptions{MinInterval: minInterval}))
 	return query.execute(context.Background())
 }
 

pkg/opensearch/response_parser.go

@@ -42,35 +42,43 @@ const (
 )
 
 type responseParser struct {
-	Responses []*es.SearchResponse
-	Targets   []*Query
-	DebugInfo *es.SearchDebugInfo
+	Responses        []*es.SearchResponse
+	Targets          []*Query
+	DebugInfo        *es.SearchDebugInfo
+	ConfiguredFields es.ConfiguredFields
+	DSSettings       *backend.DataSourceInstanceSettings
 }
 
-func newResponseParser(responses []*es.SearchResponse, targets []*Query, debugInfo *es.SearchDebugInfo) *responseParser {
+func newResponseParser(responses []*es.SearchResponse, targets []*Query, debugInfo *es.SearchDebugInfo, configuredFields es.ConfiguredFields, dsSettings *backend.DataSourceInstanceSettings) *responseParser {
 	return &responseParser{
-		Responses: responses,
-		Targets:   targets,
-		DebugInfo: debugInfo,
+		Responses:        responses,
+		Targets:          targets,
+		DebugInfo:        debugInfo,
+		ConfiguredFields: configuredFields,
+		DSSettings:       dsSettings,
 	}
 }
 
-func (rp *responseParser) getTimeSeries(configuredFields es.ConfiguredFields) (*backend.QueryDataResponse, error) {
+func (rp *responseParser) parseResponse() (*backend.QueryDataResponse, error) {
 	result := backend.NewQueryDataResponse()
 
 	if rp.Responses == nil {
 		return result, nil
 	}
 
+	// go through each response, create data frames based on type, add them to result
 	for i, res := range rp.Responses {
+		// grab the associated query
 		target := rp.Targets[i]
 
-		var debugInfo *simplejson.Json
-		if rp.DebugInfo != nil && i == 0 {
-			debugInfo = utils.NewJsonFromAny(rp.DebugInfo)
-		}
-
+		// if one of the responses is an error add debug info and error
+		// and keep trying to process other responses
 		if res.Error != nil {
+			var debugInfo *simplejson.Json
+			if rp.DebugInfo != nil && i == 0 {
+				debugInfo = utils.NewJsonFromAny(rp.DebugInfo)
+			}
+
 			result.Responses[target.RefID] = backend.DataResponse{
 				Error: getErrorFromOpenSearchResponse(res),
 				Frames: []*data.Frame{
@@ -98,11 +106,11 @@ func (rp *responseParser) getTimeSeries(configuredFields es.ConfiguredFields) (*
 
 		switch target.Metrics[0].Type {
 		case rawDataType:
-			queryRes = processRawDataResponse(res, configuredFields, queryRes)
+			queryRes = processRawDataResponse(res, rp.ConfiguredFields, queryRes)
 		case rawDocumentType:
 			queryRes = processRawDocumentResponse(res, target.RefID, queryRes)
 		case logsType:
-			queryRes = processLogsResponse(res, configuredFields, queryRes)
+			queryRes = processLogsResponse(res, rp.ConfiguredFields, queryRes)
 		default:
 			props := make(map[string]string)
 			err := rp.processBuckets(res.Aggregations, target, &queryRes, props, 0)