Improve error handling (#177)

* - SQL errors should be downstream
- add more PGX specific errors

* - better check for retryable errors

* - cleanup the approach

Author: sasklacz

datasource.go

@@ -268,6 +268,14 @@ func (ds *SQLDatasource) handleQuery(ctx context.Context, req backend.DataQuery,
 		}
 	}
 
+	// Check if the error is retryable and convert to downstream error if so
+	if errors.Is(err, ErrorQuery) && shouldRetry(ds.DriverSettings().RetryOn, err.Error()) {
+		// Convert retryable errors to downstream errors
+		if !backend.IsDownstreamError(err) {
+			err = backend.DownstreamError(err)
+		}
+	}
+
 	// allow retries on timeouts
 	if errors.Is(err, context.DeadlineExceeded) {
 		for i := 0; i < ds.DriverSettings().Retries; i++ {

datasource_test.go

@@ -269,7 +269,7 @@ func Test_query_panic_in_rows_validation(t *testing.T) {
 
 	res := data.Responses["foo"]
 	assert.NotNil(t, res.Error)
-	assert.Contains(t, res.Error.Error(), "failed to validate rows")
+	assert.Contains(t, res.Error.Error(), "SQL rows validation failed")
 	assert.NotNil(t, res.Frames) // Error frame is returned, not nil
 }
 

errors.go

@@ -2,6 +2,7 @@ package sqlds
 
 import (
 	"errors"
+	"strings"
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 )
@@ -17,6 +18,12 @@ var (
 	ErrorTimeout = errors.New("query timeout exceeded")
 	// ErrorNoResults is returned if there were no results returned
 	ErrorNoResults = errors.New("no results returned from query")
+	// ErrorRowValidation is returned when SQL rows validation fails (e.g., connection issues, corrupt results)
+	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 {
@@ -25,3 +32,83 @@ 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
+}
+
+// IsGenericDownstreamError checks if an error is a generic downstream error
+func IsGenericDownstreamError(err error) bool {
+	if err == nil {
+		return false
+	}
+
+	errStr := strings.ToLower(err.Error())
+	genericDownstreamErrors := []string{
+		"invalid memory address",
+		"nil pointer dereference",
+	}
+
+	for _, genericErr := range genericDownstreamErrors {
+		if strings.Contains(errStr, genericErr) {
+			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 generic downstream errors first
+	if IsGenericDownstreamError(err) {
+		return backend.ErrorSourceDownstream, err
+	}
+
+	// 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

@@ -78,12 +78,29 @@ func (q *DBQuery) Run(ctx context.Context, query *Query, args ...interface{}) (d
 	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
+
 		if errors.Is(err, context.Canceled) {
 			errType = context.Canceled
+			errSource = backend.ErrorSourcePlugin
+		} else if IsPGXConnectionError(err) {
+			errType = ErrorPGXLifecycle
+			errSource = backend.ErrorSourceDownstream
+		} else {
+			// Use enhanced error classification for PGX v5
+			errSource, _ = ClassifyError(err)
+		}
+
+		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()))
 		}
-		errWithSource := backend.DownstreamError(fmt.Errorf("%w: %s", errType, err.Error()))
-		q.metrics.CollectDuration(SourceDownstream, StatusError, time.Since(start).Seconds())
+
+		q.metrics.CollectDuration(Source(errSource), StatusError, time.Since(start).Seconds())
 		return sqlutil.ErrorFrameFromQuery(query), errWithSource
 	}
 	q.metrics.CollectDuration(SourceDownstream, StatusOK, time.Since(start).Seconds())
@@ -96,8 +113,16 @@ func (q *DBQuery) Run(ctx context.Context, query *Query, args ...interface{}) (d
 			errWithSource := backend.DownstreamError(fmt.Errorf("%s: %w", "No results from query", err))
 			return sqlutil.ErrorFrameFromQuery(query), errWithSource
 		}
-		errWithSource := backend.DownstreamError(fmt.Errorf("%s: %w", "Error response from database", err))
-		q.metrics.CollectDuration(SourceDownstream, StatusError, time.Since(start).Seconds())
+
+		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))
+		}
+
+		q.metrics.CollectDuration(Source(errSource), StatusError, time.Since(start).Seconds())
 		return sqlutil.ErrorFrameFromQuery(query), errWithSource
 	}
 
