SQL: Migrate to use SDK contracts (#36635)

* convert SQLs to use sdk contracts

* make draft

* postgres

* intermedia

* get datasourceinfo filled at the beginning of the service

* move the interval into package because of cyclict  import and fix all postgres tests

* fix mysql test

* fix mssql

* fix the test for pr https://github.com/grafana/grafana/issues/35839

* fix some issue about intervalv2 package

* update sql test

* wire migration for SQLs

* add sqls to the background process

* make it register instead of register and start

* revert formatting

* fix tests

* fix linter

* remove integration test

* Postgres test fix

Co-authored-by: Marcus Efraimsson <marcus.efraimsson@gmail.com>

pkg/tsdb/sqleng/sql_engine.go

@@ -3,6 +3,7 @@ package sqleng
 import (
 	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net"
@@ -15,11 +16,8 @@ import (
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 	"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
-	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/infra/log"
-	"github.com/grafana/grafana/pkg/models"
-	"github.com/grafana/grafana/pkg/plugins"
-	"github.com/grafana/grafana/pkg/tsdb/interval"
+	"github.com/grafana/grafana/pkg/tsdb/intervalv2"
 	"xorm.io/core"
 	"xorm.io/xorm"
 )
@@ -32,29 +30,28 @@ var ErrConnectionFailed = errors.New("failed to connect to server - please inspe
 // SQLMacroEngine interpolates macros into sql. It takes in the Query to have access to query context and
 // timeRange to be able to generate queries that use from and to.
 type SQLMacroEngine interface {
-	Interpolate(query plugins.DataSubQuery, timeRange plugins.DataTimeRange, sql string) (string, error)
+	Interpolate(query *backend.DataQuery, timeRange backend.TimeRange, sql string) (string, error)
 }
 
 // SqlQueryResultTransformer transforms a query result row to RowValues with proper types.
 type SqlQueryResultTransformer interface {
 	// TransformQueryError transforms a query error.
 	TransformQueryError(err error) error
-
 	GetConverterList() []sqlutil.StringConverter
 }
 
 type engineCacheType struct {
-	cache    map[int64]*xorm.Engine
-	versions map[int64]int
+	cache   map[int64]*xorm.Engine
+	updates map[int64]time.Time
 	sync.Mutex
 }
 
 var engineCache = engineCacheType{
-	cache:    make(map[int64]*xorm.Engine),
-	versions: make(map[int64]int),
+	cache:   make(map[int64]*xorm.Engine),
+	updates: make(map[int64]time.Time),
 }
 
-var sqlIntervalCalculator = interval.NewCalculator()
+var sqlIntervalCalculator = intervalv2.NewCalculator()
 
 // NewXormEngine is an xorm.Engine factory, that can be stubbed by tests.
 //nolint:gocritic
@@ -62,24 +59,58 @@ var NewXormEngine = func(driverName string, connectionString string) (*xorm.Engi
 	return xorm.NewEngine(driverName, connectionString)
 }
 
-type dataPlugin struct {
+type JsonData struct {
+	MaxOpenConns        int    `json:"maxOpenConns"`
+	MaxIdleConns        int    `json:"maxIdleConns"`
+	ConnMaxLifetime     int    `json:"connMaxLifetime"`
+	Timescaledb         bool   `json:"timescaledb"`
+	Mode                string `json:"sslmode"`
+	ConfigurationMethod string `json:"tlsConfigurationMethod"`
+	RootCertFile        string `json:"sslRootCertFile"`
+	CertFile            string `json:"sslCertFile"`
+	CertKeyFile         string `json:"sslKeyFile"`
+	Timezone            string `json:"timezone"`
+	Encrypt             string `json:"encrypt"`
+	TimeInterval        string `json:"timeInterval"`
+}
+
+type DataSourceInfo struct {
+	JsonData                JsonData
+	URL                     string
+	User                    string
+	Database                string
+	ID                      int64
+	Updated                 time.Time
+	UID                     string
+	DecryptedSecureJSONData map[string]string
+}
+
+type DataPluginConfiguration struct {
+	DriverName        string
+	DSInfo            DataSourceInfo
+	ConnectionString  string
+	TimeColumnNames   []string
+	MetricColumnTypes []string
+}
+type DataSourceHandler struct {
 	macroEngine            SQLMacroEngine
 	queryResultTransformer SqlQueryResultTransformer
 	engine                 *xorm.Engine
 	timeColumnNames        []string
 	metricColumnTypes      []string
 	log                    log.Logger
+	dsInfo                 DataSourceInfo
+}
+type QueryJson struct {
+	RawSql       string  `json:"rawSql"`
+	Fill         bool    `json:"fill"`
+	FillInterval float64 `json:"fillInterval"`
+	FillMode     string  `json:"fillMode"`
+	FillValue    float64 `json:"fillValue"`
+	Format       string  `json:"format"`
 }
 
-type DataPluginConfiguration struct {
-	DriverName        string
-	Datasource        *models.DataSource
-	ConnectionString  string
-	TimeColumnNames   []string
-	MetricColumnTypes []string
-}
-
-func (e *dataPlugin) transformQueryError(err error) error {
+func (e *DataSourceHandler) transformQueryError(err error) error {
 	// OpError is the error type usually returned by functions in the net
 	// package. It describes the operation, network type, and address of
 	// an error. We log this error rather than return it to the client
@@ -93,32 +124,31 @@ func (e *dataPlugin) transformQueryError(err error) error {
 	return e.queryResultTransformer.TransformQueryError(err)
 }
 
-// NewDataPlugin returns a new plugins.DataPlugin
-//nolint: staticcheck // plugins.DataPlugin deprecated
-func NewDataPlugin(config DataPluginConfiguration, queryResultTransformer SqlQueryResultTransformer,
-	macroEngine SQLMacroEngine, log log.Logger) (plugins.DataPlugin, error) {
-	plugin := dataPlugin{
+func NewQueryDataHandler(config DataPluginConfiguration, queryResultTransformer SqlQueryResultTransformer,
+	macroEngine SQLMacroEngine, log log.Logger) (*DataSourceHandler, error) {
+	queryDataHandler := DataSourceHandler{
 		queryResultTransformer: queryResultTransformer,
 		macroEngine:            macroEngine,
 		timeColumnNames:        []string{"time"},
 		log:                    log,
+		dsInfo:                 config.DSInfo,
 	}
 
 	if len(config.TimeColumnNames) > 0 {
-		plugin.timeColumnNames = config.TimeColumnNames
+		queryDataHandler.timeColumnNames = config.TimeColumnNames
 	}
 
 	if len(config.MetricColumnTypes) > 0 {
-		plugin.metricColumnTypes = config.MetricColumnTypes
+		queryDataHandler.metricColumnTypes = config.MetricColumnTypes
 	}
 
 	engineCache.Lock()
 	defer engineCache.Unlock()
 
-	if engine, present := engineCache.cache[config.Datasource.Id]; present {
-		if version := engineCache.versions[config.Datasource.Id]; version == config.Datasource.Version {
-			plugin.engine = engine
-			return &plugin, nil
+	if engine, present := engineCache.cache[config.DSInfo.ID]; present {
+		if updateTime := engineCache.updates[config.DSInfo.ID]; updateTime.Before(config.DSInfo.Updated) {
+			queryDataHandler.engine = engine
+			return &queryDataHandler, nil
 		}
 	}
 
@@ -127,104 +157,104 @@ func NewDataPlugin(config DataPluginConfiguration, queryResultTransformer SqlQue
 		return nil, err
 	}
 
-	maxOpenConns := config.Datasource.JsonData.Get("maxOpenConns").MustInt(0)
-	engine.SetMaxOpenConns(maxOpenConns)
-	maxIdleConns := config.Datasource.JsonData.Get("maxIdleConns").MustInt(2)
-	engine.SetMaxIdleConns(maxIdleConns)
-	connMaxLifetime := config.Datasource.JsonData.Get("connMaxLifetime").MustInt(14400)
-	engine.SetConnMaxLifetime(time.Duration(connMaxLifetime) * time.Second)
+	engine.SetMaxOpenConns(config.DSInfo.JsonData.MaxOpenConns)
+	engine.SetMaxIdleConns(config.DSInfo.JsonData.MaxIdleConns)
+	engine.SetConnMaxLifetime(time.Duration(config.DSInfo.JsonData.ConnMaxLifetime) * time.Second)
 
-	engineCache.versions[config.Datasource.Id] = config.Datasource.Version
-	engineCache.cache[config.Datasource.Id] = engine
-	plugin.engine = engine
-
-	return &plugin, nil
+	engineCache.updates[config.DSInfo.ID] = config.DSInfo.Updated
+	engineCache.cache[config.DSInfo.ID] = engine
+	queryDataHandler.engine = engine
+	return &queryDataHandler, nil
 }
 
 const rowLimit = 1000000
 
-// DataQuery queries for data.
-//nolint: staticcheck // plugins.DataPlugin deprecated
-func (e *dataPlugin) DataQuery(ctx context.Context, dsInfo *models.DataSource,
-	queryContext plugins.DataQuery) (plugins.DataResponse, error) {
-	ch := make(chan plugins.DataQueryResult, len(queryContext.Queries))
+type DBDataResponse struct {
+	dataResponse backend.DataResponse
+	refID        string
+}
+
+func (e *DataSourceHandler) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	result := backend.NewQueryDataResponse()
+	ch := make(chan DBDataResponse, len(req.Queries))
 	var wg sync.WaitGroup
 	// Execute each query in a goroutine and wait for them to finish afterwards
-	for _, query := range queryContext.Queries {
-		if query.Model.Get("rawSql").MustString() == "" {
+	for _, query := range req.Queries {
+		queryjson := QueryJson{
+			Fill:   false,
+			Format: "time_series",
+		}
+		err := json.Unmarshal(query.JSON, &queryjson)
+		if err != nil {
+			return nil, fmt.Errorf("error unmarshal query json: %w", err)
+		}
+		if queryjson.RawSql == "" {
 			continue
 		}
 
 		wg.Add(1)
-		go e.executeQuery(query, &wg, queryContext, ch)
+		go e.executeQuery(query, &wg, ctx, ch, queryjson)
 	}
 
 	wg.Wait()
 
 	// Read results from channels
 	close(ch)
-	result := plugins.DataResponse{
-		Results: make(map[string]plugins.DataQueryResult),
-	}
+	result.Responses = make(map[string]backend.DataResponse)
 	for queryResult := range ch {
-		result.Results[queryResult.RefID] = queryResult
+		result.Responses[queryResult.refID] = queryResult.dataResponse
 	}
 
 	return result, nil
 }
 
-//nolint: staticcheck,gocyclo // plugins.DataQueryResult deprecated
-func (e *dataPlugin) executeQuery(query plugins.DataSubQuery, wg *sync.WaitGroup, queryContext plugins.DataQuery,
-	ch chan plugins.DataQueryResult) {
+func (e *DataSourceHandler) executeQuery(query backend.DataQuery, wg *sync.WaitGroup, queryContext context.Context,
+	ch chan DBDataResponse, queryJson QueryJson) {
 	defer wg.Done()
-
-	queryResult := plugins.DataQueryResult{
-		Meta:  simplejson.New(),
-		RefID: query.RefID,
+	queryResult := DBDataResponse{
+		dataResponse: backend.DataResponse{},
+		refID:        query.RefID,
 	}
 
 	defer func() {
 		if r := recover(); r != nil {
 			e.log.Error("executeQuery panic", "error", r, "stack", log.Stack(1))
 			if theErr, ok := r.(error); ok {
-				queryResult.Error = theErr
+				queryResult.dataResponse.Error = theErr
 			} else if theErrString, ok := r.(string); ok {
-				queryResult.Error = fmt.Errorf(theErrString)
+				queryResult.dataResponse.Error = fmt.Errorf(theErrString)
 			} else {
-				queryResult.Error = fmt.Errorf("unexpected error, see the server log for details")
+				queryResult.dataResponse.Error = fmt.Errorf("unexpected error, see the server log for details")
 			}
 			ch <- queryResult
 		}
 	}()
 
-	rawSQL := query.Model.Get("rawSql").MustString()
-	if rawSQL == "" {
+	if queryJson.RawSql == "" {
 		panic("Query model property rawSql should not be empty at this point")
 	}
-	var timeRange plugins.DataTimeRange
-	if queryContext.TimeRange != nil {
-		timeRange = *queryContext.TimeRange
-	}
+
+	timeRange := query.TimeRange
 
 	errAppendDebug := func(frameErr string, err error, query string) {
 		var emptyFrame data.Frame
 		emptyFrame.SetMeta(&data.FrameMeta{
 			ExecutedQueryString: query,
 		})
-		queryResult.Error = fmt.Errorf("%s: %w", frameErr, err)
-		queryResult.Dataframes = plugins.NewDecodedDataFrames(data.Frames{&emptyFrame})
+		queryResult.dataResponse.Error = fmt.Errorf("%s: %w", frameErr, err)
+		queryResult.dataResponse.Frames = data.Frames{&emptyFrame}
 		ch <- queryResult
 	}
 
 	// global substitutions
-	interpolatedQuery, err := Interpolate(query, timeRange, rawSQL)
+	interpolatedQuery, err := Interpolate(query, timeRange, e.dsInfo.JsonData.TimeInterval, queryJson.RawSql)
 	if err != nil {
 		errAppendDebug("interpolation failed", e.transformQueryError(err), interpolatedQuery)
 		return
 	}
 
 	// data source specific substitutions
-	interpolatedQuery, err = e.macroEngine.Interpolate(query, timeRange, interpolatedQuery)
+	interpolatedQuery, err = e.macroEngine.Interpolate(&query, timeRange, interpolatedQuery)
 	if err != nil {
 		errAppendDebug("interpolation failed", e.transformQueryError(err), interpolatedQuery)
 		return
@@ -265,7 +295,7 @@ func (e *dataPlugin) executeQuery(query plugins.DataSubQuery, wg *sync.WaitGroup
 
 	// If no rows were returned, no point checking anything else.
 	if frame.Rows() == 0 {
-		queryResult.Dataframes = plugins.NewDecodedDataFrames(data.Frames{frame})
+		queryResult.dataResponse.Frames = data.Frames{frame}
 		ch <- queryResult
 		return
 	}
@@ -338,13 +368,13 @@ func (e *dataPlugin) executeQuery(query plugins.DataSubQuery, wg *sync.WaitGroup
 		}
 	}
 
-	queryResult.Dataframes = plugins.NewDecodedDataFrames(data.Frames{frame})
+	queryResult.dataResponse.Frames = data.Frames{frame}
 	ch <- queryResult
 }
 
 // Interpolate provides global macros/substitutions for all sql datasources.
-var Interpolate = func(query plugins.DataSubQuery, timeRange plugins.DataTimeRange, sql string) (string, error) {
-	minInterval, err := interval.GetIntervalFrom(query.DataSource, query.Model, time.Second*60)
+var Interpolate = func(query backend.DataQuery, timeRange backend.TimeRange, timeInterval string, sql string) (string, error) {
+	minInterval, err := intervalv2.GetIntervalFrom(timeInterval, query.Interval.String(), query.Interval.Milliseconds(), time.Second*60)
 	if err != nil {
 		return "", err
 	}
@@ -355,14 +385,14 @@ var Interpolate = func(query plugins.DataSubQuery, timeRange plugins.DataTimeRan
 
 	sql = strings.ReplaceAll(sql, "$__interval_ms", strconv.FormatInt(interval.Milliseconds(), 10))
 	sql = strings.ReplaceAll(sql, "$__interval", interval.Text)
-	sql = strings.ReplaceAll(sql, "$__unixEpochFrom()", fmt.Sprintf("%d", timeRange.GetFromAsSecondsEpoch()))
-	sql = strings.ReplaceAll(sql, "$__unixEpochTo()", fmt.Sprintf("%d", timeRange.GetToAsSecondsEpoch()))
+	sql = strings.ReplaceAll(sql, "$__unixEpochFrom()", fmt.Sprintf("%d", timeRange.From.UTC().Unix()))
+	sql = strings.ReplaceAll(sql, "$__unixEpochTo()", fmt.Sprintf("%d", timeRange.To.UTC().Unix()))
 
 	return sql, nil
 }
 
 //nolint: staticcheck // plugins.DataPlugin deprecated
-func (e *dataPlugin) newProcessCfg(query plugins.DataSubQuery, queryContext plugins.DataQuery,
+func (e *DataSourceHandler) newProcessCfg(query backend.DataQuery, queryContext context.Context,
 	rows *core.Rows, interpolatedQuery string) (*dataQueryModel, error) {
 	columnNames, err := rows.Columns()
 	if err != nil {
@@ -383,35 +413,38 @@ func (e *dataPlugin) newProcessCfg(query plugins.DataSubQuery, queryContext plug
 		queryContext: queryContext,
 	}
 
-	if query.Model.Get("fill").MustBool(false) {
+	queryJson := QueryJson{}
+	err = json.Unmarshal(query.JSON, &queryJson)
+	if err != nil {
+		return nil, err
+	}
+
+	if queryJson.Fill {
 		qm.FillMissing = &data.FillMissing{}
-		qm.Interval = time.Duration(query.Model.Get("fillInterval").MustFloat64() * float64(time.Second))
-		switch strings.ToLower(query.Model.Get("fillMode").MustString()) {
+		qm.Interval = time.Duration(queryJson.FillInterval * float64(time.Second))
+		switch strings.ToLower(queryJson.FillMode) {
 		case "null":
 			qm.FillMissing.Mode = data.FillModeNull
 		case "previous":
 			qm.FillMissing.Mode = data.FillModePrevious
 		case "value":
 			qm.FillMissing.Mode = data.FillModeValue
-			qm.FillMissing.Value = query.Model.Get("fillValue").MustFloat64()
+			qm.FillMissing.Value = queryJson.FillValue
 		default:
 		}
 	}
 	//nolint: staticcheck // plugins.DataPlugin deprecated
 
-	if queryContext.TimeRange != nil {
-		qm.TimeRange.From = queryContext.TimeRange.GetFromAsTimeUTC()
-		qm.TimeRange.To = queryContext.TimeRange.GetToAsTimeUTC()
-	}
+	qm.TimeRange.From = query.TimeRange.From.UTC()
+	qm.TimeRange.To = query.TimeRange.To.UTC()
 
-	format := query.Model.Get("format").MustString("time_series")
-	switch format {
+	switch queryJson.Format {
 	case "time_series":
 		qm.Format = dataQueryFormatSeries
 	case "table":
 		qm.Format = dataQueryFormatTable
 	default:
-		panic(fmt.Sprintf("Unrecognized query model format: %q", format))
+		panic(fmt.Sprintf("Unrecognized query model format: %q", queryJson.Format))
 	}
 
 	for i, col := range qm.columnNames {
@@ -462,7 +495,7 @@ type dataQueryModel struct {
 	metricIndex       int
 	rows              *core.Rows
 	metricPrefix      bool
-	queryContext      plugins.DataQuery
+	queryContext      context.Context
 }
 
 func convertInt64ToFloat64(origin *data.Field, newField *data.Field) {
@@ -902,23 +935,36 @@ func convertSQLValueColumnToFloat(frame *data.Frame, Index int) (*data.Frame, er
 	return frame, nil
 }
 
-func SetupFillmode(query plugins.DataSubQuery, interval time.Duration, fillmode string) error {
-	query.Model.Set("fill", true)
-	query.Model.Set("fillInterval", interval.Seconds())
+func SetupFillmode(query *backend.DataQuery, interval time.Duration, fillmode string) error {
+	rawQueryProp := make(map[string]interface{})
+	queryBytes, err := query.JSON.MarshalJSON()
+	if err != nil {
+		return err
+	}
+	err = json.Unmarshal(queryBytes, &rawQueryProp)
+	if err != nil {
+		return err
+	}
+	rawQueryProp["fill"] = true
+	rawQueryProp["fillInterval"] = interval.Seconds()
+
 	switch fillmode {
 	case "NULL":
-		query.Model.Set("fillMode", "null")
+		rawQueryProp["fillMode"] = "null"
 	case "previous":
-		query.Model.Set("fillMode", "previous")
+		rawQueryProp["fillMode"] = "previous"
 	default:
-		query.Model.Set("fillMode", "value")
+		rawQueryProp["fillMode"] = "value"
 		floatVal, err := strconv.ParseFloat(fillmode, 64)
 		if err != nil {
 			return fmt.Errorf("error parsing fill value %v", fillmode)
 		}
-		query.Model.Set("fillValue", floatVal)
+		rawQueryProp["fillValue"] = floatVal
+	}
+	query.JSON, err = json.Marshal(rawQueryProp)
+	if err != nil {
+		return err
 	}
-
 	return nil
 }
 

pkg/tsdb/sqleng/sql_engine_test.go

@@ -7,11 +7,10 @@ import (
 	"testing"
 	"time"
 
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
 	"github.com/grafana/grafana-plugin-sdk-go/data/sqlutil"
-	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/infra/log"
-	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -25,35 +24,35 @@ func TestSQLEngine(t *testing.T) {
 	t.Run("Given a time range between 2018-04-12 00:00 and 2018-04-12 00:05", func(t *testing.T) {
 		from := time.Date(2018, 4, 12, 18, 0, 0, 0, time.UTC)
 		to := from.Add(5 * time.Minute)
-		timeRange := plugins.DataTimeRange{From: "5m", To: "now", Now: to}
-		query := plugins.DataSubQuery{DataSource: &models.DataSource{}, Model: simplejson.New()}
+		timeRange := backend.TimeRange{From: from, To: to}
+		query := backend.DataQuery{JSON: []byte("{}")}
 
 		t.Run("interpolate $__interval", func(t *testing.T) {
-			sql, err := Interpolate(query, timeRange, "select $__interval ")
+			sql, err := Interpolate(query, timeRange, "", "select $__interval ")
 			require.NoError(t, err)
 			require.Equal(t, "select 1m ", sql)
 		})
 
 		t.Run("interpolate $__interval in $__timeGroup", func(t *testing.T) {
-			sql, err := Interpolate(query, timeRange, "select $__timeGroupAlias(time,$__interval)")
+			sql, err := Interpolate(query, timeRange, "", "select $__timeGroupAlias(time,$__interval)")
 			require.NoError(t, err)
 			require.Equal(t, "select $__timeGroupAlias(time,1m)", sql)
 		})
 
 		t.Run("interpolate $__interval_ms", func(t *testing.T) {
-			sql, err := Interpolate(query, timeRange, "select $__interval_ms ")
+			sql, err := Interpolate(query, timeRange, "", "select $__interval_ms ")
 			require.NoError(t, err)
 			require.Equal(t, "select 60000 ", sql)
 		})
 
 		t.Run("interpolate __unixEpochFrom function", func(t *testing.T) {
-			sql, err := Interpolate(query, timeRange, "select $__unixEpochFrom()")
+			sql, err := Interpolate(query, timeRange, "", "select $__unixEpochFrom()")
 			require.NoError(t, err)
 			require.Equal(t, fmt.Sprintf("select %d", from.Unix()), sql)
 		})
 
 		t.Run("interpolate __unixEpochTo function", func(t *testing.T) {
-			sql, err := Interpolate(query, timeRange, "select $__unixEpochTo()")
+			sql, err := Interpolate(query, timeRange, "", "select $__unixEpochTo()")
 			require.NoError(t, err)
 			require.Equal(t, fmt.Sprintf("select %d", to.Unix()), sql)
 		})
@@ -406,7 +405,7 @@ func TestSQLEngine(t *testing.T) {
 
 		for _, tc := range tests {
 			transformer := &testQueryResultTransformer{}
-			dp := dataPlugin{
+			dp := DataSourceHandler{
 				log:                    log.New("test"),
 				queryResultTransformer: transformer,
 			}