Plugins: Remove registry dependency from process manager (#73241)

simplify
pull/73323/head
Will Browne 2 years ago committed by GitHub
parent d935e6ff57
commit 75b0788377
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 81
      pkg/plugins/manager/fakes/fakes.go
  2. 2
      pkg/plugins/manager/loader/loader_test.go
  3. 8
      pkg/plugins/manager/pipeline/initialization/steps.go
  4. 12
      pkg/plugins/manager/pipeline/termination/steps.go
  5. 12
      pkg/plugins/manager/process/ifaces.go
  6. 70
      pkg/plugins/manager/process/process.go
  7. 142
      pkg/plugins/manager/process/process_test.go
  8. 24
      pkg/plugins/manager/store/store.go
  9. 33
      pkg/plugins/manager/store/store_test.go
  10. 14
      pkg/plugins/plugins.go
  11. 6
      pkg/registry/backgroundsvcs/background_services.go
  12. 4
      pkg/services/pluginsintegration/loader/loader_test.go
  13. 4
      pkg/services/pluginsintegration/pipeline/pipeline.go
  14. 2
      pkg/services/pluginsintegration/pluginsintegration.go
  15. 6
      pkg/services/pluginsintegration/test_helper.go
  16. 5
      pkg/services/secrets/kvstore/test_helpers.go

@ -248,8 +248,8 @@ func (s *FakePluginStorage) Extract(ctx context.Context, pluginID string, dirNam
} }
type FakeProcessManager struct { type FakeProcessManager struct {
StartFunc func(_ context.Context, pluginID string) error StartFunc func(_ context.Context, p *plugins.Plugin) error
StopFunc func(_ context.Context, pluginID string) error StopFunc func(_ context.Context, p *plugins.Plugin) error
Started map[string]int Started map[string]int
Stopped map[string]int Stopped map[string]int
} }
@ -261,18 +261,18 @@ func NewFakeProcessManager() *FakeProcessManager {
} }
} }
func (m *FakeProcessManager) Start(ctx context.Context, pluginID string) error { func (m *FakeProcessManager) Start(ctx context.Context, p *plugins.Plugin) error {
m.Started[pluginID]++ m.Started[p.ID]++
if m.StartFunc != nil { if m.StartFunc != nil {
return m.StartFunc(ctx, pluginID) return m.StartFunc(ctx, p)
} }
return nil return nil
} }
func (m *FakeProcessManager) Stop(ctx context.Context, pluginID string) error { func (m *FakeProcessManager) Stop(ctx context.Context, p *plugins.Plugin) error {
m.Stopped[pluginID]++ m.Stopped[p.ID]++
if m.StopFunc != nil { if m.StopFunc != nil {
return m.StopFunc(ctx, pluginID) return m.StopFunc(ctx, p)
} }
return nil return nil
} }
@ -518,3 +518,68 @@ func (f *FakeTerminator) Terminate(ctx context.Context, pluginID string) error {
} }
return nil return nil
} }
type FakeBackendPlugin struct {
Managed bool
StartCount int
StopCount int
Decommissioned bool
Running bool
mutex sync.RWMutex
backendplugin.Plugin
}
func NewFakeBackendPlugin(managed bool) *FakeBackendPlugin {
return &FakeBackendPlugin{
Managed: managed,
}
}
func (p *FakeBackendPlugin) Start(_ context.Context) error {
p.mutex.Lock()
defer p.mutex.Unlock()
p.Running = true
p.StartCount++
return nil
}
func (p *FakeBackendPlugin) Stop(_ context.Context) error {
p.mutex.Lock()
defer p.mutex.Unlock()
p.Running = false
p.StopCount++
return nil
}
func (p *FakeBackendPlugin) Decommission() error {
p.mutex.Lock()
defer p.mutex.Unlock()
p.Decommissioned = true
return nil
}
func (p *FakeBackendPlugin) IsDecommissioned() bool {
p.mutex.RLock()
defer p.mutex.RUnlock()
return p.Decommissioned
}
func (p *FakeBackendPlugin) IsManaged() bool {
p.mutex.RLock()
defer p.mutex.RUnlock()
return p.Managed
}
func (p *FakeBackendPlugin) Exited() bool {
p.mutex.RLock()
defer p.mutex.RUnlock()
return !p.Running
}
func (p *FakeBackendPlugin) Kill() {
p.mutex.Lock()
defer p.mutex.Unlock()
p.Running = false
}

