QueryErrorMutator interface (#190)

* remove pgx errors and add query error mutator interface

* remove empty test

* add retry logic back with tests

* add to change log

* this is actually a breaking change

* this is actually a breaking change

* add titles like other change log lines

Author: bossinc

CHANGELOG.md

@@ -5,6 +5,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [5.0.0]
+
+### Added
+- Added QueryErrorMutator (#190)
+
+### Changed
+- **Breaking change** IsPGXConnectionError has been removed. Removed PGX v5/PostgreSQL-specific error detection (#190)
+
 ## [4.2.7]
 
 ### Changed

datasource.go

@@ -225,12 +225,17 @@ func (ds *SQLDatasource) handleQuery(ctx context.Context, req backend.DataQuery,
 		args = argSetter.SetQueryArgs(ctx, headers)
 	}
 
+	var queryErrorMutator QueryErrorMutator
+	if mutator, ok := ds.driver().(QueryErrorMutator); ok {
+		queryErrorMutator = mutator
+	}
+
 	// FIXES:
 	//  * Some datasources (snowflake) expire connections or have an authentication token that expires if not used in 1 or 4 hours.
 	//    Because the datasource driver does not include an option for permanent connections, we retry the connection
 	//    if the query fails. NOTE: this does not include some errors like "ErrNoRows"
 	dbQuery := NewQuery(dbConn.db, dbConn.settings, ds.driver().Converters(), fillMode, ds.rowLimit)
-	res, err := dbQuery.Run(ctx, q, args...)
+	res, err := dbQuery.Run(ctx, q, queryErrorMutator, args...)
 	if err == nil {
 		return res, nil
 	}
@@ -256,7 +261,7 @@ func (ds *SQLDatasource) handleQuery(ctx context.Context, req backend.DataQuery,
 				}
 
 				dbQuery := NewQuery(db, dbConn.settings, ds.driver().Converters(), fillMode, ds.rowLimit)
-				res, err = dbQuery.Run(ctx, q, args...)
+				res, err = dbQuery.Run(ctx, q, queryErrorMutator, args...)
 				if err == nil {
 					return res, err
 				}
@@ -286,7 +291,7 @@ func (ds *SQLDatasource) handleQuery(ctx context.Context, req backend.DataQuery,
 			}
 
 			dbQuery := NewQuery(db, dbConn.settings, ds.driver().Converters(), fillMode, ds.rowLimit)
-			res, err = dbQuery.Run(ctx, q, args...)
+			res, err = dbQuery.Run(ctx, q, queryErrorMutator, args...)
 			if err == nil {
 				return res, err
 			}

driver.go

@@ -73,3 +73,7 @@ type QueryArgSetter interface {
 type ResponseMutator interface {
 	MutateResponse(ctx context.Context, res data.Frames) (data.Frames, error)
 }
+
+type QueryErrorMutator interface {
+	MutateQueryError(err error) backend.ErrorWithSource
+}

errors.go

@@ -2,7 +2,6 @@ package sqlds
 
 import (
 	"errors"
-	"strings"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 )
@@ -22,8 +21,6 @@ var (
 	ErrorRowValidation = errors.New("SQL rows validation failed")
 	// ErrorConnectionClosed is returned when the database connection is unexpectedly closed
 	ErrorConnectionClosed = errors.New("database connection closed")
-	// ErrorPGXLifecycle is returned for PGX v5 specific connection lifecycle issues
-	ErrorPGXLifecycle = errors.New("PGX connection lifecycle error")
 )
 
 func ErrorSource(err error) backend.ErrorSource {
@@ -32,57 +29,3 @@ func ErrorSource(err error) backend.ErrorSource {
 	}
 	return backend.ErrorSourcePlugin
 }
-
-// IsPGXConnectionError checks if an error is related to PGX v5 connection issues
-func IsPGXConnectionError(err error) bool {
-	if err == nil {
-		return false
-	}
-
-	errStr := strings.ToLower(err.Error())
-	pgxConnectionErrors := []string{
-		"connection closed",
-		"connection reset",
-		"connection refused",
-		"broken pipe",
-		"eof",
-		"context canceled",
-		"context deadline exceeded",
-		"pgconn",
-		"conn is closed",
-		"bad connection",
-	}
-
-	for _, pgxErr := range pgxConnectionErrors {
-		if strings.Contains(errStr, pgxErr) {
-			return true
-		}
-	}
-
-	return false
-}
-
-// ClassifyError determines the appropriate error source and type for SQL errors
-func ClassifyError(err error) (backend.ErrorSource, error) {
-	if err == nil {
-		return backend.ErrorSourcePlugin, nil
-	}
-
-	// Check for PGX v5 specific connection errors
-	if IsPGXConnectionError(err) {
-		// These are typically downstream connection issues
-		return backend.ErrorSourceDownstream, ErrorPGXLifecycle
-	}
-
-	// Check for row validation errors
-	if errors.Is(err, ErrorRowValidation) {
-		return backend.ErrorSourceDownstream, err
-	}
-
-	// Default to existing logic
-	if backend.IsDownstreamError(err) {
-		return backend.ErrorSourceDownstream, err
-	}
-
-	return backend.ErrorSourcePlugin, err
-}

query.go

@@ -74,55 +74,51 @@ func NewQuery(db Connection, settings backend.DataSourceInstanceSettings, conver
 }
 
 // Run sends the query to the connection and converts the rows to a dataframe.
-func (q *DBQuery) Run(ctx context.Context, query *Query, args ...interface{}) (data.Frames, error) {
+func (q *DBQuery) Run(ctx context.Context, query *Query, queryErrorMutator QueryErrorMutator, args ...interface{}) (data.Frames, error) {
 	start := time.Now()
 	rows, err := q.DB.QueryContext(ctx, query.RawSQL, args...)
 	if err != nil {
-		// Determine error source based on retry configuration and error type
-		errSource := backend.ErrorSourcePlugin
-		errType := ErrorQuery
+		var errWithSource backend.ErrorWithSource
+		defer func() {
+			q.metrics.CollectDuration(Source(errWithSource.ErrorSource()), StatusError, time.Since(start).Seconds())
+		}()
 
 		if errors.Is(err, context.Canceled) {
-			errType = context.Canceled
-			errSource = backend.ErrorSourceDownstream
-		} else if IsPGXConnectionError(err) {
-			errType = ErrorPGXLifecycle
-			errSource = backend.ErrorSourceDownstream
-		} else {
-			// Use enhanced error classification for PGX v5
-			errSource, _ = ClassifyError(err)
+			errWithSource := backend.NewErrorWithSource(err, backend.ErrorSourceDownstream)
+			return sqlutil.ErrorFrameFromQuery(query), errWithSource
 		}
 
-		var errWithSource error
-		if errSource == backend.ErrorSourceDownstream {
-			errWithSource = backend.DownstreamError(fmt.Errorf("%w: %s", errType, err.Error()))
-		} else {
-			errWithSource = backend.PluginError(fmt.Errorf("%w: %s", errType, err.Error()))
+		// Wrap with ErrorQuery to enable retry logic in datasource
+		queryErr := fmt.Errorf("%w: %w", ErrorQuery, err)
+
+		// Handle driver specific errors
+		if queryErrorMutator != nil {
+			errWithSource = queryErrorMutator.MutateQueryError(queryErr)
+			return sqlutil.ErrorFrameFromQuery(query), errWithSource
 		}
 
-		q.metrics.CollectDuration(Source(errSource), StatusError, time.Since(start).Seconds())
+		// If we get to this point, assume the error is from the plugin
+		errWithSource = backend.NewErrorWithSource(queryErr, backend.DefaultErrorSource)
+
 		return sqlutil.ErrorFrameFromQuery(query), errWithSource
 	}
 	q.metrics.CollectDuration(SourceDownstream, StatusOK, time.Since(start).Seconds())
 
 	// Check for an error response
 	if err := rows.Err(); err != nil {
+		queryErr := fmt.Errorf("%w: %w", ErrorQuery, err)
+		errWithSource := backend.NewErrorWithSource(queryErr, backend.DefaultErrorSource)
 		if errors.Is(err, sql.ErrNoRows) {
 			// Should we even response with an error here?
 			// The panel will simply show "no data"
-			errWithSource := backend.DownstreamError(fmt.Errorf("%s: %w", "No results from query", err))
+			errWithSource = backend.NewErrorWithSource(fmt.Errorf("%w: %s", err, "Error response from database"), backend.ErrorSourceDownstream)
 			return sqlutil.ErrorFrameFromQuery(query), errWithSource
 		}
-
-		errSource, _ := ClassifyError(err)
-		var errWithSource error
-		if errSource == backend.ErrorSourceDownstream {
-			errWithSource = backend.DownstreamError(fmt.Errorf("%s: %w", "Error response from database", err))
-		} else {
-			errWithSource = backend.PluginError(fmt.Errorf("%s: %w", "Error response from database", err))
+		if queryErrorMutator != nil {
+			errWithSource = queryErrorMutator.MutateQueryError(queryErr)
 		}
 
-		q.metrics.CollectDuration(Source(errSource), StatusError, time.Since(start).Seconds())
+		q.metrics.CollectDuration(Source(errWithSource.ErrorSource()), StatusError, time.Since(start).Seconds())
 		return sqlutil.ErrorFrameFromQuery(query), errWithSource
 	}
 
@@ -132,23 +128,34 @@ func (q *DBQuery) Run(ctx context.Context, query *Query, args ...interface{}) (d
 		}
 	}()
 
-	start = time.Now()
-	// Convert the response to frames
+	return q.convertRowsToFrames(rows, query, queryErrorMutator)
+}
+
+func (q *DBQuery) convertRowsToFrames(rows *sql.Rows, query *Query, queryErrorMutator QueryErrorMutator) (data.Frames, error) {
+	source := SourcePlugin
+	status := StatusOK
+	start := time.Now()
+	defer func() {
+		q.metrics.CollectDuration(source, status, time.Since(start).Seconds())
+	}()
+
 	res, err := getFrames(rows, q.rowLimit, q.converters, q.fillMode, query)
 	if err != nil {
-		errSource, _ := ClassifyError(err)
+		status = StatusError
 
 		// Additional checks for processing errors
-		if backend.IsDownstreamHTTPError(err) || isProcessingDownstreamError(err) {
-			errSource = backend.ErrorSourceDownstream
+		if backend.IsDownstreamHTTPError(err) {
+			source = SourceDownstream
+		} else if queryErrorMutator != nil {
+			errWithSource := queryErrorMutator.MutateQueryError(err)
+			source = Source(errWithSource.ErrorSource())
 		}
 
-		errWithSource := backend.NewErrorWithSource(fmt.Errorf("%w: %s", err, "Could not process SQL results"), errSource)
-		q.metrics.CollectDuration(Source(errSource), StatusError, time.Since(start).Seconds())
-		return sqlutil.ErrorFrameFromQuery(query), errWithSource
+		return sqlutil.ErrorFrameFromQuery(query), backend.NewErrorWithSource(
+			fmt.Errorf("%w: %s", err, "Could not process SQL results"),
+			backend.ErrorSource(source),
+		)
 	}
-
-	q.metrics.CollectDuration(SourcePlugin, StatusOK, time.Since(start).Seconds())
 	return res, nil
 }
 
@@ -306,26 +313,3 @@ func applyHeaders(query *Query, headers http.Header) *Query {
 
 	return query
 }
-
-func isProcessingDownstreamError(err error) bool {
-	downstreamErrors := []error{
-		data.ErrorInputFieldsWithoutRows,
-		data.ErrorSeriesUnsorted,
-		data.ErrorNullTimeValues,
-		ErrorRowValidation,
-		ErrorConnectionClosed,
-		ErrorPGXLifecycle,
-	}
-	for _, e := range downstreamErrors {
-		if errors.Is(err, e) {
-			return true
-		}
-	}
-
-	// Check for PGX connection errors
-	if IsPGXConnectionError(err) {
-		return true
-	}
-
-	return false
-}

query_integration_test.go

@@ -80,7 +80,7 @@ func TestQuery_MySQL(t *testing.T) {
 		}
 
 		sqlQuery := NewQuery(db, settings, []sqlutil.Converter{}, nil, defaultRowLimit)
-		_, err := sqlQuery.Run(ctx, q)
+		_, err := sqlQuery.Run(ctx, q, nil)
 		if err == nil {
 			t.Fatal("expected an error but received none")
 		}