IntenseWebs/grafana
3
0
Fork 0
mirror of https://github.com/grafana/grafana.git synced 2025-02-25 18:55:37 -06:00
Code Issues Packages Projects Releases Wiki Activity

Plugins: Refactor creation of plugin context to dedicated service (#66451)

Browse Source 
* first pass

* fix tests

* return errs

* change signature

* tidy

* delete unnecessary fields from test

* tidy

* fix tests

* simplify

* separate error check in API

* apply nits
This commit is contained in:
Will Browne
2023-06-08 13:59:51 +02:00
committed by GitHub
parent 49ad802ca6
commit 624777258b
25 changed files with 308 additions and 261 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
pkg/services/live/features/plugin.go
View File

@@ -2,6 +2,7 @@ package features
import (
"context"
"errors"
"github.com/centrifugal/centrifuge"
"github.com/grafana/grafana-plugin-sdk-go/backend"
@@ -9,13 +10,14 @@ import (
"github.com/grafana/grafana/pkg/services/live/model"
"github.com/grafana/grafana/pkg/services/live/orgchannel"
"github.com/grafana/grafana/pkg/services/live/runstream"
"github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext"
"github.com/grafana/grafana/pkg/services/user"
)
//go:generate mockgen -destination=plugin_mock.go -package=features github.com/grafana/grafana/pkg/services/live/features PluginContextGetter
type PluginContextGetter interface {
GetPluginContext(ctx context.Context, user *user.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error)
GetPluginContext(ctx context.Context, user *user.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, error)
}
// PluginRunner can handle streaming operations for channels belonging to plugins.
@@ -62,15 +64,15 @@ type PluginPathRunner struct {
// OnSubscribe passes control to a plugin.
func (r *PluginPathRunner) OnSubscribe(ctx context.Context, user *user.SignedInUser, e model.SubscribeEvent) (model.SubscribeReply, backend.SubscribeStreamStatus, error) {
pCtx, found, err := r.pluginContextGetter.GetPluginContext(ctx, user, r.pluginID, r.datasourceUID, false)
pCtx, err := r.pluginContextGetter.GetPluginContext(ctx, user, r.pluginID, r.datasourceUID, false)
if err != nil {
if errors.Is(err, plugincontext.ErrPluginNotFound) {
logger.Error("Plugin context not found", "path", r.path)
return model.SubscribeReply{}, 0, centrifuge.ErrorInternal
}
logger.Error("Get plugin context error", "error", err, "path", r.path)
return model.SubscribeReply{}, 0, err
}
if !found {
logger.Error("Plugin context not found", "path", r.path)
return model.SubscribeReply{}, 0, centrifuge.ErrorInternal
}
resp, err := r.handler.SubscribeStream(ctx, &backend.SubscribeStreamRequest{
PluginContext: pCtx,
Path: r.path,
@@ -106,15 +108,15 @@ func (r *PluginPathRunner) OnSubscribe(ctx context.Context, user *user.SignedInU
// OnPublish passes control to a plugin.
func (r *PluginPathRunner) OnPublish(ctx context.Context, user *user.SignedInUser, e model.PublishEvent) (model.PublishReply, backend.PublishStreamStatus, error) {
pCtx, found, err := r.pluginContextGetter.GetPluginContext(ctx, user, r.pluginID, r.datasourceUID, false)
pCtx, err := r.pluginContextGetter.GetPluginContext(ctx, user, r.pluginID, r.datasourceUID, false)
if err != nil {
if errors.Is(err, plugincontext.ErrPluginNotFound) {
logger.Error("Plugin context not found", "path", r.path)
return model.PublishReply{}, 0, centrifuge.ErrorInternal
}
logger.Error("Get plugin context error", "error", err, "path", r.path)
return model.PublishReply{}, 0, err
}
if !found {
logger.Error("Plugin context not found", "path", r.path)
return model.PublishReply{}, 0, centrifuge.ErrorInternal
}
resp, err := r.handler.PublishStream(ctx, &backend.PublishStreamRequest{
PluginContext: pCtx,
Path: r.path,

18
pkg/services/live/features/plugin_mock.go
View File

@@ -5,13 +5,12 @@
package features
import (
"context"
context "context"
reflect "reflect"
gomock "github.com/golang/mock/gomock"
backend "github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/services/user"
user "github.com/grafana/grafana/pkg/services/user"
)
// MockPluginContextGetter is a mock of PluginContextGetter interface.
@@ -38,17 +37,16 @@ func (m *MockPluginContextGetter) EXPECT() *MockPluginContextGetterMockRecorder
}
// GetPluginContext mocks base method.
func (m *MockPluginContextGetter) GetPluginContext(ctx context.Context, arg0 *user.SignedInUser, arg1, arg2 string) (backend.PluginContext, bool, error) {
func (m *MockPluginContextGetter) GetPluginContext(arg0 context.Context, arg1 *user.SignedInUser, arg2, arg3 string, arg4 bool) (backend.PluginContext, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetPluginContext", arg0, arg1, arg2)
ret := m.ctrl.Call(m, "GetPluginContext", arg0, arg1, arg2, arg3, arg4)
ret0, _ := ret[0].(backend.PluginContext)
ret1, _ := ret[1].(bool)
ret2, _ := ret[2].(error)
return ret0, ret1, ret2
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetPluginContext indicates an expected call of GetPluginContext.
func (mr *MockPluginContextGetterMockRecorder) GetPluginContext(ctx context.Context, arg0, arg1, arg2 interface{}) *gomock.Call {
func (mr *MockPluginContextGetterMockRecorder) GetPluginContext(arg0, arg1, arg2, arg3, arg4 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPluginContext", reflect.TypeOf((*MockPluginContextGetter)(nil).GetPluginContext), arg0, arg1, arg2)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPluginContext", reflect.TypeOf((*MockPluginContextGetter)(nil).GetPluginContext), arg0, arg1, arg2, arg3, arg4)
}

4
pkg/services/live/liveplugin/plugin.go
View File

@@ -72,14 +72,14 @@ func NewContextGetter(pluginContextProvider *plugincontext.Provider, dataSourceC
}
}
func (g *ContextGetter) GetPluginContext(ctx context.Context, user *user.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error) {
func (g *ContextGetter) GetPluginContext(ctx context.Context, user *user.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, error) {
if datasourceUID == "" {
return g.pluginContextProvider.Get(ctx, pluginID, user)
}
ds, err := g.dataSourceCache.GetDatasourceByUID(ctx, datasourceUID, user, skipCache)
if err != nil {
return backend.PluginContext{}, false, fmt.Errorf("%v: %w", "Failed to get datasource", err)
return backend.PluginContext{}, fmt.Errorf("%v: %w", "Failed to get datasource", err)
}
return g.pluginContextProvider.GetWithDataSource(ctx, pluginID, user, ds)
}

31
pkg/services/live/runstream/manager.go
View File

@@ -11,6 +11,7 @@ import (
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/services/pluginsintegration/plugincontext"
"github.com/grafana/grafana/pkg/services/user"
)
@@ -25,7 +26,7 @@ type ChannelLocalPublisher interface {
}
type PluginContextGetter interface {
GetPluginContext(ctx context.Context, user *user.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, bool, error)
GetPluginContext(ctx context.Context, user *user.SignedInUser, pluginID string, datasourceUID string, skipCache bool) (backend.PluginContext, error)
}
type NumLocalSubscribersGetter interface {
@@ -182,15 +183,15 @@ func (s *Manager) watchStream(ctx context.Context, cancelFn func(), sr streamReq
case <-datasourceTicker.C:
if sr.PluginContext.DataSourceInstanceSettings != nil {
dsUID := sr.PluginContext.DataSourceInstanceSettings.UID
pCtx, ok, err := s.pluginContextGetter.GetPluginContext(ctx, sr.user, sr.PluginContext.PluginID, dsUID, false)
pCtx, err := s.pluginContextGetter.GetPluginContext(ctx, sr.user, sr.PluginContext.PluginID, dsUID, false)
if err != nil {
if errors.Is(err, plugincontext.ErrPluginNotFound) {
logger.Debug("Datasource not found, stop stream", "channel", sr.Channel, "path", sr.Path)
return
}
logger.Error("Error getting datasource context", "channel", sr.Channel, "path", sr.Path, "error", err)
continue
}
if !ok {
logger.Debug("Datasource not found, stop stream", "channel", sr.Channel, "path", sr.Path)
return
}
if pCtx.DataSourceInstanceSettings.Updated != sr.PluginContext.DataSourceInstanceSettings.Updated {
logger.Debug("Datasource changed, re-establish stream", "channel", sr.Channel, "path", sr.Path)
err := s.HandleDatasourceUpdate(pCtx.OrgID, dsUID)
@@ -283,16 +284,16 @@ func (s *Manager) runStream(ctx context.Context, cancelFn func(), sr streamReque
if pluginCtx.DataSourceInstanceSettings != nil {
datasourceUID = pluginCtx.DataSourceInstanceSettings.UID
}
newPluginCtx, ok, err := s.pluginContextGetter.GetPluginContext(ctx, sr.user, pluginCtx.PluginID, datasourceUID, false)
newPluginCtx, err := s.pluginContextGetter.GetPluginContext(ctx, sr.user, pluginCtx.PluginID, datasourceUID, false)
if err != nil {
if errors.Is(err, plugincontext.ErrPluginNotFound) {
logger.Info("No plugin context found, stopping stream", "path", sr.Path)
return
}
logger.Error("Error getting plugin context", "path", sr.Path, "error", err)
isReconnect = true
continue
}
if !ok {
logger.Info("No plugin context found, stopping stream", "path", sr.Path)
return
}
pluginCtx = newPluginCtx
}
@@ -407,13 +408,13 @@ func (s *Manager) SubmitStream(ctx context.Context, user *user.SignedInUser, cha
if pCtx.DataSourceInstanceSettings != nil {
datasourceUID = pCtx.DataSourceInstanceSettings.UID
}
newPluginCtx, ok, err := s.pluginContextGetter.GetPluginContext(ctx, user, pCtx.PluginID, datasourceUID, false)
newPluginCtx, err := s.pluginContextGetter.GetPluginContext(ctx, user, pCtx.PluginID, datasourceUID, false)
if err != nil {
if errors.Is(err, plugincontext.ErrPluginNotFound) {
return nil, errDatasourceNotFound
}
return nil, err
}
if !ok {
return nil, errDatasourceNotFound
}
pCtx = newPluginCtx
}

16
pkg/services/live/runstream/manager_test.go
View File

@@ -83,7 +83,7 @@ func TestStreamManager_SubmitStream_Send(t *testing.T) {
mockStreamRunner := NewMockStreamRunner(mockCtrl)
mockStreamRunner.EXPECT().RunStream(
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any(),
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
require.Equal(t, "test", req.Path)
close(startedCh)
@@ -139,7 +139,7 @@ func TestStreamManager_SubmitStream_DifferentOrgID(t *testing.T) {
mockStreamRunner1 := NewMockStreamRunner(mockCtrl)
mockStreamRunner1.EXPECT().RunStream(
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any(),
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
require.Equal(t, "test", req.Path)
close(startedCh1)
@@ -152,7 +152,7 @@ func TestStreamManager_SubmitStream_DifferentOrgID(t *testing.T) {
mockStreamRunner2 := NewMockStreamRunner(mockCtrl)
mockStreamRunner2.EXPECT().RunStream(
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any(),
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
require.Equal(t, "test", req.Path)
close(startedCh2)
@@ -212,7 +212,7 @@ func TestStreamManager_SubmitStream_CloseNoSubscribers(t *testing.T) {
mockNumSubscribersGetter.EXPECT().GetNumLocalSubscribers("1/test").Return(0, nil).Times(3)
mockStreamRunner := NewMockStreamRunner(mockCtrl)
mockStreamRunner.EXPECT().RunStream(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
mockStreamRunner.EXPECT().RunStream(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
close(startedCh)
<-ctx.Done()
close(doneCh)
@@ -264,7 +264,7 @@ func TestStreamManager_SubmitStream_ErrorRestartsRunStream(t *testing.T) {
mockStreamRunner := NewMockStreamRunner(mockCtrl)
mockStreamRunner.EXPECT().RunStream(
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any(),
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
if currentErrors >= numErrors {
return nil
@@ -302,7 +302,7 @@ func TestStreamManager_SubmitStream_NilErrorStopsRunStream(t *testing.T) {
mockStreamRunner := NewMockStreamRunner(mockCtrl)
mockStreamRunner.EXPECT().RunStream(
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any(),
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
return nil
}).Times(1)
@@ -352,7 +352,7 @@ func TestStreamManager_HandleDatasourceUpdate(t *testing.T) {
mockStreamRunner := NewMockStreamRunner(mockCtrl)
mockStreamRunner.EXPECT().RunStream(
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any(),
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
if isFirstCall {
// first RunStream will wait till context done.
@@ -415,7 +415,7 @@ func TestStreamManager_HandleDatasourceDelete(t *testing.T) {
mockStreamRunner := NewMockStreamRunner(mockCtrl)
mockStreamRunner.EXPECT().RunStream(
gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(),
gomock.Any(), gomock.Any(), gomock.Any(),
).DoAndReturn(func(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
close(doneCh)
<-ctx.Done()

18
pkg/services/live/runstream/mock.go
View File

@@ -10,8 +10,7 @@ import (
gomock "github.com/golang/mock/gomock"
backend "github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/services/user"
user "github.com/grafana/grafana/pkg/services/user"
)
// MockChannelLocalPublisher is a mock of ChannelLocalPublisher interface.
@@ -121,7 +120,7 @@ func (m *MockStreamRunner) RunStream(arg0 context.Context, arg1 *backend.RunStre
}
// RunStream indicates an expected call of RunStream.
func (mr *MockStreamRunnerMockRecorder) RunStream(ctx, arg0, arg1, arg2 interface{}) *gomock.Call {
func (mr *MockStreamRunnerMockRecorder) RunStream(arg0, arg1, arg2 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RunStream", reflect.TypeOf((*MockStreamRunner)(nil).RunStream), arg0, arg1, arg2)
}
@@ -150,17 +149,16 @@ func (m *MockPluginContextGetter) EXPECT() *MockPluginContextGetterMockRecorder
}
// GetPluginContext mocks base method.
func (m *MockPluginContextGetter) GetPluginContext(ctx context.Context, arg0 *user.SignedInUser, arg1, arg2 string, arg3 bool) (backend.PluginContext, bool, error) {
func (m *MockPluginContextGetter) GetPluginContext(arg0 context.Context, arg1 *user.SignedInUser, arg2, arg3 string, arg4 bool) (backend.PluginContext, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetPluginContext", ctx, arg0, arg1, arg2, arg3)
ret := m.ctrl.Call(m, "GetPluginContext", arg0, arg1, arg2, arg3, arg4)
ret0, _ := ret[0].(backend.PluginContext)
ret1, _ := ret[1].(bool)
ret2, _ := ret[2].(error)
return ret0, ret1, ret2
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetPluginContext indicates an expected call of GetPluginContext.
func (mr *MockPluginContextGetterMockRecorder) GetPluginContext(ctx, arg0, arg1, arg2, arg3 interface{}) *gomock.Call {
func (mr *MockPluginContextGetterMockRecorder) GetPluginContext(arg0, arg1, arg2, arg3, arg4 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPluginContext", reflect.TypeOf((*MockPluginContextGetter)(nil).GetPluginContext), ctx, arg0, arg1, arg2, arg3)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetPluginContext", reflect.TypeOf((*MockPluginContextGetter)(nil).GetPluginContext), arg0, arg1, arg2, arg3, arg4)
}