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

Plugins: Move store init to dskit service (#111823)

Browse Source
This commit is contained in:
Todd Treece
2025-10-02 19:53:31 -04:00
committed by GitHub
parent 5798181fb0
commit 2d232aa10d
16 changed files with 394 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/api/plugins_test.go
+18 -9
View File
@@ -12,19 +12,18 @@ import (
"strings"
"testing"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/prometheus/client_golang/prometheus"
dto "github.com/prometheus/client_model/go"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/grafana/grafana-plugin-sdk-go/backend"
"github.com/grafana/grafana/pkg/plugins/auth"
"github.com/grafana/grafana/pkg/api/dtos"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/infra/log/logtest"
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/auth"
"github.com/grafana/grafana/pkg/plugins/config"
"github.com/grafana/grafana/pkg/plugins/manager/fakes"
"github.com/grafana/grafana/pkg/plugins/manager/filestore"
@@ -528,9 +527,12 @@ func callGetPluginAsset(sc *scenarioContext) {
func pluginAssetScenario(t *testing.T, desc string, url string, urlPattern string,
cfg *setting.Cfg, pluginRegistry registry.Service, fn scenarioFunc) {
t.Run(fmt.Sprintf("%s %s", desc, url), func(t *testing.T) {
store, err := pluginstore.NewPluginStoreForTest(pluginRegistry, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
hs := HTTPServer{
Cfg: cfg,
pluginStore: pluginstore.New(pluginRegistry, &fakes.FakeLoader{}),
pluginStore: store,
pluginFileStore: filestore.ProvideService(pluginRegistry),
log: log.NewNopLogger(),
pluginsCDNService: pluginscdn.ProvideService(&config.PluginManagementCfg{
@@ -640,12 +642,14 @@ func Test_PluginsList_AccessControl(t *testing.T) {
for _, tc := range tcs {
t.Run(tc.desc, func(t *testing.T) {
server := SetupAPITestServer(t, func(hs *HTTPServer) {
store, err := pluginstore.NewPluginStoreForTest(pluginRegistry, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
hs.Cfg = setting.NewCfg()
hs.PluginSettings = &pluginSettings
hs.pluginStore = pluginstore.New(pluginRegistry, &fakes.FakeLoader{})
hs.pluginStore = store
hs.pluginFileStore = filestore.ProvideService(pluginRegistry)
hs.managedPluginsService = managedplugins.NewNoop()
var err error
hs.pluginsUpdateChecker, err = updatemanager.ProvidePluginsService(
hs.Cfg,
hs.pluginStore,
@@ -828,9 +832,12 @@ func Test_PluginsSettings(t *testing.T) {
for _, tc := range tcs {
t.Run(tc.desc, func(t *testing.T) {
server := SetupAPITestServer(t, func(hs *HTTPServer) {
store, err := pluginstore.NewPluginStoreForTest(pluginRegistry, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
hs.Cfg = setting.NewCfg()
hs.PluginSettings = &pluginSettings
hs.pluginStore = pluginstore.New(pluginRegistry, &fakes.FakeLoader{})
hs.pluginStore = store
hs.pluginFileStore = filestore.ProvideService(pluginRegistry)
errTracker := pluginerrs.ProvideErrorTracker()
if tc.errCode != "" {
@@ -844,7 +851,6 @@ func Test_PluginsSettings(t *testing.T) {
sig := signature.ProvideService(pCfg, statickey.New())
hs.pluginAssets = pluginassets.ProvideService(pCfg, pluginCDN, sig, hs.pluginStore)
hs.pluginErrorResolver = pluginerrs.ProvideStore(errTracker)
var err error
hs.pluginsUpdateChecker, err = updatemanager.ProvidePluginsService(
hs.Cfg,
hs.pluginStore,
@@ -896,9 +902,12 @@ func Test_UpdatePluginSetting(t *testing.T) {
t.Run("should return an error when trying to disable an auto-enabled plugin", func(t *testing.T) {
server := SetupAPITestServer(t, func(hs *HTTPServer) {
store, err := pluginstore.NewPluginStoreForTest(pluginRegistry, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
hs.Cfg = setting.NewCfg()
hs.PluginSettings = &pluginSettings
hs.pluginStore = pluginstore.New(pluginRegistry, &fakes.FakeLoader{})
hs.pluginStore = store
hs.pluginFileStore = filestore.ProvideService(pluginRegistry)
hs.managedPluginsService = managedplugins.NewNoop()
hs.log = log.NewNopLogger()
pkg/registry/apis/datasource/register.go
+50 -9
View File
@@ -3,7 +3,9 @@ package datasource
import (
"context"
"encoding/json"
"errors"
"fmt"
"path/filepath"
"github.com/prometheus/client_golang/prometheus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -20,14 +22,16 @@ import (
"github.com/grafana/grafana/pkg/apimachinery/utils"
datasource "github.com/grafana/grafana/pkg/apis/datasource/v0alpha1"
query "github.com/grafana/grafana/pkg/apis/query/v0alpha1"
"github.com/grafana/grafana/pkg/configprovider"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/plugins"
"github.com/grafana/grafana/pkg/plugins/manager/sources"
"github.com/grafana/grafana/pkg/promlib/models"
"github.com/grafana/grafana/pkg/registry/apis/query/queryschema"
"github.com/grafana/grafana/pkg/services/accesscontrol"
"github.com/grafana/grafana/pkg/services/apiserver/builder"
"github.com/grafana/grafana/pkg/services/featuremgmt"
"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/tsdb/grafana-testdata-datasource/kinds"
)
@@ -47,12 +51,12 @@ type DataSourceAPIBuilder struct {
}
func RegisterAPIService(
cfgProvider configprovider.ConfigProvider,
features featuremgmt.FeatureToggles,
apiRegistrar builder.APIRegistrar,
pluginClient plugins.Client, // access to everything
datasources ScopedPluginDatasourceProvider,
contextProvider PluginContextWrapper,
pluginStore pluginstore.Store,
accessControl accesscontrol.AccessControl,
reg prometheus.Registerer,
) (*DataSourceAPIBuilder, error) {
@@ -66,25 +70,43 @@ func RegisterAPIService(
var err error
var builder *DataSourceAPIBuilder
all := pluginStore.Plugins(context.Background(), plugins.TypeDataSource)
cfg, err := cfgProvider.Get(context.Background())
if err != nil {
return nil, err
}
pluginJSONs, err := getCorePlugins(cfg)
if err != nil {
return nil, err
}
ids := []string{
"grafana-testdata-datasource",
"prometheus",
"graphite",
}
for _, ds := range all {
if explictPluginList && !slices.Contains(ids, ds.ID) {
for _, pluginJSON := range pluginJSONs {
if explictPluginList && !slices.Contains(ids, pluginJSON.ID) {
continue // skip this one
}
if !ds.Backend {
if !pluginJSON.Backend {
continue // skip frontend only plugins
}
builder, err = NewDataSourceAPIBuilder(ds.JSONData,
pluginClient,
datasources.GetDatasourceProvider(ds.JSONData),
if pluginJSON.Type != plugins.TypeDataSource {
continue // skip non-datasource plugins
}
client, ok := pluginClient.(PluginClient)
if !ok {
return nil, fmt.Errorf("plugin client is not a PluginClient: %T", pluginClient)
}
builder, err = NewDataSourceAPIBuilder(pluginJSON,
client,
datasources.GetDatasourceProvider(pluginJSON),
contextProvider,
accessControl,
features.IsEnabledGlobally(featuremgmt.FlagDatasourceQueryTypes),
@@ -277,3 +299,22 @@ func (b *DataSourceAPIBuilder) PostProcessOpenAPI(oas *spec3.OpenAPI) (*spec3.Op
return oas, err
}
func getCorePlugins(cfg *setting.Cfg) ([]plugins.JSONData, error) {
coreDataSourcesPath := filepath.Join(cfg.StaticRootPath, "app", "plugins", "datasource")
coreDataSourcesSrc := sources.NewLocalSource(
plugins.ClassCore,
[]string{coreDataSourcesPath},
)
res, err := coreDataSourcesSrc.Discover(context.Background())
if err != nil {
return nil, errors.New("failed to load core data source plugins")
}
pluginJSONs := make([]plugins.JSONData, 0, len(res))
for _, p := range res {
pluginJSONs = append(pluginJSONs, p.Primary.JSONData)
}
return pluginJSONs, nil
}
pkg/registry/backgroundsvcs/adapter/dependencies.go
+16 -1
View File
@@ -3,9 +3,21 @@ package adapter
import (
"github.com/grafana/grafana/pkg/infra/tracing"
"github.com/grafana/grafana/pkg/modules"
"github.com/grafana/grafana/pkg/services/pluginsintegration/plugininstaller"
"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginstore"
"github.com/grafana/grafana/pkg/services/provisioning"
)
const (
// PluginStore is the module name for the plugin store service.
PluginStore = pluginstore.ServiceName
// PluginInstaller is the module name for the plugin installer service.
PluginInstaller = plugininstaller.ServiceName
// Provisioning is the module name for the provisioning service.
Provisioning = provisioning.ServiceName
// Tracing is the module name for the tracing service.
Tracing = tracing.ServiceName
@@ -29,7 +41,10 @@ func dependencyMap() map[string][]string {
return map[string][]string{
Tracing: {},
GrafanaAPIServer: {Tracing},
Core: {GrafanaAPIServer},
PluginStore: {GrafanaAPIServer},
PluginInstaller: {PluginStore},
Provisioning: {PluginStore, PluginInstaller},
Core: {GrafanaAPIServer, PluginStore, PluginInstaller, Provisioning},
BackgroundServices: {Core},
}
}
pkg/registry/backgroundsvcs/adapter/manager.go
+2 -1
View File
@@ -14,7 +14,7 @@ import (
)
var (
stopTimeout = 30 * time.Second
stopTimeout = 5 * time.Second
)
type ManagerAdapter struct {
@@ -63,6 +63,7 @@ func (m *ManagerAdapter) starting(ctx context.Context) error {
// skip disabled services
if s, ok := bgSvc.(registry.CanBeDisabled); ok && s.IsDisabled() {
logger.Debug("Skipping disabled service", "service", namedService.ServiceName())
manager.RegisterInvisibleModule(namedService.ServiceName(), nil)
continue
}
pkg/registry/backgroundsvcs/adapter/service.go
+19 -6
View File
@@ -18,9 +18,10 @@ var _ services.NamedService = &serviceAdapter{}
// The adapter uses dskit's BasicService with a custom RunningFn:
// - Starting phase: No-op, transitions immediately to Running
// - Running phase: Delegates to the wrapped service's Run method
// - Stopping phase: No-op, transitions immediately to Terminated/Failed
// - Stopping phase: Closes the stop channel to signal the service to stop
type serviceAdapter struct {
*services.BasicService
services.NamedService
stopCh chan struct{}
name string
service registry.BackgroundService
}
@@ -36,8 +37,9 @@ func asNamedService(service registry.BackgroundService) *serviceAdapter {
a := &serviceAdapter{
name: name,
service: service,
stopCh: make(chan struct{}),
}
a.BasicService = services.NewBasicService(nil, a.run, nil).WithName(name)
a.NamedService = services.NewBasicService(nil, a.running, a.stopping).WithName(name)
return a
}
@@ -46,13 +48,24 @@ func asNamedService(service registry.BackgroundService) *serviceAdapter {
// background service's Run method. If the background service completes without
// error, the adapter waits for context cancellation (service stop) before
// transitioning to Stopping state, ensuring proper dskit service lifecycle.
func (a *serviceAdapter) run(ctx context.Context) error {
err := a.service.Run(ctx)
func (a *serviceAdapter) running(ctx context.Context) error {
serviceCtx, serviceCancel := context.WithCancel(ctx)
go func() {
<-a.stopCh
serviceCancel()
}()
err := a.service.Run(serviceCtx)
if err != nil && !errors.Is(err, context.Canceled) {
return err
}
// wait for context cancellation to transition to Stopping state.
// this prevents the service from causing it's dependents to stop prematurely.
<-ctx.Done()
<-serviceCtx.Done()
return nil
}
func (a *serviceAdapter) stopping(_ error) error {
close(a.stopCh)
return nil
}
pkg/registry/backgroundsvcs/adapter/service_test.go
+1 -1
View File
@@ -16,7 +16,7 @@ func TestAsNamedService(t *testing.T) {
adapter := asNamedService(mockSvc)
require.NotNil(t, adapter)
require.NotNil(t, adapter.BasicService)
require.NotNil(t, adapter.NamedService)
require.Equal(t, mockSvc, adapter.service)
expectedName := reflect.TypeOf(mockSvc).String()
pkg/server/wire_gen.go
+4 -10
View File
@@ -547,10 +547,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
}
errorRegistry := pluginerrs.ProvideErrorTracker()
loaderLoader := loader.ProvideService(pluginManagementCfg, discovery, bootstrap, validate, initialize, terminate, errorRegistry)
pluginstoreService, err := pluginstore.ProvideService(inMemory, sourcesService, loaderLoader)
if err != nil {
return nil, err
}
pluginstoreService := pluginstore.ProvideService(inMemory, sourcesService, loaderLoader)
filestoreService := filestore.ProvideService(inMemory)
fileStoreManager := dashboards.ProvideFileStoreManager(pluginstoreService, filestoreService)
folderPermissionsService, err := ossaccesscontrol.ProvideFolderPermissions(cfg, featureToggles, routeRegisterImpl, sqlStore, accessControl, ossLicensingService, folderimplService, acimplService, teamService, userService, actionSetService)
@@ -810,7 +807,7 @@ func Initialize(ctx context.Context, cfg *setting.Cfg, opts Options, apiOpts api
apiService := api4.ProvideService(cfg, routeRegisterImpl, accessControl, userService, authinfoimplService, ossGroups, identitySynchronizer, orgService, ldapImpl, userAuthTokenService, bundleregistryService)
dashboardsAPIBuilder := dashboard.RegisterAPIService(cfg, featureToggles, apiserverService, dashboardService, dashboardProvisioningService, service15, dashboardServiceImpl, dashboardPermissionsService, accessControl, accessClient, provisioningServiceImpl, dashboardsStore, registerer, sqlStore, tracingService, resourceClient, dualwriteService, sortService, quotaService, libraryPanelService, eventualRestConfigProvider, userService)
snapshotsAPIBuilder := dashboardsnapshot.RegisterAPIService(serviceImpl, apiserverService, cfg, featureToggles, sqlStore, registerer)
dataSourceAPIBuilder, err := datasource.RegisterAPIService(featureToggles, apiserverService, middlewareHandler, scopedPluginDatasourceProvider, plugincontextProvider, pluginstoreService, accessControl, registerer)
dataSourceAPIBuilder, err := datasource.RegisterAPIService(configProvider, featureToggles, apiserverService, middlewareHandler, scopedPluginDatasourceProvider, plugincontextProvider, accessControl, registerer)
if err != nil {
return nil, err
}
@@ -1152,10 +1149,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
}
errorRegistry := pluginerrs.ProvideErrorTracker()
loaderLoader := loader.ProvideService(pluginManagementCfg, discovery, bootstrap, validate, initialize, terminate, errorRegistry)
pluginstoreService, err := pluginstore.ProvideService(inMemory, sourcesService, loaderLoader)
if err != nil {
return nil, err
}
pluginstoreService := pluginstore.ProvideService(inMemory, sourcesService, loaderLoader)
filestoreService := filestore.ProvideService(inMemory)
fileStoreManager := dashboards.ProvideFileStoreManager(pluginstoreService, filestoreService)
folderPermissionsService, err := ossaccesscontrol.ProvideFolderPermissions(cfg, featureToggles, routeRegisterImpl, sqlStore, accessControl, ossLicensingService, folderimplService, acimplService, teamService, userService, actionSetService)
@@ -1417,7 +1411,7 @@ func InitializeForTest(ctx context.Context, t sqlutil.ITestDB, testingT interfac
apiService := api4.ProvideService(cfg, routeRegisterImpl, accessControl, userService, authinfoimplService, ossGroups, identitySynchronizer, orgService, ldapImpl, userAuthTokenService, bundleregistryService)
dashboardsAPIBuilder := dashboard.RegisterAPIService(cfg, featureToggles, apiserverService, dashboardService, dashboardProvisioningService, service15, dashboardServiceImpl, dashboardPermissionsService, accessControl, accessClient, provisioningServiceImpl, dashboardsStore, registerer, sqlStore, tracingService, resourceClient, dualwriteService, sortService, quotaService, libraryPanelService, eventualRestConfigProvider, userService)
snapshotsAPIBuilder := dashboardsnapshot.RegisterAPIService(serviceImpl, apiserverService, cfg, featureToggles, sqlStore, registerer)
dataSourceAPIBuilder, err := datasource.RegisterAPIService(featureToggles, apiserverService, middlewareHandler, scopedPluginDatasourceProvider, plugincontextProvider, pluginstoreService, accessControl, registerer)
dataSourceAPIBuilder, err := datasource.RegisterAPIService(configProvider, featureToggles, apiserverService, middlewareHandler, scopedPluginDatasourceProvider, plugincontextProvider, accessControl, registerer)
if err != nil {
return nil, err
}
pkg/services/apiserver/service.go
+2 -7
View File
@@ -85,7 +85,6 @@ type service struct {
features featuremgmt.FeatureToggles
log log.Logger
stopCh chan struct{}
stoppedCh chan error
db db.DB
@@ -148,7 +147,6 @@ func ProvideService(
cfg: cfg,
features: features,
rr: rr,
stopCh: make(chan struct{}),
builders: []builder.APIGroupBuilder{},
authorizer: authorizer.NewGrafanaBuiltInSTAuthorizer(cfg),
tracing: tracing,
@@ -242,11 +240,8 @@ func (s *service) Run(ctx context.Context) error {
if err := s.StartAsync(ctx); err != nil {
return err
}
if err := s.AwaitRunning(ctx); err != nil {
return err
}
return s.AwaitTerminated(ctx)
stopCtx := context.Background()
return s.AwaitTerminated(stopCtx)
}
func (s *service) RegisterAPI(b builder.APIGroupBuilder) {
pkg/services/pluginsintegration/plugincontext/plugincontext_test.go
+3 -1
View File
@@ -41,8 +41,10 @@ func TestGet(t *testing.T) {
cfg := setting.NewCfg()
ds := &fakeDatasources.FakeDataSourceService{}
db := &dbtest.FakeDB{ExpectedError: pluginsettings.ErrPluginSettingNotFound}
store, err := pluginstore.NewPluginStoreForTest(preg, &pluginFakes.FakeLoader{}, &pluginFakes.FakeSourceRegistry{})
require.NoError(t, err)
pcp := plugincontext.ProvideService(cfg, localcache.ProvideService(),
pluginstore.New(preg, &pluginFakes.FakeLoader{}), &fakeDatasources.FakeCacheService{},
store, &fakeDatasources.FakeCacheService{},
ds, pluginSettings.ProvideService(db, secretstest.NewFakeSecretsService()), pluginconfig.NewFakePluginRequestConfigProvider(),
)
identity := &user.SignedInUser{OrgID: int64(1), Login: "admin"}
pkg/services/pluginsintegration/plugininstaller/service.go
+38 -31
View File
@@ -8,6 +8,7 @@ import (
"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/repo"
@@ -18,6 +19,8 @@ import (
"github.com/prometheus/client_golang/prometheus"
)
const ServiceName = "plugin.backgroundinstaller"
var (
installRequestCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
Namespace: "plugins",
@@ -36,6 +39,7 @@ var (
)
type Service struct {
services.NamedService
cfg *setting.Cfg
log log.Logger
pluginInstaller plugins.Installer
@@ -43,6 +47,7 @@ type Service struct {
pluginRepo repo.Service
features featuremgmt.FeatureToggles
updateChecker pluginchecker.PluginUpdateChecker
installComplete chan struct{} // closed when all plugins are installed (used for testing)
}
func ProvideService(
@@ -60,21 +65,18 @@ func ProvideService(
})
s := &Service{
log: log.New("plugin.backgroundinstaller"),
log: log.New(ServiceName),
cfg: cfg,
pluginInstaller: pluginInstaller,
pluginStore: pluginStore,
pluginRepo: pluginRepo,
features: features,
updateChecker: updateChecker,
installComplete: make(chan struct{}),
}
if len(cfg.PreinstallPluginsSync) > 0 {
// Block initialization process until plugins are installed
err := s.installPluginsWithTimeout(cfg.PreinstallPluginsSync)
if err != nil {
return nil, err
}
}
s.NamedService = services.NewBasicService(s.starting, s.running, nil).WithName(ServiceName)
return s, nil
}
@@ -83,24 +85,6 @@ func (s *Service) IsDisabled() bool {
return len(s.cfg.PreinstallPluginsAsync) == 0
}
func (s *Service) installPluginsWithTimeout(pluginsToInstall []setting.InstallPlugin) error {
// Installation process does not timeout by default nor reuses the context
// passed to the request so we need to handle the timeout here.
// We could make this timeout configurable in the future.
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
done := make(chan struct{ err error })
go func() {
done <- struct{ err error }{err: s.installPlugins(ctx, pluginsToInstall, true)}
}()
select {
case <-ctx.Done():
return fmt.Errorf("failed to install plugins: %w", ctx.Err())
case d := <-done:
return d.err
}
}
func (s *Service) shouldUpdate(ctx context.Context, pluginID, currentVersion string, pluginURL string) bool {
// If the plugin is installed from a URL, we cannot check for updates as we do not have the version information
// from the repository. Therefore, we assume that the plugin should be updated if the URL is provided.
@@ -166,11 +150,34 @@ func (s *Service) installPlugins(ctx context.Context, pluginsToInstall []setting
return nil
}
func (s *Service) Run(ctx context.Context) error {
err := s.installPlugins(ctx, s.cfg.PreinstallPluginsAsync, false)
if err != nil {
// Unexpected error, asynchronous installation should not return errors
s.log.Error("Failed to install plugins", "error", err)
func (s *Service) starting(ctx context.Context) error {
if len(s.cfg.PreinstallPluginsSync) > 0 {
s.log.Info("Installing plugins", "plugins", s.cfg.PreinstallPluginsSync)
if err := s.installPlugins(ctx, s.cfg.PreinstallPluginsSync, true); err != nil {
s.log.Error("Failed to install plugins", "error", err)
return err
}
}
s.log.Info("Plugins installed", "plugins", s.cfg.PreinstallPluginsSync)
return nil
}
func (s *Service) running(ctx context.Context) error {
if len(s.cfg.PreinstallPluginsAsync) > 0 {
s.log.Info("Installing plugins", "plugins", s.cfg.PreinstallPluginsAsync)
if err := s.installPlugins(ctx, s.cfg.PreinstallPluginsAsync, false); err != nil {
s.log.Error("Failed to install plugins", "error", err)
return err
}
}
close(s.installComplete)
<-ctx.Done()
return nil
}
func (s *Service) Run(ctx context.Context) error {
if err := s.StartAsync(ctx); err != nil {
return err
}
return s.AwaitTerminated(ctx)
}
pkg/services/pluginsintegration/plugininstaller/service_test.go
+20 -4
View File
@@ -26,7 +26,7 @@ func TestService_IsDisabled(t *testing.T) {
&setting.Cfg{
PreinstallPluginsAsync: []setting.InstallPlugin{{ID: "myplugin"}},
},
pluginstore.New(registry.NewInMemory(), &fakes.FakeLoader{}),
pluginstore.New(registry.NewInMemory(), &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{}),
&fakes.FakePluginInstaller{},
prometheus.NewRegistry(),
&fakes.FakePluginRepo{},
@@ -160,12 +160,14 @@ func TestService_Run(t *testing.T) {
}
installed := 0
installedFromURL := 0
store, err := pluginstore.NewPluginStoreForTest(preg, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
s, err := ProvideService(
&setting.Cfg{
PreinstallPluginsAsync: tt.pluginsToInstall,
PreinstallPluginsSync: tt.pluginsToInstallSync,
},
pluginstore.New(preg, &fakes.FakeLoader{}),
store,
&fakes.FakePluginInstaller{
AddFunc: func(ctx context.Context, pluginID string, version string, opts plugins.AddOpts) error {
for _, plugin := range tt.pluginsToFail {
@@ -203,13 +205,26 @@ func TestService_Run(t *testing.T) {
&pluginchecker.FakePluginPreinstall{},
),
)
require.NoError(t, err)
t.Cleanup(func() {
s.StopAsync()
err := s.AwaitTerminated(context.Background())
if tt.shouldThrowError {
require.ErrorContains(t, err, "Failed to install plugin")
return
}
require.NoError(t, err)
})
err = s.StartAsync(context.Background())
require.NoError(t, err)
err = s.AwaitRunning(context.Background())
if tt.shouldThrowError {
require.ErrorContains(t, err, "Failed to install plugin")
return
}
require.NoError(t, err)
err = s.Run(context.Background())
require.NoError(t, err)
if tt.shouldInstall {
expectedInstalled := 0
@@ -232,6 +247,7 @@ func TestService_Run(t *testing.T) {
expectedInstalled++
}
}
<-s.installComplete
require.Equal(t, expectedInstalled, installed)
require.Equal(t, expectedInstalledFromURL, installedFromURL)
}
pkg/services/pluginsintegration/pluginstore/store.go
+76 -29
View File
@@ -3,18 +3,21 @@ 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.
@@ -25,47 +28,81 @@ 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, error) {
ctx := context.Background()
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
}
stopCtx := context.Background()
return s.AwaitTerminated(stopCtx)
}
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 {
start := time.Now()
totalPlugins := 0
logger := log.New("plugin.store")
logger := log.New(ServiceName)
logger.Info("Loading plugins...")
for _, ps := range pluginSources.List(ctx) {
loadedPlugins, err := pluginLoader.Load(ctx, ps)
for _, ps := range s.pluginSources.List(ctx) {
loadedPlugins, err := s.pluginLoader.Load(ctx, ps)
if err != nil {
logger.Error("Loading plugin source failed", "source", ps.PluginClass(ctx), "error", err)
return nil, err
return err
}
totalPlugins += len(loadedPlugins)
}
logger.Info("Plugins loaded", "count", totalPlugins, "duration", time.Since(start))
return New(pluginRegistry, pluginLoader), nil
return nil
}
func (s *Service) Run(ctx context.Context) error {
func (s *Service) running(ctx context.Context) error {
<-ctx.Done()
s.shutdown(ctx)
return ctx.Err()
return nil
}
func New(pluginRegistry registry.Service, pluginLoader loader.Service) *Service {
return &Service{
pluginRegistry: pluginRegistry,
pluginLoader: pluginLoader,
}
func (s *Service) stopping(failureReason error) error {
return s.shutdown(context.Background())
}
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
@@ -75,6 +112,10 @@ 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
@@ -125,6 +166,10 @@ 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) {
@@ -135,18 +180,20 @@ func (s *Service) Routes(ctx context.Context) []*plugins.StaticRoute {
return staticRoutes
}
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)
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
}
p.Logger().Debug("Plugin stopped")
}(ctx, plugin)
plugin.Logger().Debug("Plugin stopped")
return nil
})
}
wg.Wait()
return errgroup.Wait()
}
pkg/services/pluginsintegration/pluginstore/store_test.go
+70 -39
View File
@@ -2,7 +2,7 @@ package pluginstore
import (
"context"
"sync"
"errors"
"testing"
"github.com/stretchr/testify/require"
@@ -43,7 +43,11 @@ func TestStore_ProvideService(t *testing.T) {
}
}}
_, err := ProvideService(fakes.NewFakePluginRegistry(), srcs, l)
service := ProvideService(fakes.NewFakePluginRegistry(), srcs, l)
ctx := context.Background()
err := service.StartAsync(ctx)
require.NoError(t, err)
err = service.AwaitRunning(ctx)
require.NoError(t, err)
require.Equal(t, []plugins.Class{"1", "2", "3"}, loadedSrcs)
})
@@ -55,12 +59,13 @@ func TestStore_Plugin(t *testing.T) {
p1.RegisterClient(&DecommissionedPlugin{})
p2 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-panel"}}
ps := New(&fakes.FakePluginRegistry{
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p1.ID: p1,
p2.ID: p2,
},
}, &fakes.FakeLoader{})
}, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
p, exists := ps.Plugin(context.Background(), p1.ID)
require.False(t, exists)
@@ -81,7 +86,7 @@ func TestStore_Plugins(t *testing.T) {
p5 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "e-test-panel", Type: plugins.TypePanel}}
p5.RegisterClient(&DecommissionedPlugin{})
ps := New(&fakes.FakePluginRegistry{
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p1.ID: p1,
p2.ID: p2,
@@ -89,7 +94,8 @@ func TestStore_Plugins(t *testing.T) {
p4.ID: p4,
p5.ID: p5,
},
}, &fakes.FakeLoader{})
}, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
ToGrafanaDTO(p1)
pss := ps.Plugins(context.Background())
@@ -124,7 +130,7 @@ func TestStore_Routes(t *testing.T) {
p6 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "f-test-app", Type: plugins.TypeApp}}
p6.RegisterClient(&DecommissionedPlugin{})
ps := New(&fakes.FakePluginRegistry{
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p1.ID: p1,
p2.ID: p2,
@@ -132,7 +138,8 @@ func TestStore_Routes(t *testing.T) {
p5.ID: p5,
p6.ID: p6,
},
}, &fakes.FakeLoader{})
}, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
sr := func(p *plugins.Plugin) *plugins.StaticRoute {
return &plugins.StaticRoute{PluginID: p.ID, Directory: p.FS.Base()}
@@ -144,39 +151,62 @@ func TestStore_Routes(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())
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())
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
},
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)
})
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 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())
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)
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)
})
}
@@ -186,12 +216,13 @@ func TestStore_availablePlugins(t *testing.T) {
p1.RegisterClient(&DecommissionedPlugin{})
p2 := &plugins.Plugin{JSONData: plugins.JSONData{ID: "test-app"}}
ps := New(&fakes.FakePluginRegistry{
ps, err := NewPluginStoreForTest(&fakes.FakePluginRegistry{
Store: map[string]*plugins.Plugin{
p1.ID: p1,
p2.ID: p2,
},
}, &fakes.FakeLoader{})
}, &fakes.FakeLoader{}, &fakes.FakeSourceRegistry{})
require.NoError(t, err)
aps := ps.availablePlugins(context.Background())
require.Len(t, aps, 1)
pkg/services/pluginsintegration/test_helper.go
+1 -1
View File
@@ -67,7 +67,7 @@ func CreateIntegrationTestCtx(t *testing.T, cfg *setting.Cfg, coreRegistry *core
Terminator: term,
})
ps, err := pluginstore.ProvideService(reg, sources.ProvideService(cfg, pCfg), l)
ps, err := pluginstore.NewPluginStoreForTest(reg, l, sources.ProvideService(cfg, pCfg))
require.NoError(t, err)
return &IntegrationTestCtx{
pkg/services/provisioning/provisioning.go
+72 -71
View File
@@ -7,6 +7,7 @@ import (
"path/filepath"
"sync"
"github.com/grafana/dskit/services"
"github.com/grafana/grafana/pkg/infra/db"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/infra/tracing"
@@ -39,6 +40,8 @@ import (
"github.com/grafana/grafana/pkg/storage/legacysql/dualwrite"
)
const ServiceName = "provisioning"
func ProvideService(
ac accesscontrol.AccessControl,
cfg *setting.Cfg,
@@ -91,6 +94,8 @@ func ProvideService(
dual: dual,
}
s.NamedService = services.NewBasicService(s.starting, s.running, nil).WithName(ServiceName)
if err := s.setDashboardProvisioner(); err != nil {
return nil, err
}
@@ -98,6 +103,67 @@ func ProvideService(
return s, nil
}
func (ps *ProvisioningServiceImpl) starting(ctx context.Context) error {
if err := ps.ProvisionDatasources(ctx); err != nil {
ps.log.Error("Failed to provision data sources", "error", err)
return err
}
if err := ps.ProvisionPlugins(ctx); err != nil {
ps.log.Error("Failed to provision plugins", "error", err)
return err
}
if err := ps.ProvisionAlerting(ctx); err != nil {
ps.log.Error("Failed to provision alerting", "error", err)
return err
}
// Migrating prom types relies on data source provisioning to already be completed
// If we can make services depend on other services completing first,
// then we should remove this from provisioning
if err := ps.migratePrometheusType(ctx); err != nil {
ps.log.Error("Failed to migrate Prometheus type", "error", err)
return err
}
if err := ps.ProvisionDashboards(ctx); err != nil {
ps.log.Error("Failed to provision dashboard", "error", err)
// Consider the allow list of errors for which running the provisioning service should not
// fail. For now this includes only dashboards.ErrGetOrCreateFolder.
if !errors.Is(err, dashboards.ErrGetOrCreateFolder) {
return err
}
}
if ps.dashboardProvisioner.HasDashboardSources() {
ps.searchService.TriggerReIndex()
}
return nil
}
func (ps *ProvisioningServiceImpl) running(ctx context.Context) error {
for {
// Wait for unlock. This is tied to new dashboardProvisioner to be instantiated before we start polling.
ps.mutex.Lock()
// Using background here because otherwise if root context was canceled the select later on would
// non-deterministically take one of the route possibly going into one polling loop before exiting.
pollingContext, cancelFun := context.WithCancel(context.Background())
ps.pollingCtxCancel = cancelFun
ps.dashboardProvisioner.PollChanges(pollingContext)
ps.mutex.Unlock()
select {
case <-pollingContext.Done():
// Polling was canceled.
continue
case <-ctx.Done():
// Root server context was cancelled so cancel polling and leave.
ps.cancelPolling()
return nil
}
}
}
func (ps *ProvisioningServiceImpl) setDashboardProvisioner() error {
dashboardPath := filepath.Join(ps.Cfg.ProvisioningPath, "dashboards")
dashProvisioner, err := ps.newDashboardProvisioner(context.Background(), dashboardPath, ps.dashboardProvisioningService, ps.orgService, ps.dashboardService, ps.folderService, ps.dual)
@@ -137,6 +203,8 @@ func newProvisioningServiceImpl(
migratePrometheusType: migratePrometheusType,
}
s.NamedService = services.NewBasicService(s.starting, s.running, nil).WithName(ServiceName)
if err := s.setDashboardProvisioner(); err != nil {
return nil, err
}
@@ -145,6 +213,7 @@ func newProvisioningServiceImpl(
}
type ProvisioningServiceImpl struct {
services.NamedService
Cfg *setting.Cfg
SQLStore db.DB
orgService org.Service
@@ -173,7 +242,6 @@ type ProvisioningServiceImpl struct {
resourcePermissions accesscontrol.ReceiverPermissionsService
tracer tracing.Tracer
dual dualwrite.Service
onceInitProvisioners sync.Once
migratePrometheusType func(context.Context) error
}
@@ -185,78 +253,11 @@ func (ps *ProvisioningServiceImpl) RunInitProvisioners(ctx context.Context) erro
}
func (ps *ProvisioningServiceImpl) Run(ctx context.Context) error {
var err error
// Run Datasources, Plugins and Alerting Provisioning only once.
// It can't be initialized at RunInitProvisioners because it
// depends on the /apis endpoints to be already running and listeningq
ps.onceInitProvisioners.Do(func() {
err = ps.ProvisionDatasources(ctx)
if err != nil {
ps.log.Error("Failed to provision data sources", "error", err)
return
}
err = ps.ProvisionPlugins(ctx)
if err != nil {
ps.log.Error("Failed to provision plugins", "error", err)
return
}
err = ps.ProvisionAlerting(ctx)
if err != nil {
ps.log.Error("Failed to provision alerting", "error", err)
return
}
// Migrating prom types relies on data source provisioning to already be completed
// If we can make services depend on other services completing first,
// then we should remove this from provisioning
err = ps.migratePrometheusType(ctx)
if err != nil {
ps.log.Error("Failed to migrate Prometheus type", "error", err)
return
}
})
if err != nil {
// error already logged
if err := ps.StartAsync(ctx); err != nil {
return err
}
err = ps.ProvisionDashboards(ctx)
if err != nil {
ps.log.Error("Failed to provision dashboard", "error", err)
// Consider the allow list of errors for which running the provisioning service should not
// fail. For now this includes only dashboards.ErrGetOrCreateFolder.
if !errors.Is(err, dashboards.ErrGetOrCreateFolder) {
return err
}
}
if ps.dashboardProvisioner.HasDashboardSources() {
ps.searchService.TriggerReIndex()
}
for {
// Wait for unlock. This is tied to new dashboardProvisioner to be instantiated before we start polling.
ps.mutex.Lock()
// Using background here because otherwise if root context was canceled the select later on would
// non-deterministically take one of the route possibly going into one polling loop before exiting.
pollingContext, cancelFun := context.WithCancel(context.Background())
ps.pollingCtxCancel = cancelFun
ps.dashboardProvisioner.PollChanges(pollingContext)
ps.mutex.Unlock()
select {
case <-pollingContext.Done():
// Polling was canceled.
continue
case <-ctx.Done():
// Root server context was cancelled so cancel polling and leave.
ps.cancelPolling()
return ctx.Err()
}
}
stopCtx := context.Background()
return ps.AwaitTerminated(stopCtx)
}
func (ps *ProvisioningServiceImpl) ProvisionDatasources(ctx context.Context) error {
pkg/services/provisioning/provisioning_test.go
+2 -2
View File
@@ -48,7 +48,7 @@ func TestProvisioningServiceImpl(t *testing.T) {
serviceTest.waitForStop()
assert.False(t, serviceTest.serviceRunning, "Service should not be running")
assert.Equal(t, context.Canceled, serviceTest.serviceError, "Service should have returned canceled error")
assert.NoError(t, serviceTest.serviceError, "Service should not have returned an error")
})
t.Run("Failed reloading does not stop polling with old provisioned", func(t *testing.T) {
@@ -91,7 +91,7 @@ func TestProvisioningServiceImpl(t *testing.T) {
serviceTest.cancel()
serviceTest.waitForStop()
assert.Equal(t, context.Canceled, serviceTest.serviceError)
assert.NoError(t, serviceTest.serviceError, "Service should not have returned an error")
})
t.Run("Should return run error when dashboard provisioning fails for non-allow-listed error", func(t *testing.T) {