Plugins: Add SDK stream handler support to plugin.Client (#41650)

* add stream handler support to plugin.Client

* nil instead of empty resp

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

* nil instead of empty resp #2

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

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

pkg/plugins/ifaces.go

@@ -54,6 +54,7 @@ type UpdateInfo struct {
 type Client interface {
 	backend.QueryDataHandler
 	backend.CheckHealthHandler
+	backend.StreamHandler
 
 	// CallResource calls a plugin resource.
 	CallResource(pCtx backend.PluginContext, ctx *models.ReqContext, path string)

pkg/plugins/manager/loader/loader.go

@@ -46,7 +46,7 @@ func New(license models.Licensing, cfg *setting.Cfg, authorizer plugins.PluginLo
 		cfg:                cfg,
 		pluginFinder:       finder.New(cfg),
 		pluginInitializer:  initializer.New(cfg, license),
-		signatureValidator: signature.NewValidator(cfg, authorizer),
+		signatureValidator: signature.NewValidator(authorizer),
 		errs:               make(map[string]*plugins.SignatureError),
 	}
 }

pkg/plugins/manager/loader/loader_test.go

@@ -370,6 +370,7 @@ func TestLoader_Load_MultiplePlugins(t *testing.T) {
 			{
 				name: "Load multiple plugins (broken, valid, unsigned)",
 				cfg: &setting.Cfg{
+					Env:         "production",
 					PluginsPath: filepath.Join(parentDir),
 				},
 				appURL: "http://localhost:3000",
@@ -884,7 +885,7 @@ func newLoader(cfg *setting.Cfg) *Loader {
 		cfg:                cfg,
 		pluginFinder:       finder.New(cfg),
 		pluginInitializer:  initializer.New(cfg, &fakeLicensingService{}),
-		signatureValidator: signature.NewValidator(cfg, &signature.UnsignedPluginAuthorizer{Cfg: cfg}),
+		signatureValidator: signature.NewValidator(&signature.UnsignedPluginAuthorizer{Cfg: cfg}),
 		errs:               make(map[string]*plugins.SignatureError),
 	}
 }

pkg/plugins/manager/manager.go

@@ -476,6 +476,33 @@ func (m *PluginManager) CheckHealth(ctx context.Context, req *backend.CheckHealt
 	return resp, nil
 }
 
+func (m *PluginManager) SubscribeStream(ctx context.Context, req *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
+	plugin := m.Plugin(req.PluginContext.PluginID)
+	if plugin == nil {
+		return nil, backendplugin.ErrPluginNotRegistered
+	}
+
+	return plugin.SubscribeStream(ctx, req)
+}
+
+func (m *PluginManager) PublishStream(ctx context.Context, req *backend.PublishStreamRequest) (*backend.PublishStreamResponse, error) {
+	plugin := m.Plugin(req.PluginContext.PluginID)
+	if plugin == nil {
+		return nil, backendplugin.ErrPluginNotRegistered
+	}
+
+	return plugin.PublishStream(ctx, req)
+}
+
+func (m *PluginManager) RunStream(ctx context.Context, req *backend.RunStreamRequest, sender *backend.StreamSender) error {
+	plugin := m.Plugin(req.PluginContext.PluginID)
+	if plugin == nil {
+		return backendplugin.ErrPluginNotRegistered
+	}
+
+	return plugin.RunStream(ctx, req, sender)
+}
+
 func (m *PluginManager) isRegistered(pluginID string) bool {
 	p := m.Plugin(pluginID)
 	if p == nil {

pkg/plugins/manager/signature/signature.go

@@ -3,19 +3,16 @@ package signature
 import (
 	"github.com/grafana/grafana/pkg/infra/log"
 	"github.com/grafana/grafana/pkg/plugins"
-	"github.com/grafana/grafana/pkg/setting"
 )
 
 var logger = log.New("plugin.signature.validator")
 
 type Validator struct {
-	cfg        *setting.Cfg
 	authorizer plugins.PluginLoaderAuthorizer
 }
 
-func NewValidator(cfg *setting.Cfg, authorizer plugins.PluginLoaderAuthorizer) Validator {
+func NewValidator(authorizer plugins.PluginLoaderAuthorizer) Validator {
 	return Validator{
-		cfg:        cfg,
 		authorizer: authorizer,
 	}
 }