From 8ea75c940154a9d5d5458e3060a07ccc09ff2d80 Mon Sep 17 00:00:00 2001
From: Will Browne <wbrowne@users.noreply.github.com>
Date: Mon, 15 Nov 2021 13:25:13 +0000
Subject: [PATCH] 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                      |  1 +
 pkg/plugins/manager/loader/loader.go       |  2 +-
 pkg/plugins/manager/loader/loader_test.go  |  3 ++-
 pkg/plugins/manager/manager.go             | 27 ++++++++++++++++++++++
 pkg/plugins/manager/signature/signature.go |  5 +---
 5 files changed, 32 insertions(+), 6 deletions(-)

diff --git a/pkg/plugins/ifaces.go b/pkg/plugins/ifaces.go
index 7cc2877682d..b43c02aebc5 100644
--- a/pkg/plugins/ifaces.go
+++ b/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)
diff --git a/pkg/plugins/manager/loader/loader.go b/pkg/plugins/manager/loader/loader.go
index 3fae1fb07e1..6a532743ee5 100644
--- a/pkg/plugins/manager/loader/loader.go
+++ b/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),
 	}
 }
diff --git a/pkg/plugins/manager/loader/loader_test.go b/pkg/plugins/manager/loader/loader_test.go
index d271a927074..9afc6fc748c 100644
--- a/pkg/plugins/manager/loader/loader_test.go
+++ b/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),
 	}
 }
diff --git a/pkg/plugins/manager/manager.go b/pkg/plugins/manager/manager.go
index 721234b9b36..520cd2cf9af 100644
--- a/pkg/plugins/manager/manager.go
+++ b/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 {
diff --git a/pkg/plugins/manager/signature/signature.go b/pkg/plugins/manager/signature/signature.go
index 92ef990cc35..ea227e190a1 100644
--- a/pkg/plugins/manager/signature/signature.go
+++ b/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,
 	}
 }