Chore: Refactor backend plugin manager/tsdb query data (#34944)

Move QueryData method into backend plugin manager which HandleRequest uses to 
query data from plugin SDK supported data sources. This allowed us to remove a lot 
of code no longer needed.

Ref #21510

Co-authored-by: Will Browne <wbrowne@users.noreply.github.com>

pkg/expr/service_test.go

@@ -113,7 +113,3 @@ func (me *mockEndpoint) DataQuery(ctx context.Context, ds *models.DataSource, qu
 type fakeBackendPM struct {
 	backendplugin.Manager
 }
-
-func (pm fakeBackendPM) GetDataPlugin(string) interface{} {
-	return nil
-}

pkg/plugins/app_plugin.go

@@ -71,7 +71,7 @@ func (app *AppPlugin) Load(decoder *json.Decoder, base *PluginBase, backendPlugi
 	if app.Backend {
 		cmd := ComposePluginStartCommand(app.Executable)
 		fullpath := filepath.Join(base.PluginDir, cmd)
-		factory := grpcplugin.NewBackendPlugin(app.Id, fullpath, grpcplugin.PluginStartFuncs{})
+		factory := grpcplugin.NewBackendPlugin(app.Id, fullpath)
 		if err := backendPluginManager.RegisterAndStart(context.Background(), app.Id, factory); err != nil {
 			return nil, errutil.Wrapf(err, "failed to register backend plugin")
 		}

pkg/plugins/backendplugin/coreplugin/core_plugin.go

@@ -5,10 +5,7 @@ import (
 
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana/pkg/infra/log"
-	"github.com/grafana/grafana/pkg/models"
-	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/plugins/backendplugin"
-	"github.com/grafana/grafana/pkg/plugins/backendplugin/instrumentation"
 )
 
 // corePlugin represents a plugin that's part of Grafana core.
@@ -43,15 +40,6 @@ func (cp *corePlugin) Logger() log.Logger {
 	return cp.logger
 }
 
-//nolint: staticcheck // plugins.DataResponse deprecated
-func (cp *corePlugin) DataQuery(ctx context.Context, dsInfo *models.DataSource,
-	tsdbQuery plugins.DataQuery) (plugins.DataResponse, error) {
-	// TODO: Inline the adapter, since it shouldn't be necessary
-	adapter := newQueryEndpointAdapter(cp.pluginID, cp.logger, instrumentation.InstrumentQueryDataHandler(
-		cp.QueryDataHandler))
-	return adapter.DataQuery(ctx, dsInfo, tsdbQuery)
-}
-
 func (cp *corePlugin) Start(ctx context.Context) error {
 	return nil
 }
@@ -88,6 +76,14 @@ func (cp *corePlugin) CheckHealth(ctx context.Context, req *backend.CheckHealthR
 	return nil, backendplugin.ErrMethodNotImplemented
 }
 
+func (cp *corePlugin) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	if cp.QueryDataHandler != nil {
+		return cp.QueryDataHandler.QueryData(ctx, req)
+	}
+
+	return nil, backendplugin.ErrMethodNotImplemented
+}
+
 func (cp *corePlugin) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
 	if cp.CallResourceHandler != nil {
 		return cp.CallResourceHandler.CallResource(ctx, req, sender)

pkg/plugins/backendplugin/coreplugin/query_endpoint_adapter.go

@@ -1,116 +0,0 @@
-package coreplugin
-
-import (
-	"context"
-	"time"
-
-	"github.com/grafana/grafana-plugin-sdk-go/backend"
-	"github.com/grafana/grafana/pkg/infra/log"
-	"github.com/grafana/grafana/pkg/models"
-	"github.com/grafana/grafana/pkg/plugins"
-	"github.com/grafana/grafana/pkg/plugins/adapters"
-)
-
-// nolint:staticcheck // plugins.DataPlugin deprecated
-func newQueryEndpointAdapter(pluginID string, logger log.Logger, handler backend.QueryDataHandler) plugins.DataPlugin {
-	return &queryEndpointAdapter{
-		pluginID: pluginID,
-		logger:   logger,
-		handler:  handler,
-	}
-}
-
-type queryEndpointAdapter struct {
-	pluginID string
-	logger   log.Logger
-	handler  backend.QueryDataHandler
-}
-
-func modelToInstanceSettings(ds *models.DataSource) (*backend.DataSourceInstanceSettings, error) {
-	jsonDataBytes, err := ds.JsonData.MarshalJSON()
-	if err != nil {
-		return nil, err
-	}
-
-	return &backend.DataSourceInstanceSettings{
-		ID:                      ds.Id,
-		Name:                    ds.Name,
-		URL:                     ds.Url,
-		Database:                ds.Database,
-		User:                    ds.User,
-		BasicAuthEnabled:        ds.BasicAuth,
-		BasicAuthUser:           ds.BasicAuthUser,
-		JSONData:                jsonDataBytes,
-		DecryptedSecureJSONData: ds.DecryptedValues(),
-		Updated:                 ds.Updated,
-	}, nil
-}
-
-// nolint:staticcheck // plugins.DataQuery deprecated
-func (a *queryEndpointAdapter) DataQuery(ctx context.Context, ds *models.DataSource, query plugins.DataQuery) (
-	plugins.DataResponse, error) {
-	instanceSettings, err := modelToInstanceSettings(ds)
-	if err != nil {
-		return plugins.DataResponse{}, err
-	}
-
-	req := &backend.QueryDataRequest{
-		PluginContext: backend.PluginContext{
-			OrgID:                      ds.OrgId,
-			PluginID:                   a.pluginID,
-			User:                       adapters.BackendUserFromSignedInUser(query.User),
-			DataSourceInstanceSettings: instanceSettings,
-		},
-		Queries: []backend.DataQuery{},
-		Headers: query.Headers,
-	}
-
-	for _, q := range query.Queries {
-		modelJSON, err := q.Model.MarshalJSON()
-		if err != nil {
-			return plugins.DataResponse{}, err
-		}
-		req.Queries = append(req.Queries, backend.DataQuery{
-			RefID:         q.RefID,
-			Interval:      time.Duration(q.IntervalMS) * time.Millisecond,
-			MaxDataPoints: q.MaxDataPoints,
-			TimeRange: backend.TimeRange{
-				From: query.TimeRange.GetFromAsTimeUTC(),
-				To:   query.TimeRange.GetToAsTimeUTC(),
-			},
-			QueryType: q.QueryType,
-			JSON:      modelJSON,
-		})
-	}
-
-	resp, err := a.handler.QueryData(ctx, req)
-	if err != nil {
-		return plugins.DataResponse{}, err
-	}
-
-	tR := plugins.DataResponse{
-		Results: make(map[string]plugins.DataQueryResult, len(resp.Responses)),
-	}
-
-	for refID, r := range resp.Responses {
-		qr := plugins.DataQueryResult{
-			RefID: refID,
-		}
-
-		for _, f := range r.Frames {
-			if f.RefID == "" {
-				f.RefID = refID
-			}
-		}
-
-		qr.Dataframes = plugins.NewDecodedDataFrames(r.Frames)
-
-		if r.Error != nil {
-			qr.Error = r.Error
-		}
-
-		tR.Results[refID] = qr
-	}
-
-	return tR, nil
-}

pkg/plugins/backendplugin/grpcplugin/client.go

@@ -38,13 +38,8 @@ func newClientConfig(executablePath string, env []string, logger log.Logger,
 	}
 }
 
-// StartFunc callback function called when a plugin with current plugin protocol version is started.
-type StartFunc func(pluginID string, client *Client, logger log.Logger) error
-
-// PluginStartFuncs functions called for plugin when started.
-type PluginStartFuncs struct {
-	OnStart StartFunc
-}
+// StartRendererFunc callback function called when a renderer plugin is started.
+type StartRendererFunc func(pluginID string, renderer pluginextensionv2.RendererPlugin, logger log.Logger) error
 
 // PluginDescriptor is a descriptor used for registering backend plugins.
 type PluginDescriptor struct {
@@ -52,7 +47,7 @@ type PluginDescriptor struct {
 	executablePath   string
 	managed          bool
 	versionedPlugins map[int]goplugin.PluginSet
-	startFns         PluginStartFuncs
+	startRendererFn  StartRendererFunc
 }
 
 // getV2PluginSet returns list of plugins supported on v2.
@@ -67,7 +62,7 @@ func getV2PluginSet() goplugin.PluginSet {
 }
 
 // NewBackendPlugin creates a new backend plugin factory used for registering a backend plugin.
-func NewBackendPlugin(pluginID, executablePath string, startFns PluginStartFuncs) backendplugin.PluginFactoryFunc {
+func NewBackendPlugin(pluginID, executablePath string) backendplugin.PluginFactoryFunc {
 	return newPlugin(PluginDescriptor{
 		pluginID:       pluginID,
 		executablePath: executablePath,
@@ -75,12 +70,11 @@ func NewBackendPlugin(pluginID, executablePath string, startFns PluginStartFuncs
 		versionedPlugins: map[int]goplugin.PluginSet{
 			grpcplugin.ProtocolVersion: getV2PluginSet(),
 		},
-		startFns: startFns,
 	})
 }
 
 // NewRendererPlugin creates a new renderer plugin factory used for registering a backend renderer plugin.
-func NewRendererPlugin(pluginID, executablePath string, startFns PluginStartFuncs) backendplugin.PluginFactoryFunc {
+func NewRendererPlugin(pluginID, executablePath string, startFn StartRendererFunc) backendplugin.PluginFactoryFunc {
 	return newPlugin(PluginDescriptor{
 		pluginID:       pluginID,
 		executablePath: executablePath,
@@ -88,13 +82,6 @@ func NewRendererPlugin(pluginID, executablePath string, startFns PluginStartFunc
 		versionedPlugins: map[int]goplugin.PluginSet{
 			grpcplugin.ProtocolVersion: getV2PluginSet(),
 		},
-		startFns: startFns,
+		startRendererFn: startFn,
 	})
 }
-
-// Client client for communicating with a plugin using the current (v2) plugin protocol.
-type Client struct {
-	DataPlugin     grpcplugin.DataClient
-	RendererPlugin pluginextensionv2.RendererPlugin
-	StreamClient   grpcplugin.StreamClient
-}

pkg/plugins/backendplugin/grpcplugin/client_v2.go

@@ -11,11 +11,9 @@ import (
 	"github.com/grafana/grafana-plugin-sdk-go/genproto/pluginv2"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/plugins/backendplugin"
-	"github.com/grafana/grafana/pkg/plugins/backendplugin/instrumentation"
 	"github.com/grafana/grafana/pkg/plugins/backendplugin/pluginextensionv2"
 	"github.com/grafana/grafana/pkg/util/errutil"
 	"github.com/hashicorp/go-plugin"
-	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 )
@@ -69,7 +67,7 @@ func newClientV2(descriptor PluginDescriptor, logger log.Logger, rpcClient plugi
 
 	if rawData != nil {
 		if dataClient, ok := rawData.(grpcplugin.DataClient); ok {
-			c.DataClient = instrumentDataClient(dataClient)
+			c.DataClient = dataClient
 		}
 	}
 
@@ -85,13 +83,8 @@ func newClientV2(descriptor PluginDescriptor, logger log.Logger, rpcClient plugi
 		}
 	}
 
-	if descriptor.startFns.OnStart != nil {
-		client := &Client{
-			DataPlugin:     c.DataClient,
-			RendererPlugin: c.RendererPlugin,
-			StreamClient:   c.StreamClient,
-		}
-		if err := descriptor.startFns.OnStart(descriptor.pluginID, client, logger); err != nil {
+	if descriptor.startRendererFn != nil {
+		if err := descriptor.startRendererFn(descriptor.pluginID, c.RendererPlugin, logger); err != nil {
 			return nil, err
 		}
 	}
@@ -137,6 +130,25 @@ func (c *clientV2) CheckHealth(ctx context.Context, req *backend.CheckHealthRequ
 	return backend.FromProto().CheckHealthResponse(protoResp), nil
 }
 
+func (c *clientV2) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	if c.DataClient == nil {
+		return nil, backendplugin.ErrMethodNotImplemented
+	}
+
+	protoReq := backend.ToProto().QueryDataRequest(req)
+	protoResp, err := c.DataClient.QueryData(ctx, protoReq)
+
+	if err != nil {
+		if status.Code(err) == codes.Unimplemented {
+			return nil, backendplugin.ErrMethodNotImplemented
+		}
+
+		return nil, errutil.Wrap("Failed to query data", err)
+	}
+
+	return backend.FromProto().QueryDataResponse(protoResp)
+}
+
 func (c *clientV2) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
 	if c.ResourceClient == nil {
 		return backendplugin.ErrMethodNotImplemented
@@ -226,24 +238,3 @@ func (c *clientV2) RunStream(ctx context.Context, req *backend.RunStreamRequest,
 		}
 	}
 }
-
-type dataClientQueryDataFunc func(ctx context.Context, req *pluginv2.QueryDataRequest, opts ...grpc.CallOption) (*pluginv2.QueryDataResponse, error)
-
-func (fn dataClientQueryDataFunc) QueryData(ctx context.Context, req *pluginv2.QueryDataRequest, opts ...grpc.CallOption) (*pluginv2.QueryDataResponse, error) {
-	return fn(ctx, req, opts...)
-}
-
-func instrumentDataClient(plugin grpcplugin.DataClient) grpcplugin.DataClient {
-	if plugin == nil {
-		return nil
-	}
-
-	return dataClientQueryDataFunc(func(ctx context.Context, req *pluginv2.QueryDataRequest, opts ...grpc.CallOption) (*pluginv2.QueryDataResponse, error) {
-		var resp *pluginv2.QueryDataResponse
-		err := instrumentation.InstrumentQueryDataRequest(req.PluginContext.PluginId, func() (innerErr error) {
-			resp, innerErr = plugin.QueryData(ctx, req)
-			return
-		})
-		return resp, err
-	})
-}

pkg/plugins/backendplugin/grpcplugin/grpc_plugin.go

@@ -14,6 +14,7 @@ import (
 type pluginClient interface {
 	backend.CollectMetricsHandler
 	backend.CheckHealthHandler
+	backend.QueryDataHandler
 	backend.CallResourceHandler
 	backend.StreamHandler
 }
@@ -137,6 +138,15 @@ func (p *grpcPlugin) CheckHealth(ctx context.Context, req *backend.CheckHealthRe
 	return pluginClient.CheckHealth(ctx, req)
 }
 
+func (p *grpcPlugin) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	pluginClient, ok := p.getPluginClient()
+	if !ok {
+		return nil, backendplugin.ErrPluginUnavailable
+	}
+
+	return pluginClient.QueryData(ctx, req)
+}
+
 func (p *grpcPlugin) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
 	pluginClient, ok := p.getPluginClient()
 	if !ok {

pkg/plugins/backendplugin/ifaces.go

@@ -24,13 +24,12 @@ type Manager interface {
 	CollectMetrics(ctx context.Context, pluginID string) (*backend.CollectMetricsResult, error)
 	// CheckHealth checks the health of a registered backend plugin.
 	CheckHealth(ctx context.Context, pCtx backend.PluginContext) (*backend.CheckHealthResult, error)
+	// QueryData query data from a registered backend plugin.
+	QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error)
 	// CallResource calls a plugin resource.
 	CallResource(pluginConfig backend.PluginContext, ctx *models.ReqContext, path string)
 	// Get plugin by its ID.
 	Get(pluginID string) (Plugin, bool)
-	// GetDataPlugin gets a DataPlugin with a certain ID or nil if it doesn't exist.
-	// TODO: interface{} is the return type in order to break a dependency cycle. Should be plugins.DataPlugin.
-	GetDataPlugin(pluginID string) interface{}
 }
 
 // Plugin is the backend plugin interface.
@@ -45,6 +44,7 @@ type Plugin interface {
 	IsDecommissioned() bool
 	backend.CollectMetricsHandler
 	backend.CheckHealthHandler
+	backend.QueryDataHandler
 	backend.CallResourceHandler
 	backend.StreamHandler
 }

pkg/plugins/backendplugin/manager/manager.go

@@ -17,7 +17,6 @@ import (
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/models"
-	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/plugins/backendplugin"
 	"github.com/grafana/grafana/pkg/plugins/backendplugin/instrumentation"
 	"github.com/grafana/grafana/pkg/registry"
@@ -151,6 +150,10 @@ func (m *manager) Get(pluginID string) (backendplugin.Plugin, bool) {
 	p, ok := m.plugins[pluginID]
 	m.pluginsMu.RUnlock()
 
+	if ok && p.IsDecommissioned() {
+		return nil, false
+	}
+
 	return p, ok
 }
 
@@ -181,21 +184,6 @@ func (m *manager) getAzureEnvironmentVariables() []string {
 	return variables
 }
 
-//nolint: staticcheck // plugins.DataPlugin deprecated
-func (m *manager) GetDataPlugin(pluginID string) interface{} {
-	p, _ := m.Get(pluginID)
-
-	if p == nil {
-		return nil
-	}
-
-	if dataPlugin, ok := p.(plugins.DataPlugin); ok {
-		return dataPlugin
-	}
-
-	return nil
-}
-
 // start starts a managed backend plugin
 func (m *manager) start(ctx context.Context, p backendplugin.Plugin) {
 	if !p.IsManaged() {
@@ -244,10 +232,7 @@ func (m *manager) stop(ctx context.Context) {
 
 // CollectMetrics collects metrics from a registered backend plugin.
 func (m *manager) CollectMetrics(ctx context.Context, pluginID string) (*backend.CollectMetricsResult, error) {
-	m.pluginsMu.RLock()
-	p, registered := m.plugins[pluginID]
-	m.pluginsMu.RUnlock()
-
+	p, registered := m.Get(pluginID)
 	if !registered {
 		return nil, backendplugin.ErrPluginNotRegistered
 	}
@@ -279,10 +264,7 @@ func (m *manager) CheckHealth(ctx context.Context, pluginContext backend.PluginC
 		}, nil
 	}
 
-	m.pluginsMu.RLock()
-	p, registered := m.plugins[pluginContext.PluginID]
-	m.pluginsMu.RUnlock()
-
+	p, registered := m.Get(pluginContext.PluginID)
 	if !registered {
 		return nil, backendplugin.ErrPluginNotRegistered
 	}
@@ -308,15 +290,39 @@ func (m *manager) CheckHealth(ctx context.Context, pluginContext backend.PluginC
 	return resp, nil
 }
 
+func (m *manager) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	p, registered := m.Get(req.PluginContext.PluginID)
+	if !registered {
+		return nil, backendplugin.ErrPluginNotRegistered
+	}
+
+	var resp *backend.QueryDataResponse
+	err := instrumentation.InstrumentQueryDataRequest(p.PluginID(), func() (innerErr error) {
+		resp, innerErr = p.QueryData(ctx, req)
+		return
+	})
+
+	if err != nil {
+		if errors.Is(err, backendplugin.ErrMethodNotImplemented) {
+			return nil, err
+		}
+
+		if errors.Is(err, backendplugin.ErrPluginUnavailable) {
+			return nil, err
+		}
+
+		return nil, errutil.Wrap("failed to query data", err)
+	}
+
+	return resp, nil
+}
+
 type keepCookiesJSONModel struct {
 	KeepCookies []string `json:"keepCookies"`
 }
 
 func (m *manager) callResourceInternal(w http.ResponseWriter, req *http.Request, pCtx backend.PluginContext) error {
-	m.pluginsMu.RLock()
-	p, registered := m.plugins[pCtx.PluginID]
-	m.pluginsMu.RUnlock()
-
+	p, registered := m.Get(pCtx.PluginID)
 	if !registered {
 		return backendplugin.ErrPluginNotRegistered
 	}

pkg/plugins/backendplugin/manager/manager_test.go

@@ -367,6 +367,7 @@ type testPlugin struct {
 	decommissioned bool
 	backend.CollectMetricsHandlerFunc
 	backend.CheckHealthHandlerFunc
+	backend.QueryDataHandlerFunc
 	backend.CallResourceHandlerFunc
 	mutex sync.RWMutex
 }
@@ -441,6 +442,14 @@ func (tp *testPlugin) CheckHealth(ctx context.Context, req *backend.CheckHealthR
 	return nil, backendplugin.ErrMethodNotImplemented
 }
 
+func (tp *testPlugin) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	if tp.QueryDataHandlerFunc != nil {
+		return tp.QueryDataHandlerFunc(ctx, req)
+	}
+
+	return nil, backendplugin.ErrMethodNotImplemented
+}
+
 func (tp *testPlugin) CallResource(ctx context.Context, req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
 	if tp.CallResourceHandlerFunc != nil {
 		return tp.CallResourceHandlerFunc(ctx, req, sender)

pkg/plugins/datasource_plugin.go

@@ -3,11 +3,8 @@ package plugins
 import (
 	"context"
 	"encoding/json"
-	"fmt"
 	"path/filepath"
 
-	"github.com/grafana/grafana/pkg/infra/log"
-	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/plugins/backendplugin"
 	"github.com/grafana/grafana/pkg/plugins/backendplugin/grpcplugin"
 	"github.com/grafana/grafana/pkg/util/errutil"
@@ -32,9 +29,6 @@ type DataSourcePlugin struct {
 	Backend    bool   `json:"backend,omitempty"`
 	Executable string `json:"executable,omitempty"`
 	SDK        bool   `json:"sdk,omitempty"`
-
-	client *grpcplugin.Client
-	logger log.Logger
 }
 
 func (p *DataSourcePlugin) Load(decoder *json.Decoder, base *PluginBase, backendPluginManager backendplugin.Manager) (
@@ -46,9 +40,7 @@ func (p *DataSourcePlugin) Load(decoder *json.Decoder, base *PluginBase, backend
 	if p.Backend {
 		cmd := ComposePluginStartCommand(p.Executable)
 		fullpath := filepath.Join(base.PluginDir, cmd)
-		factory := grpcplugin.NewBackendPlugin(p.Id, fullpath, grpcplugin.PluginStartFuncs{
-			OnStart: p.onPluginStart,
-		})
+		factory := grpcplugin.NewBackendPlugin(p.Id, fullpath)
 		if err := backendPluginManager.RegisterAndStart(context.Background(), p.Id, factory); err != nil {
 			return nil, errutil.Wrapf(err, "failed to register backend plugin")
 		}
@@ -56,27 +48,3 @@ func (p *DataSourcePlugin) Load(decoder *json.Decoder, base *PluginBase, backend
 
 	return p, nil
 }
-
-func (p *DataSourcePlugin) DataQuery(ctx context.Context, dsInfo *models.DataSource, query DataQuery) (DataResponse, error) {
-	if !p.CanHandleDataQueries() {
-		return DataResponse{}, fmt.Errorf("plugin %q can't handle data queries", p.Id)
-	}
-
-	endpoint := newDataSourcePluginWrapperV2(p.logger, p.Id, p.Type, p.client.DataPlugin)
-	return endpoint.Query(ctx, dsInfo, query)
-}
-
-func (p *DataSourcePlugin) CanHandleDataQueries() bool {
-	return p.client != nil
-}
-
-func (p *DataSourcePlugin) onPluginStart(pluginID string, client *grpcplugin.Client, logger log.Logger) error {
-	if client.DataPlugin == nil {
-		return nil
-	}
-
-	p.client = client
-	p.logger = logger
-
-	return nil
-}

pkg/plugins/datasource_plugin_wrapper_v2.go

@@ -1,99 +0,0 @@
-package plugins
-
-import (
-	"context"
-	"fmt"
-
-	"github.com/grafana/grafana-plugin-sdk-go/backend"
-	"github.com/grafana/grafana-plugin-sdk-go/backend/grpcplugin"
-	"github.com/grafana/grafana-plugin-sdk-go/genproto/pluginv2"
-	"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/adapters"
-	"github.com/grafana/grafana/pkg/services/oauthtoken"
-)
-
-func newDataSourcePluginWrapperV2(log log.Logger, pluginId, pluginType string, dataClient grpcplugin.DataClient) *DatasourcePluginWrapperV2 {
-	return &DatasourcePluginWrapperV2{DataClient: dataClient, logger: log, pluginId: pluginId, pluginType: pluginType}
-}
-
-type DatasourcePluginWrapperV2 struct {
-	grpcplugin.DataClient
-	logger     log.Logger
-	pluginId   string
-	pluginType string
-}
-
-func (tw *DatasourcePluginWrapperV2) Query(ctx context.Context, ds *models.DataSource, query DataQuery) (DataResponse, error) {
-	instanceSettings, err := adapters.ModelToInstanceSettings(ds)
-	if err != nil {
-		return DataResponse{}, err
-	}
-
-	if query.Headers == nil {
-		query.Headers = make(map[string]string)
-	}
-
-	if oauthtoken.IsOAuthPassThruEnabled(ds) {
-		if token := oauthtoken.GetCurrentOAuthToken(ctx, query.User); token != nil {
-			delete(query.Headers, "Authorization")
-			query.Headers["Authorization"] = fmt.Sprintf("%s %s", token.Type(), token.AccessToken)
-		}
-	}
-
-	pbQuery := &pluginv2.QueryDataRequest{
-		PluginContext: &pluginv2.PluginContext{
-			OrgId:                      ds.OrgId,
-			PluginId:                   tw.pluginId,
-			User:                       backend.ToProto().User(adapters.BackendUserFromSignedInUser(query.User)),
-			DataSourceInstanceSettings: backend.ToProto().DataSourceInstanceSettings(instanceSettings),
-		},
-		Queries: []*pluginv2.DataQuery{},
-		Headers: query.Headers,
-	}
-
-	for _, q := range query.Queries {
-		modelJSON, err := q.Model.MarshalJSON()
-		if err != nil {
-			return DataResponse{}, err
-		}
-		pbQuery.Queries = append(pbQuery.Queries, &pluginv2.DataQuery{
-			Json:          modelJSON,
-			IntervalMS:    q.IntervalMS,
-			RefId:         q.RefID,
-			MaxDataPoints: q.MaxDataPoints,
-			TimeRange: &pluginv2.TimeRange{
-				ToEpochMS:   query.TimeRange.GetToAsMsEpoch(),
-				FromEpochMS: query.TimeRange.GetFromAsMsEpoch(),
-			},
-			QueryType: q.QueryType,
-		})
-	}
-
-	pbRes, err := tw.DataClient.QueryData(ctx, pbQuery)
-	if err != nil {
-		return DataResponse{}, err
-	}
-
-	tR := DataResponse{
-		Results: make(map[string]DataQueryResult, len(pbRes.Responses)),
-	}
-
-	for refID, pRes := range pbRes.Responses {
-		qr := DataQueryResult{
-			RefID:      refID,
-			Dataframes: NewEncodedDataFrames(pRes.Frames),
-		}
-		if len(pRes.JsonMeta) != 0 {
-			qr.Meta = simplejson.NewFromAny(pRes.JsonMeta)
-		}
-		if pRes.Error != "" {
-			qr.Error = fmt.Errorf(pRes.Error)
-			qr.ErrorString = pRes.Error
-		}
-		tR.Results[refID] = qr
-	}
-
-	return tR, nil
-}

pkg/plugins/ifaces.go

@@ -13,8 +13,6 @@ type Manager interface {
 	Renderer() *RendererPlugin
 	// GetDataSource gets a data source plugin with a certain ID.
 	GetDataSource(id string) *DataSourcePlugin
-	// GetDataPlugin gets a data plugin with a certain ID.
-	GetDataPlugin(id string) DataPlugin
 	// GetPlugin gets a plugin with a certain ID.
 	GetPlugin(id string) *PluginBase
 	// GetApp gets an app plugin with a certain ID.

pkg/plugins/manager/manager.go

@@ -745,26 +745,6 @@ func collectPluginFilesWithin(rootDir string) ([]string, error) {
 	return files, err
 }
 
-// GetDataPlugin gets a DataPlugin with a certain name. If none is found, nil is returned.
-//nolint: staticcheck // plugins.DataPlugin deprecated
-func (pm *PluginManager) GetDataPlugin(id string) plugins.DataPlugin {
-	pm.pluginsMu.RLock()
-	defer pm.pluginsMu.RUnlock()
-
-	if p := pm.GetDataSource(id); p != nil && p.CanHandleDataQueries() {
-		return p
-	}
-
-	// XXX: Might other plugins implement DataPlugin?
-
-	p := pm.BackendPluginManager.GetDataPlugin(id)
-	if p != nil {
-		return p.(plugins.DataPlugin)
-	}
-
-	return nil
-}
-
 func (pm *PluginManager) StaticRoutes() []*plugins.PluginStaticRoute {
 	return pm.staticRoutes
 }

pkg/plugins/manager/manager_test.go

@@ -551,8 +551,6 @@ func verifyBundledPluginCatalogue(t *testing.T, pm *PluginManager) {
 }
 
 type fakeBackendPluginManager struct {
-	backendplugin.Manager
-
 	registeredPlugins []string
 }
 
@@ -566,6 +564,10 @@ func (f *fakeBackendPluginManager) RegisterAndStart(ctx context.Context, pluginI
 	return nil
 }
 
+func (f *fakeBackendPluginManager) Get(pluginID string) (backendplugin.Plugin, bool) {
+	return nil, false
+}
+
 func (f *fakeBackendPluginManager) UnregisterAndStop(ctx context.Context, pluginID string) error {
 	var result []string
 
@@ -600,9 +602,15 @@ func (f *fakeBackendPluginManager) CheckHealth(ctx context.Context, pCtx backend
 	return nil, nil
 }
 
+func (f *fakeBackendPluginManager) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	return nil, nil
+}
+
 func (f *fakeBackendPluginManager) CallResource(pluginConfig backend.PluginContext, ctx *models.ReqContext, path string) {
 }
 
+var _ backendplugin.Manager = &fakeBackendPluginManager{}
+
 type fakePluginInstaller struct {
 	installCount   int
 	uninstallCount int

pkg/plugins/renderer_plugin.go

@@ -30,9 +30,7 @@ func (r *RendererPlugin) Load(decoder *json.Decoder, base *PluginBase,
 
 	cmd := ComposePluginStartCommand("plugin_start")
 	fullpath := filepath.Join(base.PluginDir, cmd)
-	factory := grpcplugin.NewRendererPlugin(r.Id, fullpath, grpcplugin.PluginStartFuncs{
-		OnStart: r.onPluginStart,
-	})
+	factory := grpcplugin.NewRendererPlugin(r.Id, fullpath, r.onPluginStart)
 	if err := backendPluginManager.Register(r.Id, factory); err != nil {
 		return nil, errutil.Wrapf(err, "failed to register backend plugin")
 	}
@@ -48,7 +46,7 @@ func (r *RendererPlugin) Start(ctx context.Context) error {
 	return nil
 }
 
-func (r *RendererPlugin) onPluginStart(pluginID string, client *grpcplugin.Client, logger log.Logger) error {
-	r.GrpcPluginV2 = client.RendererPlugin
+func (r *RendererPlugin) onPluginStart(pluginID string, renderer pluginextensionv2.RendererPlugin, logger log.Logger) error {
+	r.GrpcPluginV2 = renderer
 	return nil
 }

pkg/plugins/tsdb.go

@@ -236,6 +236,12 @@ type DataPlugin interface {
 	DataQuery(ctx context.Context, ds *models.DataSource, query DataQuery) (DataResponse, error)
 }
 
+type DataPluginFunc func(ctx context.Context, ds *models.DataSource, query DataQuery) (DataResponse, error)
+
+func (f DataPluginFunc) DataQuery(ctx context.Context, ds *models.DataSource, query DataQuery) (DataResponse, error) {
+	return f(ctx, ds, query)
+}
+
 func NewDataTimeRange(from, to string) DataTimeRange {
 	return DataTimeRange{
 		From: from,

pkg/tsdb/cloudwatch/cloudwatch.go

@@ -100,7 +100,15 @@ func newExecutor(logsService *LogsService, im instancemgmt.InstanceManager, cfg
 
 func NewInstanceSettings() datasource.InstanceFactoryFunc {
 	return func(settings backend.DataSourceInstanceSettings) (instancemgmt.Instance, error) {
-		var jsonData map[string]string
+		jsonData := struct {
+			Profile       string `json:"profile"`
+			Region        string `json:"defaulRegion"`
+			AssumeRoleARN string `json:"assumeRoleArn"`
+			ExternalID    string `json:"externalId"`
+			Endpoint      string `json:"endpoint"`
+			Namespace     string `json:"customMetricsNamespaces"`
+			AuthType      string `json:"authType"`
+		}{}
 
 		err := json.Unmarshal(settings.JSONData, &jsonData)
 		if err != nil {
@@ -108,18 +116,17 @@ func NewInstanceSettings() datasource.InstanceFactoryFunc {
 		}
 
 		model := datasourceInfo{
-			profile:       jsonData["profile"],
-			region:        jsonData["defaultRegion"],
-			assumeRoleARN: jsonData["assumeRoleArn"],
-			externalID:    jsonData["externalId"],
-			endpoint:      jsonData["endpoint"],
-			namespace:     jsonData["customMetricsNamespaces"],
+			profile:       jsonData.Profile,
+			region:        jsonData.Region,
+			assumeRoleARN: jsonData.AssumeRoleARN,
+			externalID:    jsonData.ExternalID,
+			endpoint:      jsonData.Endpoint,
+			namespace:     jsonData.Namespace,
 			datasourceID:  settings.ID,
 		}
 
-		atStr := jsonData["authType"]
 		at := awsds.AuthTypeDefault
-		switch atStr {
+		switch jsonData.AuthType {
 		case "credentials":
 			at = awsds.AuthTypeSharedCreds
 		case "keys":
@@ -132,7 +139,7 @@ func NewInstanceSettings() datasource.InstanceFactoryFunc {
 			at = awsds.AuthTypeDefault
 			plog.Warn("Authentication type \"arn\" is deprecated, falling back to default")
 		default:
-			plog.Warn("Unrecognized AWS authentication type", "type", atStr)
+			plog.Warn("Unrecognized AWS authentication type", "type", jsonData.AuthType)
 		}
 
 		model.authType = at

pkg/tsdb/data_plugin_adapter.go

@@ -0,0 +1,115 @@
+package tsdb
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/models"
+	"github.com/grafana/grafana/pkg/plugins"
+	"github.com/grafana/grafana/pkg/plugins/adapters"
+	"github.com/grafana/grafana/pkg/services/oauthtoken"
+)
+
+// nolint:staticcheck // plugins.DataQuery deprecated
+func dataPluginQueryAdapter(pluginID string, handler backend.QueryDataHandler) plugins.DataPluginFunc {
+	return plugins.DataPluginFunc(func(ctx context.Context, ds *models.DataSource, query plugins.DataQuery) (plugins.DataResponse, error) {
+		instanceSettings, err := modelToInstanceSettings(ds)
+		if err != nil {
+			return plugins.DataResponse{}, err
+		}
+
+		if query.Headers == nil {
+			query.Headers = make(map[string]string)
+		}
+
+		if oauthtoken.IsOAuthPassThruEnabled(ds) {
+			if token := oauthtoken.GetCurrentOAuthToken(ctx, query.User); token != nil {
+				delete(query.Headers, "Authorization")
+				query.Headers["Authorization"] = fmt.Sprintf("%s %s", token.Type(), token.AccessToken)
+			}
+		}
+
+		req := &backend.QueryDataRequest{
+			PluginContext: backend.PluginContext{
+				OrgID:                      ds.OrgId,
+				PluginID:                   pluginID,
+				User:                       adapters.BackendUserFromSignedInUser(query.User),
+				DataSourceInstanceSettings: instanceSettings,
+			},
+			Queries: []backend.DataQuery{},
+			Headers: query.Headers,
+		}
+
+		for _, q := range query.Queries {
+			modelJSON, err := q.Model.MarshalJSON()
+			if err != nil {
+				return plugins.DataResponse{}, err
+			}
+			req.Queries = append(req.Queries, backend.DataQuery{
+				RefID:         q.RefID,
+				Interval:      time.Duration(q.IntervalMS) * time.Millisecond,
+				MaxDataPoints: q.MaxDataPoints,
+				TimeRange: backend.TimeRange{
+					From: query.TimeRange.GetFromAsTimeUTC(),
+					To:   query.TimeRange.GetToAsTimeUTC(),
+				},
+				QueryType: q.QueryType,
+				JSON:      modelJSON,
+			})
+		}
+
+		resp, err := handler.QueryData(ctx, req)
+		if err != nil {
+			return plugins.DataResponse{}, err
+		}
+
+		tR := plugins.DataResponse{
+			Results: make(map[string]plugins.DataQueryResult, len(resp.Responses)),
+		}
+
+		for refID, r := range resp.Responses {
+			qr := plugins.DataQueryResult{
+				RefID: refID,
+			}
+
+			for _, f := range r.Frames {
+				if f.RefID == "" {
+					f.RefID = refID
+				}
+			}
+
+			qr.Dataframes = plugins.NewDecodedDataFrames(r.Frames)
+
+			if r.Error != nil {
+				qr.Error = r.Error
+			}
+
+			tR.Results[refID] = qr
+		}
+
+		return tR, nil
+	})
+}
+
+func modelToInstanceSettings(ds *models.DataSource) (*backend.DataSourceInstanceSettings, error) {
+	jsonDataBytes, err := ds.JsonData.MarshalJSON()
+	if err != nil {
+		return nil, err
+	}
+
+	return &backend.DataSourceInstanceSettings{
+		ID:                      ds.Id,
+		Name:                    ds.Name,
+		URL:                     ds.Url,
+		Database:                ds.Database,
+		User:                    ds.User,
+		BasicAuthEnabled:        ds.BasicAuth,
+		BasicAuthUser:           ds.BasicAuthUser,
+		JSONData:                jsonDataBytes,
+		DecryptedSecureJSONData: ds.DecryptedValues(),
+		Updated:                 ds.Updated,
+		UID:                     ds.Uid,
+	}, nil
+}

pkg/tsdb/service.go

@@ -7,11 +7,11 @@ import (
 	"github.com/grafana/grafana/pkg/infra/httpclient"
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/plugins"
+	"github.com/grafana/grafana/pkg/plugins/backendplugin"
 	"github.com/grafana/grafana/pkg/registry"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/tsdb/azuremonitor"
 	"github.com/grafana/grafana/pkg/tsdb/cloudmonitoring"
-	"github.com/grafana/grafana/pkg/tsdb/cloudwatch"
 	"github.com/grafana/grafana/pkg/tsdb/elasticsearch"
 	"github.com/grafana/grafana/pkg/tsdb/graphite"
 	"github.com/grafana/grafana/pkg/tsdb/influxdb"
@@ -42,13 +42,13 @@ func init() {
 
 // Service handles data requests to data sources.
 type Service struct {
-	Cfg                    *setting.Cfg                  `inject:""`
-	CloudWatchService      *cloudwatch.CloudWatchService `inject:""`
-	PostgresService        *postgres.PostgresService     `inject:""`
-	CloudMonitoringService *cloudmonitoring.Service      `inject:""`
-	AzureMonitorService    *azuremonitor.Service         `inject:""`
-	PluginManager          plugins.Manager               `inject:""`
-	HTTPClientProvider     httpclient.Provider           `inject:""`
+	Cfg                    *setting.Cfg              `inject:""`
+	PostgresService        *postgres.PostgresService `inject:""`
+	CloudMonitoringService *cloudmonitoring.Service  `inject:""`
+	AzureMonitorService    *azuremonitor.Service     `inject:""`
+	PluginManager          plugins.Manager           `inject:""`
+	BackendPluginManager   backendplugin.Manager     `inject:""`
+	HTTPClientProvider     httpclient.Provider       `inject:""`
 
 	//nolint: staticcheck // plugins.DataPlugin deprecated
 	registry map[string]func(*models.DataSource) (plugins.DataPlugin, error)
@@ -72,25 +72,19 @@ func (s *Service) Init() error {
 }
 
 //nolint: staticcheck // plugins.DataPlugin deprecated
-func (s *Service) HandleRequest(ctx context.Context, ds *models.DataSource, query plugins.DataQuery) (
-	plugins.DataResponse, error) {
-	plugin := s.PluginManager.GetDataPlugin(ds.Type)
-	if plugin == nil {
-		factory, exists := s.registry[ds.Type]
-		if !exists {
-			return plugins.DataResponse{}, fmt.Errorf(
-				"could not find plugin corresponding to data source type: %q", ds.Type)
-		}
-
+func (s *Service) HandleRequest(ctx context.Context, ds *models.DataSource, query plugins.DataQuery) (plugins.DataResponse, error) {
+	if factory, exists := s.registry[ds.Type]; exists {
 		var err error
-		plugin, err = factory(ds)
+		plugin, err := factory(ds)
 		if err != nil {
 			return plugins.DataResponse{}, fmt.Errorf("could not instantiate endpoint for data plugin %q: %w",
 				ds.Type, err)
 		}
+
+		return plugin.DataQuery(ctx, ds, query)
 	}
 
-	return plugin.DataQuery(ctx, ds, query)
+	return dataPluginQueryAdapter(ds.Type, s.BackendPluginManager).DataQuery(ctx, ds, query)
 }
 
 // RegisterQueryHandler registers a query handler factory.

pkg/tsdb/service_test.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"testing"
 
+	"github.com/grafana/grafana-plugin-sdk-go/backend"
+	"github.com/grafana/grafana/pkg/components/simplejson"
 	"github.com/grafana/grafana/pkg/models"
 	"github.com/grafana/grafana/pkg/plugins"
 	"github.com/grafana/grafana/pkg/plugins/backendplugin"
@@ -19,7 +21,7 @@ func TestHandleRequest(t *testing.T) {
 			},
 		}
 
-		svc, exe := createService()
+		svc, exe, _ := createService()
 		exe.Return("A", plugins.DataTimeSeriesSlice{plugins.DataTimeSeries{Name: "argh"}})
 
 		res, err := svc.HandleRequest(context.TODO(), &models.DataSource{Id: 1, Type: "test"}, req)
@@ -36,7 +38,7 @@ func TestHandleRequest(t *testing.T) {
 			},
 		}
 
-		svc, exe := createService()
+		svc, exe, _ := createService()
 		exe.Return("A", plugins.DataTimeSeriesSlice{plugins.DataTimeSeries{Name: "argh"}})
 		exe.Return("B", plugins.DataTimeSeriesSlice{plugins.DataTimeSeries{Name: "barg"}})
 
@@ -48,16 +50,28 @@ func TestHandleRequest(t *testing.T) {
 		require.Equal(t, "barg", res.Results["B"].Series[0].Name)
 	})
 
-	t.Run("Should return error when handling request for query with unknown type", func(t *testing.T) {
-		svc, _ := createService()
+	t.Run("Should fallback to backend plugin manager when handling request for query with unregistered type", func(t *testing.T) {
+		svc, _, manager := createService()
+		backendPluginManagerCalled := false
+		manager.QueryDataHandlerFunc = backend.QueryDataHandlerFunc(func(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+			backendPluginManagerCalled = true
+			return &backend.QueryDataResponse{}, nil
+		})
 
+		ds := &models.DataSource{Id: 12, Type: "unregisteredType", JsonData: simplejson.New()}
 		req := plugins.DataQuery{
+			TimeRange: &plugins.DataTimeRange{},
 			Queries: []plugins.DataSubQuery{
-				{RefID: "A", DataSource: &models.DataSource{Id: 1, Type: "asdasdas"}},
+				{
+					RefID:      "A",
+					DataSource: ds,
+					Model:      simplejson.New(),
+				},
 			},
 		}
-		_, err := svc.HandleRequest(context.TODO(), &models.DataSource{Id: 12, Type: "testjughjgjg"}, req)
-		require.Error(t, err)
+		_, err := svc.HandleRequest(context.Background(), ds, req)
+		require.NoError(t, err)
+		require.True(t, backendPluginManagerCalled)
 	})
 }
 
@@ -99,17 +113,22 @@ func (e *fakeExecutor) HandleQuery(refId string, fn resultsFn) {
 
 type fakeBackendPM struct {
 	backendplugin.Manager
+	backend.QueryDataHandlerFunc
 }
 
-func (pm fakeBackendPM) GetDataPlugin(string) interface{} {
-	return nil
-}
-
-func createService() (Service, *fakeExecutor) {
-	s := NewService()
-	s.PluginManager = &manager.PluginManager{
-		BackendPluginManager: fakeBackendPM{},
+func (m *fakeBackendPM) QueryData(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
+	if m.QueryDataHandlerFunc != nil {
+		return m.QueryDataHandlerFunc.QueryData(ctx, req)
 	}
+
+	return nil, nil
+}
+
+func createService() (Service, *fakeExecutor, *fakeBackendPM) {
+	s := NewService()
+	fakeBackendPluginManager := &fakeBackendPM{}
+	s.PluginManager = &manager.PluginManager{}
+	s.BackendPluginManager = fakeBackendPluginManager
 	e := &fakeExecutor{
 		//nolint: staticcheck // plugins.DataPlugin deprecated
 		results:   make(map[string]plugins.DataQueryResult),
@@ -120,5 +139,5 @@ func createService() (Service, *fakeExecutor) {
 		return e, nil
 	}
 
-	return s, e
+	return s, e, fakeBackendPluginManager
 }