mirror of
https://github.com/grafana/grafana.git
synced 2025-09-22 11:43:24 +08:00
Plugins: Refactor Plugin Management (#40477)
* add core plugin flow * add instrumentation * move func * remove cruft * support external backend plugins * refactor + clean up * remove comments * refactor loader * simplify core plugin path arg * cleanup loggers * move signature validator to plugins package * fix sig packaging * cleanup plugin model * remove unnecessary plugin field * add start+stop for pm * fix failures * add decommissioned state * export fields just to get things flowing * fix comments * set static routes * make image loading idempotent * merge with backend plugin manager * re-use funcs * reorder imports + remove unnecessary interface * add some TODOs + remove unused func * remove unused instrumentation func * simplify client usage * remove import alias * re-use backendplugin.Plugin interface * re order funcs * improve var name * fix log statements * refactor data model * add logic for dupe check during loading * cleanup state setting * refactor loader * cleanup manager interface * add rendering flow * refactor loading + init * add renderer support * fix renderer plugin * reformat imports * track errors * fix plugin signature inheritance * name param in interface * update func comment * fix func arg name * introduce class concept * remove func * fix external plugin check * apply changes from pm-experiment * fix core plugins * fix imports * rename interface * comment API interface * add support for testdata plugin * enable alerting + use correct core plugin contracts * slim manager API * fix param name * fix filter * support static routes * fix rendering * tidy rendering * get tests compiling * fix install+uninstall * start finder test * add finder test coverage * start loader tests * add test for core plugins * load core + bundled test * add test for nested plugin loading * add test files * clean interface + fix registering some core plugins * refactoring * reformat and create sub packages * simplify core plugin init * fix ctx cancel scenario * migrate initializer * remove Init() funcs * add test starter * new logger * flesh out initializer tests * refactoring * remove unused svc * refactor rendering flow * fixup loader tests * add enabled helper func * fix logger name * fix data fetchers * fix case where plugin dir doesn't exist * improve coverage + move dupe checking to loader * remove noisy debug logs * register core plugins automagically * add support for renderer in catalog * make private func + fix req validation * use interface * re-add check for renderer in catalog * tidy up from moving to auto reg core plugins * core plugin registrar * guards * copy over core plugins for test infra * all tests green * renames * propagate new interfaces * kill old manager * get compiling * tidy up * update naming * refactor manager test + cleanup * add more cases to finder test * migrate validator to field * more coverage * refactor dupe checking * add test for plugin class * add coverage for initializer * split out rendering * move * fixup tests * fix uss test * fix frontend settings * fix grafanads test * add check when checking sig errors * fix enabled map * fixup * allow manual setup of CM * rename to cloud-monitoring * remove TODO * add installer interface for testing * loader interface returns * tests passing * refactor + add more coverage * support 'stackdriver' * fix frontend settings loading * improve naming based on package name * small tidy * refactor test * fix renderer start * make cloud-monitoring plugin ID clearer * add plugin update test * add integration tests * don't break all if sig can't be calculated * add root URL check test * add more signature verification tests * update DTO name * update enabled plugins comment * update comments * fix linter * revert fe naming change * fix errors endpoint * reset error code field name * re-order test to help verify * assert -> require * pm check * add missing entry + re-order * re-check * dump icon log * verify manager contents first * reformat * apply PR feedback * apply style changes * fix one vs all loading err * improve log output * only start when no signature error * move log * rework plugin update check * fix test * fix multi loading from cfg.PluginSettings * improve log output #2 * add error abstraction to capture errors without registering a plugin * add debug log * add unsigned warning * e2e test attempt * fix logger * set home path * prevent panic * alternate * ugh.. fix home path * return renderer even if not started * make renderer plugin managed * add fallback renderer icon, update renderer badge + prevent changes when renderer is installed * fix icon loading * rollback renderer changes * use correct field * remove unneccessary block * remove newline * remove unused func * fix bundled plugins base + module fields * remove unused field since refactor * add authorizer abstraction * loader only returns plugins expected to run * fix multi log output
This commit is contained in:
Will Browne
@ -5,33 +5,8 @@ import (
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/models"
|
||||
)
|
||||
|
||||
// Manager manages backend plugins.
|
||||
type Manager interface {
|
||||
//Register registers a backend plugin
|
||||
Register(pluginID string, factory PluginFactoryFunc) error
|
||||
// RegisterAndStart registers and starts a backend plugin
|
||||
RegisterAndStart(ctx context.Context, pluginID string, factory PluginFactoryFunc) error
|
||||
// UnregisterAndStop unregisters and stops a backend plugin
|
||||
UnregisterAndStop(ctx context.Context, pluginID string) error
|
||||
// IsRegistered checks if a plugin is registered with the manager
|
||||
IsRegistered(pluginID string) bool
|
||||
// StartPlugin starts a non-managed backend plugin
|
||||
StartPlugin(ctx context.Context, pluginID string) error
|
||||
// CollectMetrics collects metrics from a registered backend plugin.
|
||||
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(pCtx backend.PluginContext, reqCtx *models.ReqContext, path string)
|
||||
// Get plugin by its ID.
|
||||
Get(pluginID string) (Plugin, bool)
|
||||
}
|
||||
|
||||
// Plugin is the backend plugin interface.
|
||||
type Plugin interface {
|
||||
PluginID() string
|
||||
|
||||
@ -2,10 +2,8 @@
|
||||
package instrumentation
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
)
|
||||
|
||||
@ -68,19 +66,3 @@ func InstrumentCallResourceRequest(pluginID string, fn func() error) error {
|
||||
func InstrumentQueryDataRequest(pluginID string, fn func() error) error {
|
||||
return instrumentPluginRequest(pluginID, "queryData", fn)
|
||||
}
|
||||
|
||||
// InstrumentQueryDataHandler wraps a backend.QueryDataHandler with instrumentation of success rate and latency.
|
||||
func InstrumentQueryDataHandler(handler backend.QueryDataHandler) backend.QueryDataHandler {
|
||||
if handler == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
return backend.QueryDataHandlerFunc(func(ctx context.Context, req *backend.QueryDataRequest) (*backend.QueryDataResponse, error) {
|
||||
var resp *backend.QueryDataResponse
|
||||
err := InstrumentQueryDataRequest(req.PluginContext.PluginID, func() (innerErr error) {
|
||||
resp, innerErr = handler.QueryData(ctx, req)
|
||||
return
|
||||
})
|
||||
return resp, err
|
||||
})
|
||||
}
|
||||
|
||||
@ -1,531 +0,0 @@
|
||||
package manager
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-aws-sdk/pkg/awsds"
|
||||
"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/backendplugin"
|
||||
"github.com/grafana/grafana/pkg/plugins/backendplugin/instrumentation"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/grafana/grafana/pkg/util/errutil"
|
||||
"github.com/grafana/grafana/pkg/util/proxyutil"
|
||||
)
|
||||
|
||||
func ProvideService(cfg *setting.Cfg, licensing models.Licensing,
|
||||
pluginRequestValidator models.PluginRequestValidator) *Manager {
|
||||
s := &Manager{
|
||||
Cfg: cfg,
|
||||
License: licensing,
|
||||
PluginRequestValidator: pluginRequestValidator,
|
||||
logger: log.New("plugins.backend"),
|
||||
plugins: map[string]backendplugin.Plugin{},
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
type Manager struct {
|
||||
Cfg *setting.Cfg
|
||||
License models.Licensing
|
||||
PluginRequestValidator models.PluginRequestValidator
|
||||
pluginsMu sync.RWMutex
|
||||
plugins map[string]backendplugin.Plugin
|
||||
logger log.Logger
|
||||
}
|
||||
|
||||
func (m *Manager) Run(ctx context.Context) error {
|
||||
<-ctx.Done()
|
||||
m.stop(ctx)
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
// Register registers a backend plugin
|
||||
func (m *Manager) Register(pluginID string, factory backendplugin.PluginFactoryFunc) error {
|
||||
m.logger.Debug("Registering backend plugin", "pluginId", pluginID)
|
||||
m.pluginsMu.Lock()
|
||||
defer m.pluginsMu.Unlock()
|
||||
|
||||
if _, exists := m.plugins[pluginID]; exists {
|
||||
return fmt.Errorf("backend plugin %s already registered", pluginID)
|
||||
}
|
||||
|
||||
hostEnv := []string{
|
||||
fmt.Sprintf("GF_VERSION=%s", m.Cfg.BuildVersion),
|
||||
fmt.Sprintf("GF_EDITION=%s", m.License.Edition()),
|
||||
}
|
||||
|
||||
if m.License.HasLicense() {
|
||||
hostEnv = append(
|
||||
hostEnv,
|
||||
fmt.Sprintf("GF_ENTERPRISE_LICENSE_PATH=%s", m.Cfg.EnterpriseLicensePath),
|
||||
)
|
||||
|
||||
if envProvider, ok := m.License.(models.LicenseEnvironment); ok {
|
||||
for k, v := range envProvider.Environment() {
|
||||
hostEnv = append(hostEnv, fmt.Sprintf("%s=%s", k, v))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
hostEnv = append(hostEnv, m.getAWSEnvironmentVariables()...)
|
||||
hostEnv = append(hostEnv, m.getAzureEnvironmentVariables()...)
|
||||
|
||||
pluginSettings := getPluginSettings(pluginID, m.Cfg)
|
||||
env := pluginSettings.ToEnv("GF_PLUGIN", hostEnv)
|
||||
|
||||
pluginLogger := m.logger.New("pluginId", pluginID)
|
||||
plugin, err := factory(pluginID, pluginLogger, env)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
m.plugins[pluginID] = plugin
|
||||
m.logger.Debug("Backend plugin registered", "pluginId", pluginID)
|
||||
return nil
|
||||
}
|
||||
|
||||
// RegisterAndStart registers and starts a backend plugin
|
||||
func (m *Manager) RegisterAndStart(ctx context.Context, pluginID string, factory backendplugin.PluginFactoryFunc) error {
|
||||
err := m.Register(pluginID, factory)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
p, exists := m.Get(pluginID)
|
||||
if !exists {
|
||||
return fmt.Errorf("backend plugin %s is not registered", pluginID)
|
||||
}
|
||||
|
||||
m.start(ctx, p)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// UnregisterAndStop unregisters and stops a backend plugin
|
||||
func (m *Manager) UnregisterAndStop(ctx context.Context, pluginID string) error {
|
||||
m.logger.Debug("Unregistering backend plugin", "pluginId", pluginID)
|
||||
m.pluginsMu.Lock()
|
||||
defer m.pluginsMu.Unlock()
|
||||
|
||||
p, exists := m.plugins[pluginID]
|
||||
if !exists {
|
||||
return fmt.Errorf("backend plugin %s is not registered", pluginID)
|
||||
}
|
||||
|
||||
m.logger.Debug("Stopping backend plugin process", "pluginId", pluginID)
|
||||
if err := p.Decommission(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := p.Stop(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
delete(m.plugins, pluginID)
|
||||
|
||||
m.logger.Debug("Backend plugin unregistered", "pluginId", pluginID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) IsRegistered(pluginID string) bool {
|
||||
p, _ := m.Get(pluginID)
|
||||
|
||||
return p != nil && !p.IsDecommissioned()
|
||||
}
|
||||
|
||||
func (m *Manager) Get(pluginID string) (backendplugin.Plugin, bool) {
|
||||
m.pluginsMu.RLock()
|
||||
p, ok := m.plugins[pluginID]
|
||||
m.pluginsMu.RUnlock()
|
||||
|
||||
if ok && p.IsDecommissioned() {
|
||||
return nil, false
|
||||
}
|
||||
|
||||
return p, ok
|
||||
}
|
||||
|
||||
func (m *Manager) getAWSEnvironmentVariables() []string {
|
||||
variables := []string{}
|
||||
if m.Cfg.AWSAssumeRoleEnabled {
|
||||
variables = append(variables, awsds.AssumeRoleEnabledEnvVarKeyName+"=true")
|
||||
}
|
||||
if len(m.Cfg.AWSAllowedAuthProviders) > 0 {
|
||||
variables = append(variables, awsds.AllowedAuthProvidersEnvVarKeyName+"="+strings.Join(m.Cfg.AWSAllowedAuthProviders, ","))
|
||||
}
|
||||
|
||||
return variables
|
||||
}
|
||||
|
||||
func (m *Manager) getAzureEnvironmentVariables() []string {
|
||||
variables := []string{}
|
||||
if m.Cfg.Azure.Cloud != "" {
|
||||
variables = append(variables, "AZURE_CLOUD="+m.Cfg.Azure.Cloud)
|
||||
}
|
||||
if m.Cfg.Azure.ManagedIdentityClientId != "" {
|
||||
variables = append(variables, "AZURE_MANAGED_IDENTITY_CLIENT_ID="+m.Cfg.Azure.ManagedIdentityClientId)
|
||||
}
|
||||
if m.Cfg.Azure.ManagedIdentityEnabled {
|
||||
variables = append(variables, "AZURE_MANAGED_IDENTITY_ENABLED=true")
|
||||
}
|
||||
|
||||
return variables
|
||||
}
|
||||
|
||||
// start starts a managed backend plugin
|
||||
func (m *Manager) start(ctx context.Context, p backendplugin.Plugin) {
|
||||
if !p.IsManaged() {
|
||||
return
|
||||
}
|
||||
|
||||
if err := startPluginAndRestartKilledProcesses(ctx, p); err != nil {
|
||||
p.Logger().Error("Failed to start plugin", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
// StartPlugin starts a non-managed backend plugin
|
||||
func (m *Manager) StartPlugin(ctx context.Context, pluginID string) error {
|
||||
m.pluginsMu.RLock()
|
||||
p, registered := m.plugins[pluginID]
|
||||
m.pluginsMu.RUnlock()
|
||||
if !registered {
|
||||
return backendplugin.ErrPluginNotRegistered
|
||||
}
|
||||
|
||||
if p.IsManaged() {
|
||||
return errors.New("backend plugin is managed and cannot be manually started")
|
||||
}
|
||||
|
||||
return startPluginAndRestartKilledProcesses(ctx, p)
|
||||
}
|
||||
|
||||
// stop stops all managed backend plugins
|
||||
func (m *Manager) stop(ctx context.Context) {
|
||||
m.pluginsMu.RLock()
|
||||
defer m.pluginsMu.RUnlock()
|
||||
var wg sync.WaitGroup
|
||||
for _, p := range m.plugins {
|
||||
wg.Add(1)
|
||||
go func(p backendplugin.Plugin, ctx context.Context) {
|
||||
defer wg.Done()
|
||||
p.Logger().Debug("Stopping plugin")
|
||||
if err := p.Stop(ctx); err != nil {
|
||||
p.Logger().Error("Failed to stop plugin", "error", err)
|
||||
}
|
||||
p.Logger().Debug("Plugin stopped")
|
||||
}(p, ctx)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// CollectMetrics collects metrics from a registered backend plugin.
|
||||
func (m *Manager) CollectMetrics(ctx context.Context, pluginID string) (*backend.CollectMetricsResult, error) {
|
||||
p, registered := m.Get(pluginID)
|
||||
if !registered {
|
||||
return nil, backendplugin.ErrPluginNotRegistered
|
||||
}
|
||||
|
||||
var resp *backend.CollectMetricsResult
|
||||
err := instrumentation.InstrumentCollectMetrics(p.PluginID(), func() (innerErr error) {
|
||||
resp, innerErr = p.CollectMetrics(ctx)
|
||||
return
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return resp, nil
|
||||
}
|
||||
|
||||
// CheckHealth checks the health of a registered backend plugin.
|
||||
func (m *Manager) CheckHealth(ctx context.Context, pluginContext backend.PluginContext) (*backend.CheckHealthResult, error) {
|
||||
var dsURL string
|
||||
if pluginContext.DataSourceInstanceSettings != nil {
|
||||
dsURL = pluginContext.DataSourceInstanceSettings.URL
|
||||
}
|
||||
|
||||
err := m.PluginRequestValidator.Validate(dsURL, nil)
|
||||
if err != nil {
|
||||
return &backend.CheckHealthResult{
|
||||
Status: http.StatusForbidden,
|
||||
Message: "Access denied",
|
||||
}, nil
|
||||
}
|
||||
|
||||
p, registered := m.Get(pluginContext.PluginID)
|
||||
if !registered {
|
||||
return nil, backendplugin.ErrPluginNotRegistered
|
||||
}
|
||||
|
||||
var resp *backend.CheckHealthResult
|
||||
err = instrumentation.InstrumentCheckHealthRequest(p.PluginID(), func() (innerErr error) {
|
||||
resp, innerErr = p.CheckHealth(ctx, &backend.CheckHealthRequest{PluginContext: pluginContext})
|
||||
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 check plugin health", backendplugin.ErrHealthCheckFailed)
|
||||
}
|
||||
|
||||
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 {
|
||||
p, registered := m.Get(pCtx.PluginID)
|
||||
if !registered {
|
||||
return backendplugin.ErrPluginNotRegistered
|
||||
}
|
||||
|
||||
keepCookieModel := keepCookiesJSONModel{}
|
||||
if dis := pCtx.DataSourceInstanceSettings; dis != nil {
|
||||
err := json.Unmarshal(dis.JSONData, &keepCookieModel)
|
||||
if err != nil {
|
||||
p.Logger().Error("Failed to to unpack JSONData in datasource instance settings", "error", err)
|
||||
}
|
||||
}
|
||||
|
||||
proxyutil.ClearCookieHeader(req, keepCookieModel.KeepCookies)
|
||||
proxyutil.PrepareProxyRequest(req)
|
||||
|
||||
body, err := ioutil.ReadAll(req.Body)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to read request body: %w", err)
|
||||
}
|
||||
|
||||
crReq := &backend.CallResourceRequest{
|
||||
PluginContext: pCtx,
|
||||
Path: req.URL.Path,
|
||||
Method: req.Method,
|
||||
URL: req.URL.String(),
|
||||
Headers: req.Header,
|
||||
Body: body,
|
||||
}
|
||||
|
||||
return instrumentation.InstrumentCallResourceRequest(p.PluginID(), func() error {
|
||||
childCtx, cancel := context.WithCancel(req.Context())
|
||||
defer cancel()
|
||||
stream := newCallResourceResponseStream(childCtx)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
|
||||
defer func() {
|
||||
if err := stream.Close(); err != nil {
|
||||
m.logger.Warn("Failed to close stream", "err", err)
|
||||
}
|
||||
wg.Wait()
|
||||
}()
|
||||
|
||||
var flushStreamErr error
|
||||
go func() {
|
||||
flushStreamErr = flushStream(p, stream, w)
|
||||
wg.Done()
|
||||
}()
|
||||
|
||||
if err := p.CallResource(req.Context(), crReq, stream); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return flushStreamErr
|
||||
})
|
||||
}
|
||||
|
||||
// CallResource calls a plugin resource.
|
||||
func (m *Manager) CallResource(pCtx backend.PluginContext, reqCtx *models.ReqContext, path string) {
|
||||
var dsURL string
|
||||
if pCtx.DataSourceInstanceSettings != nil {
|
||||
dsURL = pCtx.DataSourceInstanceSettings.URL
|
||||
}
|
||||
|
||||
err := m.PluginRequestValidator.Validate(dsURL, reqCtx.Req)
|
||||
if err != nil {
|
||||
reqCtx.JsonApiErr(http.StatusForbidden, "Access denied", err)
|
||||
return
|
||||
}
|
||||
|
||||
clonedReq := reqCtx.Req.Clone(reqCtx.Req.Context())
|
||||
rawURL := path
|
||||
if clonedReq.URL.RawQuery != "" {
|
||||
rawURL += "?" + clonedReq.URL.RawQuery
|
||||
}
|
||||
urlPath, err := url.Parse(rawURL)
|
||||
if err != nil {
|
||||
handleCallResourceError(err, reqCtx)
|
||||
return
|
||||
}
|
||||
clonedReq.URL = urlPath
|
||||
err = m.callResourceInternal(reqCtx.Resp, clonedReq, pCtx)
|
||||
if err != nil {
|
||||
handleCallResourceError(err, reqCtx)
|
||||
}
|
||||
}
|
||||
|
||||
func handleCallResourceError(err error, reqCtx *models.ReqContext) {
|
||||
if errors.Is(err, backendplugin.ErrPluginUnavailable) {
|
||||
reqCtx.JsonApiErr(503, "Plugin unavailable", err)
|
||||
return
|
||||
}
|
||||
|
||||
if errors.Is(err, backendplugin.ErrMethodNotImplemented) {
|
||||
reqCtx.JsonApiErr(404, "Not found", err)
|
||||
return
|
||||
}
|
||||
|
||||
reqCtx.JsonApiErr(500, "Failed to call resource", err)
|
||||
}
|
||||
|
||||
func flushStream(plugin backendplugin.Plugin, stream callResourceClientResponseStream, w http.ResponseWriter) error {
|
||||
processedStreams := 0
|
||||
|
||||
for {
|
||||
resp, err := stream.Recv()
|
||||
if errors.Is(err, io.EOF) {
|
||||
if processedStreams == 0 {
|
||||
return errors.New("received empty resource response")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
if processedStreams == 0 {
|
||||
return errutil.Wrap("failed to receive response from resource call", err)
|
||||
}
|
||||
|
||||
plugin.Logger().Error("Failed to receive response from resource call", "error", err)
|
||||
return stream.Close()
|
||||
}
|
||||
|
||||
// Expected that headers and status are only part of first stream
|
||||
if processedStreams == 0 && resp.Headers != nil {
|
||||
// Make sure a content type always is returned in response
|
||||
if _, exists := resp.Headers["Content-Type"]; !exists {
|
||||
resp.Headers["Content-Type"] = []string{"application/json"}
|
||||
}
|
||||
|
||||
for k, values := range resp.Headers {
|
||||
// Due to security reasons we don't want to forward
|
||||
// cookies from a backend plugin to clients/browsers.
|
||||
if k == "Set-Cookie" {
|
||||
continue
|
||||
}
|
||||
|
||||
for _, v := range values {
|
||||
// TODO: Figure out if we should use Set here instead
|
||||
// nolint:gocritic
|
||||
w.Header().Add(k, v)
|
||||
}
|
||||
}
|
||||
|
||||
w.WriteHeader(resp.Status)
|
||||
}
|
||||
|
||||
if _, err := w.Write(resp.Body); err != nil {
|
||||
plugin.Logger().Error("Failed to write resource response", "error", err)
|
||||
}
|
||||
|
||||
if flusher, ok := w.(http.Flusher); ok {
|
||||
flusher.Flush()
|
||||
}
|
||||
processedStreams++
|
||||
}
|
||||
}
|
||||
|
||||
func startPluginAndRestartKilledProcesses(ctx context.Context, p backendplugin.Plugin) error {
|
||||
if err := p.Start(ctx); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
go func(ctx context.Context, p backendplugin.Plugin) {
|
||||
if err := restartKilledProcess(ctx, p); err != nil {
|
||||
p.Logger().Error("Attempt to restart killed plugin process failed", "error", err)
|
||||
}
|
||||
}(ctx, p)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func restartKilledProcess(ctx context.Context, p backendplugin.Plugin) error {
|
||||
ticker := time.NewTicker(time.Second * 1)
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
if err := ctx.Err(); err != nil && !errors.Is(err, context.Canceled) {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
case <-ticker.C:
|
||||
if p.IsDecommissioned() {
|
||||
p.Logger().Debug("Plugin decommissioned")
|
||||
return nil
|
||||
}
|
||||
|
||||
if !p.Exited() {
|
||||
continue
|
||||
}
|
||||
|
||||
p.Logger().Debug("Restarting plugin")
|
||||
if err := p.Start(ctx); err != nil {
|
||||
p.Logger().Error("Failed to restart plugin", "error", err)
|
||||
continue
|
||||
}
|
||||
p.Logger().Debug("Plugin restarted")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// callResourceClientResponseStream is used for receiving resource call responses.
|
||||
type callResourceClientResponseStream interface {
|
||||
Recv() (*backend.CallResourceResponse, error)
|
||||
Close() error
|
||||
}
|
||||
@ -1,511 +0,0 @@
|
||||
package manager
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana-aws-sdk/pkg/awsds"
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
"github.com/grafana/grafana/pkg/infra/log"
|
||||
"github.com/grafana/grafana/pkg/plugins/backendplugin"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
const testPluginID = "test-plugin"
|
||||
|
||||
func TestManager(t *testing.T) {
|
||||
newManagerScenario(t, false, func(t *testing.T, ctx *managerScenarioCtx) {
|
||||
t.Run("Unregistered plugin scenario", func(t *testing.T) {
|
||||
err := ctx.manager.StartPlugin(context.Background(), testPluginID)
|
||||
require.Equal(t, backendplugin.ErrPluginNotRegistered, err)
|
||||
|
||||
_, err = ctx.manager.CollectMetrics(context.Background(), testPluginID)
|
||||
require.Equal(t, backendplugin.ErrPluginNotRegistered, err)
|
||||
|
||||
_, err = ctx.manager.CheckHealth(context.Background(), backend.PluginContext{PluginID: testPluginID})
|
||||
require.Equal(t, backendplugin.ErrPluginNotRegistered, err)
|
||||
|
||||
req, err := http.NewRequest(http.MethodGet, "/test", nil)
|
||||
require.NoError(t, err)
|
||||
w := httptest.NewRecorder()
|
||||
err = ctx.manager.callResourceInternal(w, req, backend.PluginContext{PluginID: testPluginID})
|
||||
require.Equal(t, backendplugin.ErrPluginNotRegistered, err)
|
||||
})
|
||||
})
|
||||
|
||||
newManagerScenario(t, true, func(t *testing.T, ctx *managerScenarioCtx) {
|
||||
t.Run("Managed plugin scenario", func(t *testing.T) {
|
||||
ctx.license.edition = "Open Source"
|
||||
ctx.license.hasLicense = false
|
||||
ctx.cfg.BuildVersion = "7.0.0"
|
||||
|
||||
t.Run("Should be able to register plugin", func(t *testing.T) {
|
||||
err := ctx.manager.RegisterAndStart(context.Background(), testPluginID, ctx.factory)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, ctx.plugin)
|
||||
require.Equal(t, testPluginID, ctx.plugin.pluginID)
|
||||
require.NotNil(t, ctx.plugin.logger)
|
||||
require.Equal(t, 1, ctx.plugin.startCount)
|
||||
require.True(t, ctx.manager.IsRegistered(testPluginID))
|
||||
|
||||
t.Run("Should not be able to register an already registered plugin", func(t *testing.T) {
|
||||
err := ctx.manager.RegisterAndStart(context.Background(), testPluginID, ctx.factory)
|
||||
require.Equal(t, 1, ctx.plugin.startCount)
|
||||
require.Error(t, err)
|
||||
})
|
||||
|
||||
t.Run("Should provide expected host environment variables", func(t *testing.T) {
|
||||
require.Len(t, ctx.env, 7)
|
||||
require.EqualValues(t, []string{
|
||||
"GF_VERSION=7.0.0",
|
||||
"GF_EDITION=Open Source",
|
||||
fmt.Sprintf("%s=true", awsds.AssumeRoleEnabledEnvVarKeyName),
|
||||
fmt.Sprintf("%s=keys,credentials", awsds.AllowedAuthProvidersEnvVarKeyName),
|
||||
"AZURE_CLOUD=AzureCloud",
|
||||
"AZURE_MANAGED_IDENTITY_CLIENT_ID=client-id",
|
||||
"AZURE_MANAGED_IDENTITY_ENABLED=true"},
|
||||
ctx.env)
|
||||
})
|
||||
|
||||
t.Run("When manager runs should start and stop plugin", func(t *testing.T) {
|
||||
pCtx := context.Background()
|
||||
cCtx, cancel := context.WithCancel(pCtx)
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
var runErr error
|
||||
go func() {
|
||||
runErr = ctx.manager.Run(cCtx)
|
||||
wg.Done()
|
||||
}()
|
||||
time.Sleep(time.Millisecond)
|
||||
cancel()
|
||||
wg.Wait()
|
||||
require.Equal(t, context.Canceled, runErr)
|
||||
require.Equal(t, 1, ctx.plugin.startCount)
|
||||
require.Equal(t, 1, ctx.plugin.stopCount)
|
||||
})
|
||||
|
||||
t.Run("When manager runs should restart plugin process when killed", func(t *testing.T) {
|
||||
ctx.plugin.stopCount = 0
|
||||
ctx.plugin.startCount = 0
|
||||
pCtx := context.Background()
|
||||
cCtx, cancel := context.WithCancel(pCtx)
|
||||
var wgRun sync.WaitGroup
|
||||
wgRun.Add(1)
|
||||
var runErr error
|
||||
go func() {
|
||||
runErr = ctx.manager.Run(cCtx)
|
||||
wgRun.Done()
|
||||
}()
|
||||
|
||||
time.Sleep(time.Millisecond)
|
||||
|
||||
var wgKill sync.WaitGroup
|
||||
wgKill.Add(1)
|
||||
go func() {
|
||||
ctx.plugin.kill()
|
||||
for {
|
||||
if !ctx.plugin.Exited() {
|
||||
break
|
||||
}
|
||||
}
|
||||
cancel()
|
||||
wgKill.Done()
|
||||
}()
|
||||
wgKill.Wait()
|
||||
wgRun.Wait()
|
||||
require.Equal(t, context.Canceled, runErr)
|
||||
require.Equal(t, 1, ctx.plugin.stopCount)
|
||||
require.Equal(t, 1, ctx.plugin.startCount)
|
||||
})
|
||||
|
||||
t.Run("Shouldn't be able to start managed plugin", func(t *testing.T) {
|
||||
err := ctx.manager.StartPlugin(context.Background(), testPluginID)
|
||||
require.NotNil(t, err)
|
||||
})
|
||||
|
||||
t.Run("Unimplemented handlers", func(t *testing.T) {
|
||||
t.Run("Collect metrics should return method not implemented error", func(t *testing.T) {
|
||||
_, err = ctx.manager.CollectMetrics(context.Background(), testPluginID)
|
||||
require.Equal(t, backendplugin.ErrMethodNotImplemented, err)
|
||||
})
|
||||
|
||||
t.Run("Check health should return method not implemented error", func(t *testing.T) {
|
||||
_, err = ctx.manager.CheckHealth(context.Background(), backend.PluginContext{PluginID: testPluginID})
|
||||
require.Equal(t, backendplugin.ErrMethodNotImplemented, err)
|
||||
})
|
||||
|
||||
t.Run("Call resource should return method not implemented error", func(t *testing.T) {
|
||||
req, err := http.NewRequest(http.MethodGet, "/test", bytes.NewReader([]byte{}))
|
||||
require.NoError(t, err)
|
||||
w := httptest.NewRecorder()
|
||||
err = ctx.manager.callResourceInternal(w, req, backend.PluginContext{PluginID: testPluginID})
|
||||
require.Equal(t, backendplugin.ErrMethodNotImplemented, err)
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("Implemented handlers", func(t *testing.T) {
|
||||
t.Run("Collect metrics should return expected result", func(t *testing.T) {
|
||||
ctx.plugin.CollectMetricsHandlerFunc = func(ctx context.Context) (*backend.CollectMetricsResult, error) {
|
||||
return &backend.CollectMetricsResult{
|
||||
PrometheusMetrics: []byte("hello"),
|
||||
}, nil
|
||||
}
|
||||
|
||||
res, err := ctx.manager.CollectMetrics(context.Background(), testPluginID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
require.Equal(t, "hello", string(res.PrometheusMetrics))
|
||||
})
|
||||
|
||||
t.Run("Check health should return expected result", func(t *testing.T) {
|
||||
json := []byte(`{
|
||||
"key": "value"
|
||||
}`)
|
||||
ctx.plugin.CheckHealthHandlerFunc = func(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
|
||||
return &backend.CheckHealthResult{
|
||||
Status: backend.HealthStatusOk,
|
||||
Message: "All good",
|
||||
JSONDetails: json,
|
||||
}, nil
|
||||
}
|
||||
|
||||
res, err := ctx.manager.CheckHealth(context.Background(), backend.PluginContext{PluginID: testPluginID})
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, res)
|
||||
require.Equal(t, backend.HealthStatusOk, res.Status)
|
||||
require.Equal(t, "All good", res.Message)
|
||||
require.Equal(t, json, res.JSONDetails)
|
||||
})
|
||||
|
||||
t.Run("Call resource should return expected response", func(t *testing.T) {
|
||||
ctx.plugin.CallResourceHandlerFunc = func(ctx context.Context,
|
||||
req *backend.CallResourceRequest, sender backend.CallResourceResponseSender) error {
|
||||
return sender.Send(&backend.CallResourceResponse{
|
||||
Status: http.StatusOK,
|
||||
})
|
||||
}
|
||||
|
||||
req, err := http.NewRequest(http.MethodGet, "/test", bytes.NewReader([]byte{}))
|
||||
require.NoError(t, err)
|
||||
w := httptest.NewRecorder()
|
||||
err = ctx.manager.callResourceInternal(w, req, backend.PluginContext{PluginID: testPluginID})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, http.StatusOK, w.Code)
|
||||
})
|
||||
})
|
||||
|
||||
t.Run("Should be able to decommission a running plugin", func(t *testing.T) {
|
||||
require.True(t, ctx.manager.IsRegistered(testPluginID))
|
||||
|
||||
err := ctx.manager.UnregisterAndStop(context.Background(), testPluginID)
|
||||
require.NoError(t, err)
|
||||
|
||||
require.Equal(t, 2, ctx.plugin.stopCount)
|
||||
require.False(t, ctx.manager.IsRegistered(testPluginID))
|
||||
p := ctx.manager.plugins[testPluginID]
|
||||
require.Nil(t, p)
|
||||
|
||||
err = ctx.manager.StartPlugin(context.Background(), testPluginID)
|
||||
require.Equal(t, backendplugin.ErrPluginNotRegistered, err)
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
newManagerScenario(t, false, func(t *testing.T, ctx *managerScenarioCtx) {
|
||||
t.Run("Unmanaged plugin scenario", func(t *testing.T) {
|
||||
ctx.license.edition = "Open Source"
|
||||
ctx.license.hasLicense = false
|
||||
ctx.cfg.BuildVersion = "7.0.0"
|
||||
|
||||
t.Run("Should be able to register plugin", func(t *testing.T) {
|
||||
err := ctx.manager.RegisterAndStart(context.Background(), testPluginID, ctx.factory)
|
||||
require.NoError(t, err)
|
||||
require.True(t, ctx.manager.IsRegistered(testPluginID))
|
||||
require.False(t, ctx.plugin.managed)
|
||||
|
||||
t.Run("When manager runs should not start plugin", func(t *testing.T) {
|
||||
pCtx := context.Background()
|
||||
cCtx, cancel := context.WithCancel(pCtx)
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
var runErr error
|
||||
go func() {
|
||||
runErr = ctx.manager.Run(cCtx)
|
||||
wg.Done()
|
||||
}()
|
||||
go func() {
|
||||
cancel()
|
||||
}()
|
||||
wg.Wait()
|
||||
require.Equal(t, context.Canceled, runErr)
|
||||
require.Equal(t, 0, ctx.plugin.startCount)
|
||||
require.Equal(t, 1, ctx.plugin.stopCount)
|
||||
})
|
||||
|
||||
t.Run("Should be able to start unmanaged plugin and be restarted when process is killed", func(t *testing.T) {
|
||||
pCtx := context.Background()
|
||||
cCtx, cancel := context.WithCancel(pCtx)
|
||||
defer cancel()
|
||||
err := ctx.manager.StartPlugin(cCtx, testPluginID)
|
||||
require.Nil(t, err)
|
||||
require.Equal(t, 1, ctx.plugin.startCount)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
ctx.plugin.kill()
|
||||
for {
|
||||
if !ctx.plugin.Exited() {
|
||||
break
|
||||
}
|
||||
}
|
||||
wg.Done()
|
||||
}()
|
||||
wg.Wait()
|
||||
require.Equal(t, 2, ctx.plugin.startCount)
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
newManagerScenario(t, true, func(t *testing.T, ctx *managerScenarioCtx) {
|
||||
t.Run("Plugin registration scenario when Grafana is licensed", func(t *testing.T) {
|
||||
ctx.license.edition = "Enterprise"
|
||||
ctx.license.hasLicense = true
|
||||
ctx.license.tokenRaw = "testtoken"
|
||||
ctx.cfg.BuildVersion = "7.0.0"
|
||||
ctx.cfg.EnterpriseLicensePath = "/license.txt"
|
||||
|
||||
err := ctx.manager.RegisterAndStart(context.Background(), testPluginID, ctx.factory)
|
||||
require.NoError(t, err)
|
||||
|
||||
t.Run("Should provide expected host environment variables", func(t *testing.T) {
|
||||
require.Len(t, ctx.env, 9)
|
||||
require.EqualValues(t, []string{
|
||||
"GF_VERSION=7.0.0",
|
||||
"GF_EDITION=Enterprise",
|
||||
"GF_ENTERPRISE_LICENSE_PATH=/license.txt",
|
||||
"GF_ENTERPRISE_LICENSE_TEXT=testtoken",
|
||||
fmt.Sprintf("%s=true", awsds.AssumeRoleEnabledEnvVarKeyName),
|
||||
fmt.Sprintf("%s=keys,credentials", awsds.AllowedAuthProvidersEnvVarKeyName),
|
||||
"AZURE_CLOUD=AzureCloud",
|
||||
"AZURE_MANAGED_IDENTITY_CLIENT_ID=client-id",
|
||||
"AZURE_MANAGED_IDENTITY_ENABLED=true"},
|
||||
ctx.env)
|
||||
})
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
type managerScenarioCtx struct {
|
||||
cfg *setting.Cfg
|
||||
license *testLicensingService
|
||||
manager *Manager
|
||||
factory backendplugin.PluginFactoryFunc
|
||||
plugin *testPlugin
|
||||
env []string
|
||||
}
|
||||
|
||||
func newManagerScenario(t *testing.T, managed bool, fn func(t *testing.T, ctx *managerScenarioCtx)) {
|
||||
t.Helper()
|
||||
cfg := setting.NewCfg()
|
||||
cfg.AWSAllowedAuthProviders = []string{"keys", "credentials"}
|
||||
cfg.AWSAssumeRoleEnabled = true
|
||||
|
||||
cfg.Azure.ManagedIdentityEnabled = true
|
||||
cfg.Azure.Cloud = "AzureCloud"
|
||||
cfg.Azure.ManagedIdentityClientId = "client-id"
|
||||
|
||||
license := &testLicensingService{}
|
||||
validator := &testPluginRequestValidator{}
|
||||
ctx := &managerScenarioCtx{
|
||||
cfg: cfg,
|
||||
license: license,
|
||||
manager: &Manager{
|
||||
Cfg: cfg,
|
||||
License: license,
|
||||
PluginRequestValidator: validator,
|
||||
logger: log.New("test"),
|
||||
plugins: map[string]backendplugin.Plugin{},
|
||||
},
|
||||
}
|
||||
|
||||
ctx.factory = func(pluginID string, logger log.Logger, env []string) (backendplugin.Plugin, error) {
|
||||
ctx.plugin = &testPlugin{
|
||||
pluginID: pluginID,
|
||||
logger: logger,
|
||||
managed: managed,
|
||||
}
|
||||
ctx.env = env
|
||||
|
||||
return ctx.plugin, nil
|
||||
}
|
||||
|
||||
fn(t, ctx)
|
||||
}
|
||||
|
||||
type testPlugin struct {
|
||||
pluginID string
|
||||
logger log.Logger
|
||||
startCount int
|
||||
stopCount int
|
||||
managed bool
|
||||
exited bool
|
||||
decommissioned bool
|
||||
backend.CollectMetricsHandlerFunc
|
||||
backend.CheckHealthHandlerFunc
|
||||
backend.QueryDataHandlerFunc
|
||||
backend.CallResourceHandlerFunc
|
||||
mutex sync.RWMutex
|
||||
}
|
||||
|
||||
func (tp *testPlugin) PluginID() string {
|
||||
return tp.pluginID
|
||||
}
|
||||
|
||||
func (tp *testPlugin) Logger() log.Logger {
|
||||
return tp.logger
|
||||
}
|
||||
|
||||
func (tp *testPlugin) Start(ctx context.Context) error {
|
||||
tp.mutex.Lock()
|
||||
defer tp.mutex.Unlock()
|
||||
tp.exited = false
|
||||
tp.startCount++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (tp *testPlugin) Stop(ctx context.Context) error {
|
||||
tp.mutex.Lock()
|
||||
defer tp.mutex.Unlock()
|
||||
tp.stopCount++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (tp *testPlugin) IsManaged() bool {
|
||||
return tp.managed
|
||||
}
|
||||
|
||||
func (tp *testPlugin) Exited() bool {
|
||||
tp.mutex.RLock()
|
||||
defer tp.mutex.RUnlock()
|
||||
return tp.exited
|
||||
}
|
||||
|
||||
func (tp *testPlugin) Decommission() error {
|
||||
tp.mutex.Lock()
|
||||
defer tp.mutex.Unlock()
|
||||
|
||||
tp.decommissioned = true
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (tp *testPlugin) IsDecommissioned() bool {
|
||||
tp.mutex.RLock()
|
||||
defer tp.mutex.RUnlock()
|
||||
return tp.decommissioned
|
||||
}
|
||||
|
||||
func (tp *testPlugin) kill() {
|
||||
tp.mutex.Lock()
|
||||
defer tp.mutex.Unlock()
|
||||
tp.exited = true
|
||||
}
|
||||
|
||||
func (tp *testPlugin) CollectMetrics(ctx context.Context) (*backend.CollectMetricsResult, error) {
|
||||
if tp.CollectMetricsHandlerFunc != nil {
|
||||
return tp.CollectMetricsHandlerFunc(ctx)
|
||||
}
|
||||
|
||||
return nil, backendplugin.ErrMethodNotImplemented
|
||||
}
|
||||
|
||||
func (tp *testPlugin) CheckHealth(ctx context.Context, req *backend.CheckHealthRequest) (*backend.CheckHealthResult, error) {
|
||||
if tp.CheckHealthHandlerFunc != nil {
|
||||
return tp.CheckHealthHandlerFunc(ctx, req)
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
return backendplugin.ErrMethodNotImplemented
|
||||
}
|
||||
|
||||
func (tp *testPlugin) SubscribeStream(ctx context.Context, request *backend.SubscribeStreamRequest) (*backend.SubscribeStreamResponse, error) {
|
||||
return nil, backendplugin.ErrMethodNotImplemented
|
||||
}
|
||||
|
||||
func (tp *testPlugin) PublishStream(ctx context.Context, request *backend.PublishStreamRequest) (*backend.PublishStreamResponse, error) {
|
||||
return nil, backendplugin.ErrMethodNotImplemented
|
||||
}
|
||||
|
||||
func (tp *testPlugin) RunStream(ctx context.Context, request *backend.RunStreamRequest, sender *backend.StreamSender) error {
|
||||
return backendplugin.ErrMethodNotImplemented
|
||||
}
|
||||
|
||||
type testLicensingService struct {
|
||||
edition string
|
||||
hasLicense bool
|
||||
tokenRaw string
|
||||
}
|
||||
|
||||
func (t *testLicensingService) HasLicense() bool {
|
||||
return t.hasLicense
|
||||
}
|
||||
|
||||
func (t *testLicensingService) Expiry() int64 {
|
||||
return 0
|
||||
}
|
||||
|
||||
func (t *testLicensingService) Edition() string {
|
||||
return t.edition
|
||||
}
|
||||
|
||||
func (t *testLicensingService) StateInfo() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (t *testLicensingService) ContentDeliveryPrefix() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (t *testLicensingService) LicenseURL(showAdminLicensingPage bool) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
func (t *testLicensingService) HasValidLicense() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (t *testLicensingService) Environment() map[string]string {
|
||||
return map[string]string{"GF_ENTERPRISE_LICENSE_TEXT": t.tokenRaw}
|
||||
}
|
||||
|
||||
type testPluginRequestValidator struct{}
|
||||
|
||||
func (t *testPluginRequestValidator) Validate(string, *http.Request) error {
|
||||
return nil
|
||||
}
|
||||
@ -1,40 +0,0 @@
|
||||
package manager
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
)
|
||||
|
||||
type pluginSettings map[string]string
|
||||
|
||||
func (ps pluginSettings) ToEnv(prefix string, hostEnv []string) []string {
|
||||
var env []string
|
||||
for k, v := range ps {
|
||||
key := fmt.Sprintf("%s_%s", prefix, strings.ToUpper(k))
|
||||
if value := os.Getenv(key); value != "" {
|
||||
v = value
|
||||
}
|
||||
|
||||
env = append(env, fmt.Sprintf("%s=%s", key, v))
|
||||
}
|
||||
|
||||
env = append(env, hostEnv...)
|
||||
|
||||
return env
|
||||
}
|
||||
|
||||
func getPluginSettings(plugID string, cfg *setting.Cfg) pluginSettings {
|
||||
ps := pluginSettings{}
|
||||
for k, v := range cfg.PluginSettings[plugID] {
|
||||
if k == "path" || strings.ToLower(k) == "id" {
|
||||
continue
|
||||
}
|
||||
|
||||
ps[k] = v
|
||||
}
|
||||
|
||||
return ps
|
||||
}
|
||||
@ -1,96 +0,0 @@
|
||||
package manager
|
||||
|
||||
import (
|
||||
"os"
|
||||
"sort"
|
||||
"testing"
|
||||
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestPluginSettings(t *testing.T) {
|
||||
t.Run("Should only extract from sections beginning with 'plugin.' in config", func(t *testing.T) {
|
||||
cfg := &setting.Cfg{
|
||||
PluginSettings: setting.PluginSettings{
|
||||
"plugin": map[string]string{
|
||||
"key1": "value1",
|
||||
"key2": "value2",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
ps := getPluginSettings("plugin", cfg)
|
||||
require.Len(t, ps, 2)
|
||||
|
||||
t.Run("Should skip path setting", func(t *testing.T) {
|
||||
cfg.PluginSettings["plugin"]["path"] = "value"
|
||||
ps := getPluginSettings("plugin", cfg)
|
||||
require.Len(t, ps, 2)
|
||||
})
|
||||
|
||||
t.Run("Should skip id setting", func(t *testing.T) {
|
||||
cfg.PluginSettings["plugin"]["id"] = "value"
|
||||
ps := getPluginSettings("plugin", cfg)
|
||||
require.Len(t, ps, 2)
|
||||
})
|
||||
|
||||
t.Run("Should return expected environment variables from plugin settings ", func(t *testing.T) {
|
||||
ps := getPluginSettings("plugin", cfg)
|
||||
env := ps.ToEnv("GF_PLUGIN", []string{"GF_VERSION=6.7.0"})
|
||||
sort.Strings(env)
|
||||
require.Len(t, env, 3)
|
||||
require.EqualValues(t, []string{"GF_PLUGIN_KEY1=value1", "GF_PLUGIN_KEY2=value2", "GF_VERSION=6.7.0"}, env)
|
||||
})
|
||||
|
||||
t.Run("Should override config variable with environment variable ", func(t *testing.T) {
|
||||
_ = os.Setenv("GF_PLUGIN_KEY1", "sth")
|
||||
t.Cleanup(func() {
|
||||
_ = os.Unsetenv("GF_PLUGIN_KEY1")
|
||||
})
|
||||
|
||||
ps := getPluginSettings("plugin", cfg)
|
||||
env := ps.ToEnv("GF_PLUGIN", []string{"GF_VERSION=6.7.0"})
|
||||
sort.Strings(env)
|
||||
require.Len(t, env, 3)
|
||||
require.EqualValues(t, []string{"GF_PLUGIN_KEY1=sth", "GF_PLUGIN_KEY2=value2", "GF_VERSION=6.7.0"}, env)
|
||||
})
|
||||
|
||||
t.Run("Config variable doesn't match env variable ", func(t *testing.T) {
|
||||
_ = os.Setenv("GF_PLUGIN_KEY3", "value3")
|
||||
t.Cleanup(func() {
|
||||
_ = os.Unsetenv("GF_PLUGIN_KEY3")
|
||||
})
|
||||
|
||||
ps := getPluginSettings("plugin", cfg)
|
||||
env := ps.ToEnv("GF_PLUGIN", []string{"GF_VERSION=6.7.0"})
|
||||
sort.Strings(env)
|
||||
require.Len(t, env, 3)
|
||||
require.EqualValues(t, []string{"GF_PLUGIN_KEY1=value1", "GF_PLUGIN_KEY2=value2", "GF_VERSION=6.7.0"}, env)
|
||||
})
|
||||
|
||||
t.Run("Should override missing config variable with environment variable ", func(t *testing.T) {
|
||||
cfg := &setting.Cfg{
|
||||
PluginSettings: setting.PluginSettings{
|
||||
"plugin": map[string]string{
|
||||
"key1": "value1",
|
||||
"key2": "",
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
ps := getPluginSettings("plugin", cfg)
|
||||
require.Len(t, ps, 2)
|
||||
|
||||
_ = os.Setenv("GF_PLUGIN_KEY2", "sth")
|
||||
t.Cleanup(func() {
|
||||
_ = os.Unsetenv("GF_PLUGIN_KEY1")
|
||||
})
|
||||
|
||||
env := ps.ToEnv("GF_PLUGIN", []string{"GF_VERSION=6.7.0"})
|
||||
sort.Strings(env)
|
||||
require.Len(t, env, 3)
|
||||
require.EqualValues(t, []string{"GF_PLUGIN_KEY1=value1", "GF_PLUGIN_KEY2=sth", "GF_VERSION=6.7.0"}, env)
|
||||
})
|
||||
})
|
||||
}
|
||||
@ -1,57 +0,0 @@
|
||||
package manager
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
|
||||
"github.com/grafana/grafana-plugin-sdk-go/backend"
|
||||
)
|
||||
|
||||
func newCallResourceResponseStream(ctx context.Context) *callResourceResponseStream {
|
||||
return &callResourceResponseStream{
|
||||
ctx: ctx,
|
||||
stream: make(chan *backend.CallResourceResponse),
|
||||
}
|
||||
}
|
||||
|
||||
type callResourceResponseStream struct {
|
||||
ctx context.Context
|
||||
stream chan *backend.CallResourceResponse
|
||||
closed bool
|
||||
}
|
||||
|
||||
func (s *callResourceResponseStream) Send(res *backend.CallResourceResponse) error {
|
||||
if s.closed {
|
||||
return errors.New("cannot send to a closed stream")
|
||||
}
|
||||
|
||||
select {
|
||||
case <-s.ctx.Done():
|
||||
return errors.New("cancelled")
|
||||
case s.stream <- res:
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
func (s *callResourceResponseStream) Recv() (*backend.CallResourceResponse, error) {
|
||||
select {
|
||||
case <-s.ctx.Done():
|
||||
return nil, s.ctx.Err()
|
||||
case res, ok := <-s.stream:
|
||||
if !ok {
|
||||
return nil, io.EOF
|
||||
}
|
||||
return res, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (s *callResourceResponseStream) Close() error {
|
||||
if s.closed {
|
||||
return errors.New("cannot close a closed stream")
|
||||
}
|
||||
|
||||
close(s.stream)
|
||||
s.closed = true
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user