@@ -111,11 +136,13 @@ func (q *DBQuery) Run(ctx context.Context, query *Query, args ...interface{}) (d
 	// Convert the response to frames
 	res, err := getFrames(rows, q.rowLimit, q.converters, q.fillMode, query)
 	if err != nil {
-		// We default to plugin error source
-		errSource := backend.ErrorSourcePlugin
+		errSource, _ := ClassifyError(err)
+
+		// Additional checks for processing errors
 		if backend.IsDownstreamHTTPError(err) || isProcessingDownstreamError(err) {
 			errSource = backend.ErrorSourceDownstream
 		}
+
 		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
@@ -212,12 +239,12 @@ func accessColumns(rows *sql.Rows) (columnErr error) {
 // validateRows performs safety checks on SQL rows to prevent panics
 func validateRows(rows *sql.Rows) error {
 	if rows == nil {
-		return fmt.Errorf("rows is nil")
+		return fmt.Errorf("%w: rows is nil", ErrorRowValidation)
 	}
 
 	err := accessColumns(rows)
 	if err != nil {
-		return fmt.Errorf("failed to validate rows: %w", err)
+		return fmt.Errorf("%w: %w", ErrorRowValidation, err)
 	}
 	return nil
 }
@@ -285,11 +312,25 @@ func isProcessingDownstreamError(err error) bool {
 		data.ErrorInputFieldsWithoutRows,
 		data.ErrorSeriesUnsorted,
 		data.ErrorNullTimeValues,
+		ErrorRowValidation,
+		ErrorConnectionClosed,
+		ErrorPGXLifecycle,
 	}
 	for _, e := range downstreamErrors {
 		if errors.Is(err, e) {
 			return true
 		}
 	}
+
+	// Check for generic downstream errors
+	if IsGenericDownstreamError(err) {
+		return true
+	}
+
+	// Check for PGX connection errors
+	if IsPGXConnectionError(err) {
+		return true
+	}
+
 	return false
 }

query_test.go

@@ -231,3 +231,122 @@ func TestIsProcessingDownstreamError(t *testing.T) {
 		})
 	}
 }
+
+func TestPGXErrorClassification(t *testing.T) {
+	tests := []struct {
+		name           string
+		errorMsg       string
+		expectedSource backend.ErrorSource
+		expectedIsPGX  bool
+	}{
+		{
+			name:           "nil pointer dereference",
+			errorMsg:       "runtime error: invalid memory address or nil pointer dereference",
+			expectedSource: backend.ErrorSourceDownstream,
+			expectedIsPGX:  false, // Now handled as generic downstream error
+		},
+		{
+			name:           "connection closed",
+			errorMsg:       "connection closed",
+			expectedSource: backend.ErrorSourceDownstream,
+			expectedIsPGX:  true,
+		},
+		{
+			name:           "broken pipe",
+			errorMsg:       "broken pipe",
+			expectedSource: backend.ErrorSourceDownstream,
+			expectedIsPGX:  true,
+		},
+		{
+			name:           "pgconn error",
+			errorMsg:       "pgconn: connection failed",
+			expectedSource: backend.ErrorSourceDownstream,
+			expectedIsPGX:  true,
+		},
+		{
+			name:           "regular SQL error",
+			errorMsg:       "syntax error at position 1",
+			expectedSource: backend.ErrorSourcePlugin,
+			expectedIsPGX:  false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := errors.New(tt.errorMsg)
+
+			// Test IsPGXConnectionError
+			isPGX := IsPGXConnectionError(err)
+			if isPGX != tt.expectedIsPGX {
+				t.Errorf("IsPGXConnectionError() = %v, expected %v", isPGX, tt.expectedIsPGX)
+			}
+
+			// Test ClassifyError
+			source, _ := ClassifyError(err)
+			if source != tt.expectedSource {
+				t.Errorf("ClassifyError() source = %v, expected %v", source, tt.expectedSource)
+			}
+
+			// Test isProcessingDownstreamError
+			if tt.expectedSource == backend.ErrorSourceDownstream {
+				isDownstream := isProcessingDownstreamError(err)
+				if !isDownstream {
+					t.Errorf("isProcessingDownstreamError() = %v, expected true for downstream error", isDownstream)
+				}
+			}
+		})
+	}
+}
+
+func TestIsGenericDownstreamError(t *testing.T) {
+	tests := []struct {
+		name     string
+		errorMsg string
+		expected bool
+	}{
+		{
+			name:     "nil pointer dereference",
+			errorMsg: "runtime error: invalid memory address or nil pointer dereference",
+			expected: true,
+		},
+		{
+			name:     "invalid memory address",
+			errorMsg: "runtime error: invalid memory address",
+			expected: true,
+		},
+		{
+			name:     "nil pointer dereference uppercase",
+			errorMsg: "NIL POINTER DEREFERENCE",
+			expected: true,
+		},
+		{
+			name:     "regular SQL error",
+			errorMsg: "syntax error at position 1",
+			expected: false,
+		},
+		{
+			name:     "connection error",
+			errorMsg: "connection closed",
+			expected: false,
+		},
+		{
+			name:     "nil error",
+			errorMsg: "",
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			var err error
+			if tt.errorMsg != "" {
+				err = errors.New(tt.errorMsg)
+			}
+
+			result := IsGenericDownstreamError(err)
+			if result != tt.expected {
+				t.Errorf("IsGenericDownstreamError(%v) = %v, expected %v", tt.errorMsg, result, tt.expected)
+			}
+		})
+	}
+}