@ -23,7 +23,7 @@ import (
"github.com/grafana/grafana/pkg/services/org" "github.com/grafana/grafana/pkg/services/org"
) )
var compareOpts = []cmp.Option{cmpopts.IgnoreFields(plugins.Plugin{}, "client", "log"), fsComparer} var compareOpts = []cmp.Option{cmpopts.IgnoreFields(plugins.Plugin{}, "client", "log", "mu"), fsComparer}
var fsComparer = cmp.Comparer(func(fs1 plugins.FS, fs2 plugins.FS) bool { var fsComparer = cmp.Comparer(func(fs1 plugins.FS, fs2 plugins.FS) bool {
fs1Files, err := fs1.Files() fs1Files, err := fs1.Files()

@ -57,16 +57,16 @@ func (b *BackendClientInit) Initialize(ctx context.Context, p *plugins.Plugin) (
// BackendClientStarter implements an InitializeFunc for starting a backend plugin process. // BackendClientStarter implements an InitializeFunc for starting a backend plugin process.
type BackendClientStarter struct { type BackendClientStarter struct {
processManager process.Service processManager process.Manager
log log.Logger log log.Logger
} }
// BackendProcessStartStep returns a new InitializeFunc for starting a backend plugin process. // BackendProcessStartStep returns a new InitializeFunc for starting a backend plugin process.
func BackendProcessStartStep(processManager process.Service) InitializeFunc { func BackendProcessStartStep(processManager process.Manager) InitializeFunc {
return newBackendProcessStarter(processManager).Start return newBackendProcessStarter(processManager).Start
} }
func newBackendProcessStarter(processManager process.Service) *BackendClientStarter { func newBackendProcessStarter(processManager process.Manager) *BackendClientStarter {
return &BackendClientStarter{ return &BackendClientStarter{
processManager: processManager, processManager: processManager,
log: log.New("plugins.backend.start"), log: log.New("plugins.backend.start"),
@ -75,7 +75,7 @@ func newBackendProcessStarter(processManager process.Service) *BackendClientStar
// Start will start the backend plugin process. // Start will start the backend plugin process.
func (b *BackendClientStarter) Start(ctx context.Context, p *plugins.Plugin) (*plugins.Plugin, error) { func (b *BackendClientStarter) Start(ctx context.Context, p *plugins.Plugin) (*plugins.Plugin, error) {
if err := b.processManager.Start(ctx, p.ID); err != nil { if err := b.processManager.Start(ctx, p); err != nil {
b.log.Error("Could not start plugin", "pluginId", p.ID, "error", err) b.log.Error("Could not start plugin", "pluginId", p.ID, "error", err)
return nil, err return nil, err
} }

@ -44,18 +44,18 @@ func (r *TerminablePluginResolver) Resolve(ctx context.Context, pluginID string)
// BackendProcessTerminator implements a TerminateFunc for stopping a backend plugin process. // BackendProcessTerminator implements a TerminateFunc for stopping a backend plugin process.
// //
// It uses the process.Service to stop the backend plugin process. // It uses the process.Manager to stop the backend plugin process.
type BackendProcessTerminator struct { type BackendProcessTerminator struct {
processManager process.Service processManager process.Manager
log log.Logger log log.Logger
} }
// BackendProcessTerminatorStep returns a new TerminateFunc for stopping a backend plugin process. // BackendProcessTerminatorStep returns a new TerminateFunc for stopping a backend plugin process.
func BackendProcessTerminatorStep(processManager process.Service) TerminateFunc { func BackendProcessTerminatorStep(processManager process.Manager) TerminateFunc {
return newBackendProcessTerminator(processManager).Terminate return newBackendProcessTerminator(processManager).Terminate
} }
func newBackendProcessTerminator(processManager process.Service) *BackendProcessTerminator { func newBackendProcessTerminator(processManager process.Manager) *BackendProcessTerminator {
return &BackendProcessTerminator{ return &BackendProcessTerminator{
processManager: processManager, processManager: processManager,
log: log.New("plugins.backend.termination"), log: log.New("plugins.backend.termination"),
@ -64,9 +64,7 @@ func newBackendProcessTerminator(processManager process.Service) *BackendProcess
// Terminate stops a backend plugin process. // Terminate stops a backend plugin process.
func (t *BackendProcessTerminator) Terminate(ctx context.Context, p *plugins.Plugin) error { func (t *BackendProcessTerminator) Terminate(ctx context.Context, p *plugins.Plugin) error {
t.log.Debug("Stopping plugin process", "pluginId", p.ID) return t.processManager.Stop(ctx, p)
return t.processManager.Stop(ctx, p.ID)
} }
// Deregister implements a TerminateFunc for removing a plugin from the plugin registry. // Deregister implements a TerminateFunc for removing a plugin from the plugin registry.

@ -1,10 +1,14 @@
package process package process
import "context" import (
"context"
type Service interface { "github.com/grafana/grafana/pkg/plugins"
)
type Manager interface {
// Start executes a backend plugin process. // Start executes a backend plugin process.
Start(ctx context.Context, pluginID string) error Start(ctx context.Context, p *plugins.Plugin) error
// Stop terminates a backend plugin process. // Stop terminates a backend plugin process.
Stop(ctx context.Context, pluginID string) error Stop(ctx context.Context, p *plugins.Plugin) error
} }

@ -3,54 +3,22 @@ package process
import ( import (
"context" "context"
"errors" "errors"
"sync"
"time" "time"
"github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/backendplugin"
"github.com/grafana/grafana/pkg/plugins/log"
"github.com/grafana/grafana/pkg/plugins/manager/registry"
) )
var _ Service = (*Manager)(nil) type Service struct{}
type Manager struct { func ProvideService() *Service {
pluginRegistry registry.Service return &Service{}
mu sync.Mutex
log log.Logger
}
func ProvideService(pluginRegistry registry.Service) *Manager {
return NewManager(pluginRegistry)
} }
func NewManager(pluginRegistry registry.Service) *Manager { func (*Service) Start(ctx context.Context, p *plugins.Plugin) error {
return &Manager{
pluginRegistry: pluginRegistry,
log: log.New("plugin.process.manager"),
}
}
func (m *Manager) Run(ctx context.Context) error {
<-ctx.Done()
m.shutdown(ctx)
return ctx.Err()
}
func (m *Manager) Start(ctx context.Context, pluginID string) error {
p, exists := m.pluginRegistry.Plugin(ctx, pluginID)
if !exists {
return backendplugin.ErrPluginNotRegistered
}
if !p.IsManaged() || !p.Backend || p.SignatureError != nil { if !p.IsManaged() || !p.Backend || p.SignatureError != nil {
return nil return nil
} }
m.mu.Lock()
defer m.mu.Unlock()
if err := startPluginAndRestartKilledProcesses(ctx, p); err != nil { if err := startPluginAndRestartKilledProcesses(ctx, p); err != nil {
return err return err
} }
@ -59,15 +27,8 @@ func (m *Manager) Start(ctx context.Context, pluginID string) error {
return nil return nil
} }
func (m *Manager) Stop(ctx context.Context, pluginID string) error { func (*Service) Stop(ctx context.Context, p *plugins.Plugin) error {
p, exists := m.pluginRegistry.Plugin(ctx, pluginID) p.Logger().Debug("Stopping plugin process")
if !exists {
return backendplugin.ErrPluginNotRegistered
}
m.log.Debug("Stopping plugin process", "pluginId", p.ID)
m.mu.Lock()
defer m.mu.Unlock()
if err := p.Decommission(); err != nil { if err := p.Decommission(); err != nil {
return err return err
} }
@ -79,23 +40,6 @@ func (m *Manager) Stop(ctx context.Context, pluginID string) error {
return nil return nil
} }
// shutdown stops all backend plugin processes
func (m *Manager) shutdown(ctx context.Context) {
var wg sync.WaitGroup
for _, p := range m.pluginRegistry.Plugins(ctx) {
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()
}
func startPluginAndRestartKilledProcesses(ctx context.Context, p *plugins.Plugin) error { func startPluginAndRestartKilledProcesses(ctx context.Context, p *plugins.Plugin) error {
if err := p.Start(ctx); err != nil { if err := p.Start(ctx); err != nil {
return err return err
@ -123,7 +67,7 @@ func restartKilledProcess(ctx context.Context, p *plugins.Plugin) error {
if err := ctx.Err(); err != nil && !errors.Is(err, context.Canceled) { if err := ctx.Err(); err != nil && !errors.Is(err, context.Canceled) {
return err return err
} }
return nil return p.Stop(ctx)
case <-ticker.C: case <-ticker.C:
if p.IsDecommissioned() { if p.IsDecommissioned() {
p.Logger().Debug("Plugin decommissioned") p.Logger().Debug("Plugin decommissioned")

@ -14,12 +14,6 @@ import (
) )
func TestProcessManager_Start(t *testing.T) { func TestProcessManager_Start(t *testing.T) {
t.Run("Plugin not found in registry", func(t *testing.T) {
m := NewManager(fakes.NewFakePluginRegistry())
err := m.Start(context.Background(), "non-existing-datasource")
require.ErrorIs(t, err, backendplugin.ErrPluginNotRegistered)
})
t.Run("Plugin state determines process start", func(t *testing.T) { t.Run("Plugin state determines process start", func(t *testing.T) {
tcs := []struct { tcs := []struct {
name string name string
@ -58,21 +52,16 @@ func TestProcessManager_Start(t *testing.T) {
} }
for _, tc := range tcs { for _, tc := range tcs {
t.Run(tc.name, func(t *testing.T) { t.Run(tc.name, func(t *testing.T) {
bp := newFakeBackendPlugin(tc.managed) bp := fakes.NewFakeBackendPlugin(tc.managed)
p := createPlugin(t, bp, func(plugin *plugins.Plugin) { p := createPlugin(t, bp, func(plugin *plugins.Plugin) {
plugin.Backend = tc.backend plugin.Backend = tc.backend
plugin.SignatureError = tc.signatureError plugin.SignatureError = tc.signatureError
}) })
m := NewManager(&fakes.FakePluginRegistry{ m := &Service{}
Store: map[string]*plugins.Plugin{ err := m.Start(context.Background(), p)
p.ID: p,
}},
)
err := m.Start(context.Background(), p.ID)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, tc.expectedStartCount, bp.startCount) require.Equal(t, tc.expectedStartCount, bp.StartCount)
if tc.expectedStartCount > 0 { if tc.expectedStartCount > 0 {
require.True(t, !p.Exited()) require.True(t, !p.Exited())
@ -85,67 +74,42 @@ func TestProcessManager_Start(t *testing.T) {
} }
func TestProcessManager_Stop(t *testing.T) { func TestProcessManager_Stop(t *testing.T) {
t.Run("Plugin not found in registry", func(t *testing.T) {
m := NewManager(fakes.NewFakePluginRegistry())
err := m.Stop(context.Background(), "non-existing-datasource")
require.ErrorIs(t, err, backendplugin.ErrPluginNotRegistered)
})
t.Run("Can stop a running plugin", func(t *testing.T) { t.Run("Can stop a running plugin", func(t *testing.T) {
pluginID := "test-datasource" pluginID := "test-datasource"
bp := newFakeBackendPlugin(true) bp := fakes.NewFakeBackendPlugin(true)
p := createPlugin(t, bp, func(plugin *plugins.Plugin) { p := createPlugin(t, bp, func(plugin *plugins.Plugin) {
plugin.ID = pluginID plugin.ID = pluginID
plugin.Backend = true plugin.Backend = true
}) })
m := NewManager(&fakes.FakePluginRegistry{ m := &Service{}
Store: map[string]*plugins.Plugin{ err := m.Stop(context.Background(), p)
pluginID: p,
}},
)
err := m.Stop(context.Background(), pluginID)
require.NoError(t, err) require.NoError(t, err)
require.True(t, p.IsDecommissioned()) require.True(t, p.IsDecommissioned())
require.True(t, bp.decommissioned)
require.True(t, p.Exited()) require.True(t, p.Exited())
require.Equal(t, 1, bp.stopCount) require.Equal(t, 1, bp.StopCount)
}) })
} }
func TestProcessManager_ManagedBackendPluginLifecycle(t *testing.T) { func TestProcessManager_ManagedBackendPluginLifecycle(t *testing.T) {
bp := newFakeBackendPlugin(true) bp := fakes.NewFakeBackendPlugin(true)
p := createPlugin(t, bp, func(plugin *plugins.Plugin) { p := createPlugin(t, bp, func(plugin *plugins.Plugin) {
plugin.Backend = true plugin.Backend = true
}) })
m := NewManager(&fakes.FakePluginRegistry{ m := &Service{}
Store: map[string]*plugins.Plugin{
p.ID: p,
}},
)
err := m.Start(context.Background(), p.ID) err := m.Start(context.Background(), p)
require.NoError(t, err) require.NoError(t, err)
require.Equal(t, 1, bp.startCount) require.Equal(t, 1, bp.StartCount)
t.Run("When plugin process is killed, the process is restarted", func(t *testing.T) { t.Run("When plugin process is killed, the process is restarted", func(t *testing.T) {
pCtx := context.Background()
cCtx, cancel := context.WithCancel(pCtx)
var wgRun sync.WaitGroup
wgRun.Add(1)
var runErr error
go func() {
runErr = m.Run(cCtx)
wgRun.Done()
}()
var wgKill sync.WaitGroup var wgKill sync.WaitGroup
wgKill.Add(1) wgKill.Add(1)
go func() { go func() {
bp.kill() // manually kill process bp.Kill() // manually kill process
for { for {
if !bp.Exited() { if !bp.Exited() {
break break
@ -155,85 +119,15 @@ func TestProcessManager_ManagedBackendPluginLifecycle(t *testing.T) {
}() }()
wgKill.Wait() wgKill.Wait()
require.True(t, !p.Exited()) require.True(t, !p.Exited())
require.Equal(t, 2, bp.startCount) require.Equal(t, 2, bp.StartCount)
require.Equal(t, 0, bp.stopCount) require.Equal(t, 0, bp.StopCount)
t.Run("When context is cancelled the plugin is stopped", func(t *testing.T) { t.Cleanup(func() {
cancel() require.NoError(t, m.Stop(context.Background(), p))
wgRun.Wait()
require.ErrorIs(t, runErr, context.Canceled)
require.True(t, p.Exited())
require.Equal(t, 2, bp.startCount)
require.Equal(t, 1, bp.stopCount)
}) })
}) })
} }
type fakeBackendPlugin struct {
managed bool
startCount int
stopCount int
decommissioned bool
running bool
mutex sync.RWMutex
backendplugin.Plugin
}
func newFakeBackendPlugin(managed bool) *fakeBackendPlugin {
return &fakeBackendPlugin{
managed: managed,
}
}
func (p *fakeBackendPlugin) Start(_ context.Context) error {
p.mutex.Lock()
defer p.mutex.Unlock()
p.running = true
p.startCount++
return nil
}
func (p *fakeBackendPlugin) Stop(_ context.Context) error {
p.mutex.Lock()
defer p.mutex.Unlock()
p.running = false
p.stopCount++
return nil
}
func (p *fakeBackendPlugin) Decommission() error {
p.mutex.Lock()
defer p.mutex.Unlock()
p.decommissioned = true
return nil
}
func (p *fakeBackendPlugin) IsDecommissioned() bool {
p.mutex.RLock()
defer p.mutex.RUnlock()
return p.decommissioned
}
func (p *fakeBackendPlugin) IsManaged() bool {
p.mutex.RLock()
defer p.mutex.RUnlock()
return p.managed
}
func (p *fakeBackendPlugin) Exited() bool {
p.mutex.RLock()
defer p.mutex.RUnlock()
return !p.running
}
func (p *fakeBackendPlugin) kill() {
p.mutex.Lock()
defer p.mutex.Unlock()
p.running = false
}
func createPlugin(t *testing.T, bp backendplugin.Plugin, cbs ...func(p *plugins.Plugin)) *plugins.Plugin { func createPlugin(t *testing.T, bp backendplugin.Plugin, cbs ...func(p *plugins.Plugin)) *plugins.Plugin {
t.Helper() t.Helper()

@ -3,8 +3,10 @@ package store
import ( import (
"context" "context"
"sort" "sort"
"sync"
"github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/backendplugin"
"github.com/grafana/grafana/pkg/plugins/manager/loader" "github.com/grafana/grafana/pkg/plugins/manager/loader"
"github.com/grafana/grafana/pkg/plugins/manager/registry" "github.com/grafana/grafana/pkg/plugins/manager/registry"
"github.com/grafana/grafana/pkg/plugins/manager/sources" "github.com/grafana/grafana/pkg/plugins/manager/sources"
@ -27,6 +29,12 @@ func ProvideService(pluginRegistry registry.Service, pluginSources sources.Regis
return New(pluginRegistry), nil return New(pluginRegistry), nil
} }
func (s *Service) Run(ctx context.Context) error {
<-ctx.Done()
s.shutdown(ctx)
return ctx.Err()
}
func New(pluginRegistry registry.Service) *Service { func New(pluginRegistry registry.Service) *Service {
return &Service{ return &Service{
pluginRegistry: pluginRegistry, pluginRegistry: pluginRegistry,
@ -120,3 +128,19 @@ func (s *Service) Routes(ctx context.Context) []*plugins.StaticRoute {
} }
return staticRoutes return staticRoutes
} }
func (s *Service) shutdown(ctx context.Context) {
var wg sync.WaitGroup
for _, p := range s.pluginRegistry.Plugins(ctx) {
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()
}

@ -2,12 +2,14 @@ package store
import ( import (
"context" "context"
"sync"
"testing" "testing"
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
"github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/backendplugin" "github.com/grafana/grafana/pkg/plugins/backendplugin"
"github.com/grafana/grafana/pkg/plugins/log"
"github.com/grafana/grafana/pkg/plugins/manager/fakes" "github.com/grafana/grafana/pkg/plugins/manager/fakes"
) )
@ -177,6 +179,37 @@ func TestStore_SecretsManager(t *testing.T) {
}) })
} }
func TestProcessManager_shutdown(t *testing.T) {
p := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-datasource", Type: plugins.TypeDataSource}} // Backend: true
backend := &fakes.FakeBackendPlugin{}
p.RegisterClient(backend)
p.SetLogger(log.NewTestLogger())
ps := New(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p.ID: p,
},
})
pCtx := context.Background()
cCtx, cancel := context.WithCancel(pCtx)
var wgRun sync.WaitGroup
wgRun.Add(1)
var runErr error
go func() {
runErr = ps.Run(cCtx)
wgRun.Done()
}()
t.Run("When context is cancelled the plugin is stopped", func(t *testing.T) {
cancel()
wgRun.Wait()
require.ErrorIs(t, runErr, context.Canceled)
require.True(t, p.Exited())
require.Equal(t, 1, backend.StopCount)
})
}
func TestStore_availablePlugins(t *testing.T) { func TestStore_availablePlugins(t *testing.T) {
t.Run("Decommissioned plugins are excluded from availablePlugins", func(t *testing.T) { t.Run("Decommissioned plugins are excluded from availablePlugins", func(t *testing.T) {
p1 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-datasource"}} p1 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-datasource"}}

@ -10,6 +10,7 @@ import (
"path" "path"
"runtime" "runtime"
"strings" "strings"
"sync"
"github.com/grafana/grafana-plugin-sdk-go/backend" "github.com/grafana/grafana-plugin-sdk-go/backend"
@ -62,6 +63,8 @@ type Plugin struct {
client backendplugin.Plugin client backendplugin.Plugin
log log.Logger log log.Logger
mu sync.Mutex
// This will be moved to plugin.json when we have general support in gcom // This will be moved to plugin.json when we have general support in gcom
Alias string `json:"alias,omitempty"` Alias string `json:"alias,omitempty"`
} }
@ -264,16 +267,24 @@ func (p *Plugin) SetLogger(l log.Logger) {
} }
func (p *Plugin) Start(ctx context.Context) error { func (p *Plugin) Start(ctx context.Context) error {
p.mu.Lock()
defer p.mu.Unlock()
if p.client == nil { if p.client == nil {
return fmt.Errorf("could not start plugin %s as no plugin client exists", p.ID) return fmt.Errorf("could not start plugin %s as no plugin client exists", p.ID)
} }
return p.client.Start(ctx) return p.client.Start(ctx)
} }
func (p *Plugin) Stop(ctx context.Context) error { func (p *Plugin) Stop(ctx context.Context) error {
p.mu.Lock()
defer p.mu.Unlock()
if p.client == nil { if p.client == nil {
return nil return nil
} }
return p.client.Stop(ctx) return p.client.Stop(ctx)
} }
@ -285,6 +296,9 @@ func (p *Plugin) IsManaged() bool {
} }
func (p *Plugin) Decommission() error { func (p *Plugin) Decommission() error {
p.mu.Lock()
defer p.mu.Unlock()
if p.client != nil { if p.client != nil {
return p.client.Decommission() return p.client.Decommission()
} }

@ -7,7 +7,7 @@ import (
"github.com/grafana/grafana/pkg/infra/tracing" "github.com/grafana/grafana/pkg/infra/tracing"
uss "github.com/grafana/grafana/pkg/infra/usagestats/service" uss "github.com/grafana/grafana/pkg/infra/usagestats/service"
"github.com/grafana/grafana/pkg/infra/usagestats/statscollector" "github.com/grafana/grafana/pkg/infra/usagestats/statscollector"
"github.com/grafana/grafana/pkg/plugins/manager/process" pluginStore "github.com/grafana/grafana/pkg/plugins/manager/store"
"github.com/grafana/grafana/pkg/registry" "github.com/grafana/grafana/pkg/registry"
"github.com/grafana/grafana/pkg/services/alerting" "github.com/grafana/grafana/pkg/services/alerting"
"github.com/grafana/grafana/pkg/services/auth" "github.com/grafana/grafana/pkg/services/auth"
@ -42,7 +42,7 @@ import (
func ProvideBackgroundServiceRegistry( func ProvideBackgroundServiceRegistry(
httpServer *api.HTTPServer, ng *ngalert.AlertNG, cleanup *cleanup.CleanUpService, live *live.GrafanaLive, httpServer *api.HTTPServer, ng *ngalert.AlertNG, cleanup *cleanup.CleanUpService, live *live.GrafanaLive,
pushGateway *pushhttp.Gateway, notifications *notifications.NotificationService, processManager *process.Manager, pushGateway *pushhttp.Gateway, notifications *notifications.NotificationService, pluginStore *pluginStore.Service,
rendering *rendering.RenderingService, tokenService auth.UserTokenBackgroundService, tracing tracing.Tracer, rendering *rendering.RenderingService, tokenService auth.UserTokenBackgroundService, tracing tracing.Tracer,
provisioning *provisioning.ProvisioningServiceImpl, alerting *alerting.AlertEngine, usageStats *uss.UsageStats, provisioning *provisioning.ProvisioningServiceImpl, alerting *alerting.AlertEngine, usageStats *uss.UsageStats,
statsCollector *statscollector.Service, grafanaUpdateChecker *updatechecker.GrafanaService, statsCollector *statscollector.Service, grafanaUpdateChecker *updatechecker.GrafanaService,
@ -85,7 +85,7 @@ func ProvideBackgroundServiceRegistry(
grpcServerProvider, grpcServerProvider,
saService, saService,
authInfoService, authInfoService,
processManager, pluginStore,
secretMigrationProvider, secretMigrationProvider,
loginAttemptService, loginAttemptService,
bundleService, bundleService,

@ -30,7 +30,7 @@ import (
"github.com/grafana/grafana/pkg/setting" "github.com/grafana/grafana/pkg/setting"
) )
var compareOpts = []cmp.Option{cmpopts.IgnoreFields(plugins.Plugin{}, "client", "log"), fsComparer} var compareOpts = []cmp.Option{cmpopts.IgnoreFields(plugins.Plugin{}, "client", "log", "mu"), fsComparer}
var fsComparer = cmp.Comparer(func(fs1 plugins.FS, fs2 plugins.FS) bool { var fsComparer = cmp.Comparer(func(fs1 plugins.FS, fs2 plugins.FS) bool {
fs1Files, err := fs1.Files() fs1Files, err := fs1.Files()
@ -1311,7 +1311,7 @@ func TestLoader_Load_NestedPlugins(t *testing.T) {
}) })
} }
func newLoader(t *testing.T, cfg *config.Cfg, reg registry.Service, proc process.Service, func newLoader(t *testing.T, cfg *config.Cfg, reg registry.Service, proc process.Manager,
backendFactory plugins.BackendFactoryProvider, sigErrTracker pluginerrs.SignatureErrorTracker) *Loader { backendFactory plugins.BackendFactoryProvider, sigErrTracker pluginerrs.SignatureErrorTracker) *Loader {
assets := assetpath.ProvideService(pluginscdn.ProvideService(cfg)) assets := assetpath.ProvideService(pluginscdn.ProvideService(cfg))
lic := fakes.NewFakeLicensingService() lic := fakes.NewFakeLicensingService()

@ -53,7 +53,7 @@ func ProvideValidationStage(cfg *config.Cfg, sv signature.Validator, ai angulari
} }
func ProvideInitializationStage(cfg *config.Cfg, pr registry.Service, l plugins.Licensing, func ProvideInitializationStage(cfg *config.Cfg, pr registry.Service, l plugins.Licensing,
bp plugins.BackendFactoryProvider, pm process.Service, externalServiceRegistry oauth.ExternalServiceRegistry, bp plugins.BackendFactoryProvider, pm process.Manager, externalServiceRegistry oauth.ExternalServiceRegistry,
roleRegistry plugins.RoleRegistry) *initialization.Initialize { roleRegistry plugins.RoleRegistry) *initialization.Initialize {
return initialization.New(cfg, initialization.Opts{ return initialization.New(cfg, initialization.Opts{
InitializeFuncs: []initialization.InitializeFunc{ InitializeFuncs: []initialization.InitializeFunc{
@ -67,7 +67,7 @@ func ProvideInitializationStage(cfg *config.Cfg, pr registry.Service, l plugins.
}) })
} }
func ProvideTerminationStage(cfg *config.Cfg, pr registry.Service, pm process.Service) (*termination.Terminate, error) { func ProvideTerminationStage(cfg *config.Cfg, pr registry.Service, pm process.Manager) (*termination.Terminate, error) {
return termination.New(cfg, termination.Opts{ return termination.New(cfg, termination.Opts{
ResolveFunc: termination.TerminablePluginResolverStep(pr), ResolveFunc: termination.TerminablePluginResolverStep(pr),
TerminateFuncs: []termination.TerminateFunc{ TerminateFuncs: []termination.TerminateFunc{

@ -61,7 +61,7 @@ var WireSet = wire.NewSet(
ProvideClientDecorator, ProvideClientDecorator,
wire.Bind(new(plugins.Client), new(*client.Decorator)), wire.Bind(new(plugins.Client), new(*client.Decorator)),
process.ProvideService, process.ProvideService,
wire.Bind(new(process.Service), new(*process.Manager)), wire.Bind(new(process.Manager), new(*process.Service)),
coreplugin.ProvideCoreRegistry, coreplugin.ProvideCoreRegistry,
pluginscdn.ProvideService, pluginscdn.ProvideService,
assetpath.ProvideService, assetpath.ProvideService,

@ -48,7 +48,7 @@ func CreateIntegrationTestCtx(t *testing.T, cfg *setting.Cfg, coreRegistry *core
cdn := pluginscdn.ProvideService(pCfg) cdn := pluginscdn.ProvideService(pCfg)
reg := registry.ProvideService() reg := registry.ProvideService()
angularInspector := angularinspector.NewStaticInspector() angularInspector := angularinspector.NewStaticInspector()
proc := process.NewManager(reg) proc := process.ProvideService()
errTracker := pluginerrs.ProvideSignatureErrorTracker() errTracker := pluginerrs.ProvideSignatureErrorTracker()
disc := pipeline.ProvideDiscoveryStage(pCfg, finder.NewLocalFinder(true), reg) disc := pipeline.ProvideDiscoveryStage(pCfg, finder.NewLocalFinder(true), reg)
@ -100,13 +100,13 @@ func CreateTestLoader(t *testing.T, cfg *pluginsCfg.Cfg, opts LoaderOpts) *loade
if opts.Initializer == nil { if opts.Initializer == nil {
reg := registry.ProvideService() reg := registry.ProvideService()
coreRegistry := coreplugin.NewRegistry(make(map[string]backendplugin.PluginFactoryFunc)) coreRegistry := coreplugin.NewRegistry(make(map[string]backendplugin.PluginFactoryFunc))
opts.Initializer = pipeline.ProvideInitializationStage(cfg, reg, fakes.NewFakeLicensingService(), provider.ProvideService(coreRegistry), process.NewManager(reg), &fakes.FakeOauthService{}, fakes.NewFakeRoleRegistry()) opts.Initializer = pipeline.ProvideInitializationStage(cfg, reg, fakes.NewFakeLicensingService(), provider.ProvideService(coreRegistry), process.ProvideService(), &fakes.FakeOauthService{}, fakes.NewFakeRoleRegistry())
} }
if opts.Terminator == nil { if opts.Terminator == nil {
var err error var err error
reg := registry.ProvideService() reg := registry.ProvideService()
opts.Terminator, err = pipeline.ProvideTerminationStage(cfg, reg, process.NewManager(reg)) opts.Terminator, err = pipeline.ProvideTerminationStage(cfg, reg, process.ProvideService())
require.NoError(t, err) require.NoError(t, err)
} }

@ -16,6 +16,7 @@ import (
"github.com/grafana/grafana/pkg/plugins" "github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/backendplugin" "github.com/grafana/grafana/pkg/plugins/backendplugin"
"github.com/grafana/grafana/pkg/plugins/backendplugin/secretsmanagerplugin" "github.com/grafana/grafana/pkg/plugins/backendplugin/secretsmanagerplugin"
pluginsLogger "github.com/grafana/grafana/pkg/plugins/log"
"github.com/grafana/grafana/pkg/services/featuremgmt" "github.com/grafana/grafana/pkg/services/featuremgmt"
"github.com/grafana/grafana/pkg/services/secrets/fakes" "github.com/grafana/grafana/pkg/services/secrets/fakes"
secretsmng "github.com/grafana/grafana/pkg/services/secrets/manager" secretsmng "github.com/grafana/grafana/pkg/services/secrets/manager"
@ -257,6 +258,10 @@ func (pc *fakePluginClient) Stop(_ context.Context) error {
return nil return nil
} }
func (pc *fakePluginClient) Logger() pluginsLogger.Logger {
return pluginsLogger.NewTestLogger()
}
func SetupFatalCrashTest( func SetupFatalCrashTest(
t *testing.T, t *testing.T,
shouldFailOnStart bool, shouldFailOnStart bool,

Loading…
Cancel
Save