public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Revert "Plugins: Move store init to dskit service" (#111806)

Browse Source 
Revert "Plugins: Move store init to dskit service (#111206)"

This reverts commit 4cff7237d0.
This commit is contained in:
Will Browne
2025-09-30 12:48:27 +01:00
committed by GitHub
parent 1ebfc3fea2
commit cc72599321
9 changed files with 101 additions and 231 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/services/pluginsintegration/pluginstore/store.go
+29 -75
View File
@@ -3,21 +3,18 @@ package pluginstore
import (
"context"
"sort"
"sync"
"time"
"github.com/grafana/dskit/services"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/manager/loader"
"github.com/grafana/grafana/pkg/plugins/manager/registry"
"github.com/grafana/grafana/pkg/plugins/manager/sources"
"golang.org/x/sync/errgroup"
)
var _ Store = (*Service)(nil)
const ServiceName = "plugins.store"
// Store is the publicly accessible storage for plugins.
type Store interface {
// Plugin finds a plugin by its ID.
@@ -28,80 +25,47 @@ type Store interface {
}
type Service struct {
services.NamedService
pluginRegistry registry.Service
pluginLoader loader.Service
pluginSources sources.Registry
}
func ProvideService(pluginRegistry registry.Service, pluginSources sources.Registry,
pluginLoader loader.Service) *Service {
return New(pluginRegistry, pluginLoader, pluginSources)
}
func (s *Service) Run(ctx context.Context) error {
if err := s.StartAsync(ctx); err != nil {
return err
}
return s.AwaitTerminated(ctx)
}
func NewPluginStoreForTest(pluginRegistry registry.Service, pluginLoader loader.Service, pluginSources sources.Registry) (*Service, error) {
s := New(pluginRegistry, pluginLoader, pluginSources)
if err := s.StartAsync(context.Background()); err != nil {
return nil, err
}
if err := s.AwaitRunning(context.Background()); err != nil {
return nil, err
}
return s, nil
}
func New(pluginRegistry registry.Service, pluginLoader loader.Service, pluginSources sources.Registry) *Service {
s := &Service{
pluginRegistry: pluginRegistry,
pluginLoader: pluginLoader,
pluginSources: pluginSources,
}
s.NamedService = services.NewBasicService(s.starting, s.running, s.stopping).WithName(ServiceName)
return s
}
func (s *Service) starting(ctx context.Context) error {
pluginLoader loader.Service) (*Service, error) {
ctx := context.Background()
start := time.Now()
totalPlugins := 0
logger := log.New(ServiceName)
logger := log.New("plugin.store")
logger.Info("Loading plugins...")
for _, ps := range s.pluginSources.List(ctx) {
loadedPlugins, err := s.pluginLoader.Load(ctx, ps)
for _, ps := range pluginSources.List(ctx) {
loadedPlugins, err := pluginLoader.Load(ctx, ps)
if err != nil {
logger.Error("Loading plugin source failed", "source", ps.PluginClass(ctx), "error", err)
return err
return nil, err
}
totalPlugins += len(loadedPlugins)
}
logger.Info("Plugins loaded", "count", totalPlugins, "duration", time.Since(start))
return nil
return New(pluginRegistry, pluginLoader), nil
}
func (s *Service) running(ctx context.Context) error {
func (s *Service) Run(ctx context.Context) error {
<-ctx.Done()
return nil
s.shutdown(ctx)
return ctx.Err()
}
func (s *Service) stopping(failureReason error) error {
return s.shutdown(context.Background())
func New(pluginRegistry registry.Service, pluginLoader loader.Service) *Service {
return &Service{
pluginRegistry: pluginRegistry,
pluginLoader: pluginLoader,
}
}
func (s *Service) Plugin(ctx context.Context, pluginID string) (Plugin, bool) {
if err := s.AwaitRunning(ctx); err != nil {
log.New(ServiceName).FromContext(ctx).Error("Failed to get plugin", "error", err)
return Plugin{}, false
}
p, exists := s.plugin(ctx, pluginID)
if !exists {
return Plugin{}, false
@@ -111,10 +75,6 @@ func (s *Service) Plugin(ctx context.Context, pluginID string) (Plugin, bool) {
}
func (s *Service) Plugins(ctx context.Context, pluginTypes ...plugins.Type) []Plugin {
if err := s.AwaitRunning(ctx); err != nil {
log.New(ServiceName).FromContext(ctx).Error("Failed to get plugins", "error", err)
return []Plugin{}
}
// if no types passed, assume all
if len(pluginTypes) == 0 {
pluginTypes = plugins.PluginTypes
@@ -165,10 +125,6 @@ func (s *Service) availablePlugins(ctx context.Context) []*plugins.Plugin {
}
func (s *Service) Routes(ctx context.Context) []*plugins.StaticRoute {
if err := s.AwaitRunning(ctx); err != nil {
log.New(ServiceName).FromContext(ctx).Error("Failed to get routes", "error", err)
return []*plugins.StaticRoute{}
}
staticRoutes := make([]*plugins.StaticRoute, 0)
for _, p := range s.availablePlugins(ctx) {
@@ -179,20 +135,18 @@ func (s *Service) Routes(ctx context.Context) []*plugins.StaticRoute {
return staticRoutes
}
func (s *Service) shutdown(ctx context.Context) error {
var errgroup errgroup.Group
plugins := s.pluginRegistry.Plugins(ctx)
for _, p := range plugins {
plugin := p // capture loop variable
errgroup.Go(func() error {
plugin.Logger().Debug("Stopping plugin")
if _, err := s.pluginLoader.Unload(ctx, plugin); err != nil {
plugin.Logger().Error("Failed to stop plugin", "error", err)
return err
func (s *Service) shutdown(ctx context.Context) {
var wg sync.WaitGroup
for _, plugin := range s.pluginRegistry.Plugins(ctx) {
wg.Add(1)
go func(ctx context.Context, p *plugins.Plugin) {
defer wg.Done()
p.Logger().Debug("Stopping plugin")
if _, err := s.pluginLoader.Unload(ctx, p); err != nil {
p.Logger().Error("Failed to stop plugin", "error", err)
}
plugin.Logger().Debug("Plugin stopped")
return nil
})
p.Logger().Debug("Plugin stopped")
}(ctx, plugin)
}
return errgroup.Wait()
wg.Wait()
}
pkg/services/pluginsintegration/pluginstore/store_test.go
+39 -70
View File
@@ -2,7 +2,7 @@ package pluginstore
import (
"context"
"errors"
"sync"
"testing"
"github.com/stretchr/testify/require"
@@ -43,11 +43,7 @@ func TestStore_ProvideService(t *testing.T) {
}
}}
service := ProvideService(fakes.NewFakePluginRegistry(), srcs, l)
ctx := context.Background()
err := service.StartAsync(ctx)
require.NoError(t, err)
err = service.AwaitRunning(ctx)
_, err := ProvideService(fakes.NewFakePluginRegistry(), srcs, l)
require.NoError(t, err)
require.Equal(t, []plugins.Class{"1", "2", "3"}, loadedSrcs)
})
@@ -59,13 +55,12 @@ func TestStore_Plugin(t *testing.T) {
p1.RegisterClient(&DecommissionedPlugin{})
p2 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-panel"}}
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
ps := New(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p1.ID: p1,
p2.ID: p2,
},
}, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
}, &fakes.FakeLoader{})
p, exists := ps.Plugin(context.Background(), p1.ID)
require.False(t, exists)
@@ -86,7 +81,7 @@ func TestStore_Plugins(t *testing.T) {
p5 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "e-test-panel", Type: plugins.TypePanel}}
p5.RegisterClient(&DecommissionedPlugin{})
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
ps := New(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p1.ID: p1,
p2.ID: p2,
@@ -94,8 +89,7 @@ func TestStore_Plugins(t *testing.T) {
p4.ID: p4,
p5.ID: p5,
},
}, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
}, &fakes.FakeLoader{})
ToGrafanaDTO(p1)
pss := ps.Plugins(context.Background())
@@ -130,7 +124,7 @@ func TestStore_Routes(t *testing.T) {
p6 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "f-test-app", Type: plugins.TypeApp}}
p6.RegisterClient(&DecommissionedPlugin{})
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
ps := New(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p1.ID: p1,
p2.ID: p2,
@@ -138,8 +132,7 @@ func TestStore_Routes(t *testing.T) {
p5.ID: p5,
p6.ID: p6,
},
}, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
}, &fakes.FakeLoader{})
sr := func(p *plugins.Plugin) *plugins.StaticRoute {
return &plugins.StaticRoute{PluginID: p.ID, Directory: p.FS.Base()}
@@ -151,62 +144,39 @@ func TestStore_Routes(t *testing.T) {
}
func TestProcessManager_shutdown(t *testing.T) {
t.Run("When context is cancelled the plugin is stopped", func(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())
p := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-datasource", Type: plugins.TypeDataSource}} // Backend: true
backend := &fakes.FakeBackendPlugin{}
p.RegisterClient(backend)
p.SetLogger(log.NewTestLogger())
unloaded := false
ps := New(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p.ID: p,
},
}, &fakes.FakeLoader{
UnloadFunc: func(_ context.Context, plugin *plugins.Plugin) (*plugins.Plugin, error) {
require.Equal(t, p, plugin)
unloaded = true
return nil, nil
},
}, &fakes.FakeSourceRegistry{})
ctx, cancel := context.WithCancel(context.Background())
err := ps.StartAsync(ctx)
require.NoError(t, err)
err = ps.AwaitRunning(ctx)
require.NoError(t, err)
// Cancel context to trigger shutdown
cancel()
// Wait for service to be fully terminated
err = ps.AwaitTerminated(context.Background())
require.NoError(t, err)
require.True(t, unloaded)
unloaded := false
ps := New(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p.ID: p,
},
}, &fakes.FakeLoader{
UnloadFunc: func(_ context.Context, plugin *plugins.Plugin) (*plugins.Plugin, error) {
require.Equal(t, p, plugin)
unloaded = true
return nil, nil
},
})
t.Run("When shutdown fails, stopping method returns error", func(t *testing.T) {
p := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-datasource", Type: plugins.TypeDataSource}}
backend := &fakes.FakeBackendPlugin{}
p.RegisterClient(backend)
p.SetLogger(log.NewTestLogger())
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()
}()
expectedErr := errors.New("unload failed")
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p.ID: p,
},
}, &fakes.FakeLoader{
UnloadFunc: func(_ context.Context, plugin *plugins.Plugin) (*plugins.Plugin, error) {
return nil, expectedErr
},
}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
err = ps.stopping(nil)
require.Error(t, err)
require.ErrorIs(t, err, expectedErr)
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, unloaded)
})
}
@@ -216,13 +186,12 @@ func TestStore_availablePlugins(t *testing.T) {
p1.RegisterClient(&DecommissionedPlugin{})
p2 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-app"}}
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
ps := New(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p1.ID: p1,
p2.ID: p2,
},
}, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
}, &fakes.FakeLoader{})
aps := ps.availablePlugins(context.Background())
require.Len(t, aps, 1)