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

pkg/services/pluginsintegration/pluginstore/store.go

@@ -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,80 @@ 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
+	}
+	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 {
 	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 +111,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 +165,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 +179,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()
 }