Add more errorsource attribution to InfluxDb datasource (#100969)

This PR adds errorsource attribution to the influxql and flux query paths
when the query model cannot be parsed, which is a user error.

It also catches cases where the datasource configuration does not
contain a scheme or host, and adds downstream attribution to those
errors.

Error handling on the influxql query path is updated to match 'all errors
are per query, and stashed on the response object' pattern.

Fixes https://github.com/grafana/oss-plugin-partnerships/issues/1250

pkg/tsdb/influxdb/flux/flux.go

@@ -31,7 +31,10 @@ func Query(ctx context.Context, dsInfo *models.DatasourceInfo, tsdbQuery backend
 	for _, query := range tsdbQuery.Queries {
 		qm, err := getQueryModel(query, timeRange, dsInfo)
 		if err != nil {
-			tRes.Responses[query.RefID] = backend.DataResponse{Error: err}
+			tRes.Responses[query.RefID] = backend.DataResponse{
+				Error:       err,
+				ErrorSource: backend.ErrorSourceDownstream,
+			}
 			continue
 		}
 

pkg/tsdb/influxdb/influxql/influxql.go

@@ -30,6 +30,7 @@ const (
 
 var (
 	ErrInvalidHttpMode = errors.New("'httpMode' should be either 'GET' or 'POST'")
+	ErrInvalidUrl      = errors.New("URL must contain scheme and host")
 	glog               = log.New("tsdb.influx_influxql")
 )
 
@@ -51,13 +52,17 @@ func Query(ctx context.Context, tracer trace.Tracer, dsInfo *models.DatasourceIn
 			reqQuery := req.Queries[idx]
 			query, err := models.QueryParse(reqQuery, logger)
 			if err != nil {
-				return err
+				responseLock.Lock()
+				response.Responses[query.RefID] = backend.DataResponse{
+					Error:       err,
+					ErrorSource: backend.ErrorSourceDownstream,
+				}
+				responseLock.Unlock()
+				return nil
 			}
 
-			rawQuery, err := query.Build(req)
-			if err != nil {
-				return err
-			}
+			// query.Build() unconditionally returns nil for error.
+			rawQuery, _ := query.Build(req)
 
 			query.RefID = reqQuery.RefID
 			query.RawQuery = rawQuery
@@ -68,7 +73,13 @@ func Query(ctx context.Context, tracer trace.Tracer, dsInfo *models.DatasourceIn
 
 			request, err := createRequest(ctx, logger, dsInfo, rawQuery, query.Policy)
 			if err != nil {
-				return err
+				responseLock.Lock()
+				response.Responses[query.RefID] = backend.DataResponse{
+					Error:       err,
+					ErrorSource: backend.ErrorSourceDownstream,
+				}
+				responseLock.Unlock()
+				return nil
 			}
 
 			resp, err := execute(ctx, tracer, dsInfo, logger, query, request, features.IsEnabled(ctx, featuremgmt.FlagInfluxqlStreamingParser))
@@ -90,13 +101,15 @@ func Query(ctx context.Context, tracer trace.Tracer, dsInfo *models.DatasourceIn
 		for _, reqQuery := range req.Queries {
 			query, err := models.QueryParse(reqQuery, logger)
 			if err != nil {
-				return &backend.QueryDataResponse{}, err
+				response.Responses[query.RefID] = backend.DataResponse{
+					Error:       err,
+					ErrorSource: backend.ErrorSourceDownstream,
+				}
+				continue
 			}
 
-			rawQuery, err := query.Build(req)
-			if err != nil {
-				return &backend.QueryDataResponse{}, err
-			}
+			// query.Build() unconditionally returns nil for error.
+			rawQuery, _ := query.Build(req)
 
 			query.RefID = reqQuery.RefID
 			query.RawQuery = rawQuery
@@ -107,7 +120,11 @@ func Query(ctx context.Context, tracer trace.Tracer, dsInfo *models.DatasourceIn
 
 			request, err := createRequest(ctx, logger, dsInfo, rawQuery, query.Policy)
 			if err != nil {
-				return &backend.QueryDataResponse{}, err
+				response.Responses[query.RefID] = backend.DataResponse{
+					Error:       err,
+					ErrorSource: backend.ErrorSourceDownstream,
+				}
+				continue
 			}
 
 			resp, err := execute(ctx, tracer, dsInfo, logger, query, request, features.IsEnabled(ctx, featuremgmt.FlagInfluxqlStreamingParser))
@@ -129,6 +146,13 @@ func createRequest(ctx context.Context, logger log.Logger, dsInfo *models.Dataso
 		return nil, err
 	}
 
+	// It's possible that the configuration is bad, and we'll have a URL
+	// without a scheme or host. This is valid from the PoV of the Go std
+	// library url.Parse(), but not for this data source.
+	if u.Host == "" || u.Scheme == "" {
+		return nil, ErrInvalidUrl
+	}
+
 	u.Path = path.Join(u.Path, "query")
 	httpMode := dsInfo.HTTPMode
 

pkg/tsdb/influxdb/models/model_parser.go

@@ -1,6 +1,7 @@
 package models
 
 import (
+	"errors"
 	"fmt"
 	"strconv"
 	"time"
@@ -12,6 +13,10 @@ import (
 	"github.com/grafana/grafana/pkg/infra/log"
 )
 
+var (
+	ErrInvalidQuery = errors.New("invalid InfluxDB query")
+)
+
 type InfluxdbQueryParser struct{}
 
 func QueryParse(query backend.DataQuery, logger log.Logger) (*Query, error) {
@@ -33,17 +38,17 @@ func QueryParse(query backend.DataQuery, logger log.Logger) (*Query, error) {
 
 	tags, err := parseTags(model)
 	if err != nil {
-		return nil, err
+		return nil, errors.Join(ErrInvalidQuery, err)
 	}
 
 	groupBys, err := parseGroupBy(model)
 	if err != nil {
-		return nil, err
+		return nil, errors.Join(ErrInvalidQuery, err)
 	}
 
 	selects, err := parseSelects(model)
 	if err != nil {
-		return nil, err
+		return nil, errors.Join(ErrInvalidQuery, err)
 	}
 
 	interval := query.Interval