Use plugin registry instead of pluginClients and stageBasedPluingsMap…

… in the controller, scheduler, and planner

Signed-off-by: Yoshiki Fujikane <ffjlabo@gmail.com>

pkg/app/pipedv1/cmd/piped/piped.go

@@ -61,6 +61,7 @@ import (
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller/controllermetrics"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/eventwatcher"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/notifier"
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/statsreporter"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/trigger"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
@@ -348,7 +349,7 @@ func (p *piped) run(ctx context.Context, input cli.Input) (runErr error) {
 	}
 
 	// Make grpc clients to connect to plugins.
-	pluginClis := make([]pluginapi.PluginClient, 0, len(cfg.Plugins))
+	plugins := make([]plugin.Plugin, 0, len(cfg.Plugins))
 	options := []rpcclient.DialOption{
 		rpcclient.WithBlock(),
 		rpcclient.WithInsecure(),
@@ -357,8 +358,19 @@ func (p *piped) run(ctx context.Context, input cli.Input) (runErr error) {
 		cli, err := pluginapi.NewClient(ctx, net.JoinHostPort("localhost", strconv.Itoa(plg.Port)), options...)
 		if err != nil {
 			input.Logger.Error("failed to create client to connect plugin", zap.String("plugin", plg.Name), zap.Error(err))
+			return err
 		}
-		pluginClis = append(pluginClis, cli)
+
+		plugins = append(plugins, plugin.Plugin{
+			Name: plg.Name,
+			Cli:  cli,
+		})
+	}
+
+	pluginRegistry, err := plugin.NewPluginRegistry(ctx, plugins)
+	if err != nil {
+		input.Logger.Error("failed to create plugin registry", zap.Error(err))
+		return err
 	}
 
 	// Start running application live state reporter.
@@ -383,7 +395,7 @@ func (p *piped) run(ctx context.Context, input cli.Input) (runErr error) {
 		c := controller.NewController(
 			apiClient,
 			gitClient,
-			pluginClis,
+			pluginRegistry,
 			deploymentLister,
 			commandLister,
 			notifier,

pkg/app/pipedv1/controller/controller.go

@@ -34,11 +34,10 @@ import (
 	"google.golang.org/grpc/status"
 
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller/controllermetrics"
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
 	"github.com/pipe-cd/pipecd/pkg/git"
 	"github.com/pipe-cd/pipecd/pkg/model"
-	pluginapi "github.com/pipe-cd/pipecd/pkg/plugin/api/v1alpha1"
-	"github.com/pipe-cd/pipecd/pkg/plugin/api/v1alpha1/deployment"
 )
 
 type apiClient interface {
@@ -96,10 +95,9 @@ type controller struct {
 	notifier         notifier
 	secretDecrypter  secretDecrypter
 
-	// gRPC clients to communicate with plugins.
-	pluginClients []pluginapi.PluginClient
-	// Map from stage name to the plugin client.
-	stageBasedPluginsMap map[string]pluginapi.PluginClient
+	// The registry of all plugins.
+	pluginRegistry plugin.PluginRegistry
+
 	// Map from application ID to the planner
 	// of a pending deployment of that application.
 	planners map[string]*planner
@@ -131,7 +129,7 @@ type controller struct {
 func NewController(
 	apiClient apiClient,
 	gitClient gitClient,
-	pluginClients []pluginapi.PluginClient,
+	pluginRegistry plugin.PluginRegistry,
 	deploymentLister deploymentLister,
 	commandLister commandLister,
 	notifier notifier,
@@ -144,7 +142,7 @@ func NewController(
 	return &controller{
 		apiClient:        apiClient,
 		gitClient:        gitClient,
-		pluginClients:    pluginClients,
+		pluginRegistry:   pluginRegistry,
 		deploymentLister: deploymentLister,
 		commandLister:    commandLister,
 		notifier:         notifier,
@@ -179,23 +177,6 @@ func (c *controller) Run(ctx context.Context) error {
 	c.workspaceDir = dir
 	c.logger.Info(fmt.Sprintf("workspace directory was configured to %s", c.workspaceDir))
 
-	// Build the list of stages that can be handled by piped's plugins.
-	stagesBasedPluginsMap := make(map[string]pluginapi.PluginClient)
-	for _, plugin := range c.pluginClients {
-		resp, err := plugin.FetchDefinedStages(ctx, &deployment.FetchDefinedStagesRequest{})
-		if err != nil {
-			return err
-		}
-		for _, stage := range resp.GetStages() {
-			if _, ok := stagesBasedPluginsMap[stage]; ok {
-				c.logger.Error("duplicated stage name", zap.String("stage", stage))
-				return fmt.Errorf("duplicated stage name %s", stage)
-			}
-			stagesBasedPluginsMap[stage] = plugin
-		}
-	}
-	c.stageBasedPluginsMap = stagesBasedPluginsMap
-
 	ticker := time.NewTicker(c.syncInternal)
 	defer ticker.Stop()
 	c.logger.Info("start syncing planners and schedulers")
@@ -448,8 +429,7 @@ func (c *controller) startNewPlanner(ctx context.Context, d *model.Deployment) (
 		commitHash,
 		configFilename,
 		workingDir,
-		c.pluginClients, // FIXME: Find a way to ensure the plugins only related to deployment.
-		c.stageBasedPluginsMap,
+		c.pluginRegistry,
 		c.apiClient,
 		c.gitClient,
 		c.notifier,
@@ -590,7 +570,7 @@ func (c *controller) startNewScheduler(ctx context.Context, d *model.Deployment)
 		workingDir,
 		c.apiClient,
 		c.gitClient,
-		c.stageBasedPluginsMap,
+		c.pluginRegistry,
 		c.notifier,
 		c.secretDecrypter,
 		c.logger,

pkg/app/pipedv1/controller/planner.go

@@ -31,6 +31,7 @@ import (
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/controller/controllermetrics"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/deploysource"
 	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/metadatastore"
+	"github.com/pipe-cd/pipecd/pkg/app/pipedv1/plugin"
 	"github.com/pipe-cd/pipecd/pkg/app/server/service/pipedservice"
 	config "github.com/pipe-cd/pipecd/pkg/configv1"
 	"github.com/pipe-cd/pipecd/pkg/model"
@@ -57,13 +58,6 @@ type planner struct {
 	lastSuccessfulConfigFilename string
 	workingDir                   string
 
-	// The plugin clients are used to call plugin that actually
-	// performs planning deployment.
-	plugins []pluginapi.PluginClient
-	// The map used to know which plugin is incharged for a given stage
-	// of the current deployment.
-	stageBasedPluginsMap map[string]pluginapi.PluginClient
-
 	// The apiClient is used to report the deployment status.
 	apiClient apiClient
 
@@ -79,6 +73,9 @@ type planner struct {
 	// which encrypted using PipeCD built-in secret management.
 	secretDecrypter secretDecrypter
 
+	// The pluginRegistry is used to determine which plugins to be used
+	pluginRegistry plugin.PluginRegistry
+
 	logger *zap.Logger
 	tracer trace.Tracer
 
@@ -96,8 +93,7 @@ func newPlanner(
 	lastSuccessfulCommitHash string,
 	lastSuccessfulConfigFilename string,
 	workingDir string,
-	pluginClients []pluginapi.PluginClient,
-	stageBasedPluginsMap map[string]pluginapi.PluginClient,
+	pluginRegistry plugin.PluginRegistry,
 	apiClient apiClient,
 	gitClient gitClient,
 	notifier notifier,
@@ -119,8 +115,7 @@ func newPlanner(
 		lastSuccessfulCommitHash:     lastSuccessfulCommitHash,
 		lastSuccessfulConfigFilename: lastSuccessfulConfigFilename,
 		workingDir:                   workingDir,
-		stageBasedPluginsMap:         stageBasedPluginsMap,
-		plugins:                      pluginClients,
+		pluginRegistry:               pluginRegistry,
 		apiClient:                    apiClient,
 		gitClient:                    gitClient,
 		metadataStore:                metadatastore.NewMetadataStore(apiClient, d),
@@ -266,8 +261,21 @@ func (p *planner) buildPlan(ctx context.Context, runningDS, targetDS *deployment
 		TargetDeploymentSource:  targetDS,
 	}
 
+	cfg, err := config.DecodeYAML[*config.GenericApplicationSpec](targetDS.GetApplicationConfig())
+	if err != nil {
+		p.logger.Error("unable to parse application config", zap.Error(err))
+		return nil, err
+	}
+	spec := cfg.Spec
+
+	plugins, err := p.pluginRegistry.GetPluginClientsByAppConfig(spec)
+	if err != nil {
+		p.logger.Error("unable to get plugin clients by app config", zap.Error(err))
+		return nil, err
+	}
+
 	// Build deployment target versions.
-	for _, plg := range p.plugins {
+	for _, plg := range plugins {
 		vRes, err := plg.DetermineVersions(ctx, &deployment.DetermineVersionsRequest{Input: input})
 		if err != nil {
 			p.logger.Warn("unable to determine versions", zap.Error(err))
@@ -284,13 +292,6 @@ func (p *planner) buildPlan(ctx context.Context, runningDS, targetDS *deployment
 		}
 	}
 
-	cfg, err := config.DecodeYAML[*config.GenericApplicationSpec](targetDS.GetApplicationConfig())
-	if err != nil {
-		p.logger.Error("unable to parse application config", zap.Error(err))
-		return nil, err
-	}
-	spec := cfg.Spec
-
 	// In case the strategy has been decided by trigger.
 	// For example: user triggered the deployment via web console.
 	switch p.deployment.Trigger.SyncStrategy {
@@ -375,7 +376,7 @@ func (p *planner) buildPlan(ctx context.Context, runningDS, targetDS *deployment
 		summary  string
 	)
 	// Build plan based on plugins determined strategy
-	for _, plg := range p.plugins {
+	for _, plg := range plugins {
 		res, err := plg.DetermineStrategy(ctx, &deployment.DetermineStrategyRequest{Input: input})
 		if err != nil {
 			p.logger.Warn("Unable to determine strategy using current plugin", zap.Error(err))
@@ -421,7 +422,12 @@ func (p *planner) buildQuickSyncStages(ctx context.Context, cfg *config.GenericA
 		rollbackStages = []*model.PipelineStage{}
 		rollback       = *cfg.Planner.AutoRollback
 	)
-	for _, plg := range p.plugins {
+
+	plugins, err := p.pluginRegistry.GetPluginClientsByAppConfig(cfg)
+	if err != nil {
+		return nil, err
+	}
+	for _, plg := range plugins {
 		res, err := plg.BuildQuickSyncStages(ctx, &deployment.BuildQuickSyncStagesRequest{Rollback: rollback})
 		if err != nil {
 			return nil, fmt.Errorf("failed to build quick sync stage deployment (%w)", err)
@@ -462,9 +468,9 @@ func (p *planner) buildPipelineSyncStages(ctx context.Context, cfg *config.Gener
 	// Build stages config for each plugin.
 	for i := range stagesCfg {
 		stageCfg := stagesCfg[i]
-		plg, ok := p.stageBasedPluginsMap[stageCfg.Name.String()]
-		if !ok {
-			return nil, fmt.Errorf("unable to find plugin for stage %q", stageCfg.Name.String())
+		plg, err := p.pluginRegistry.GetPluginClientByStageName(stageCfg.Name.String())
+		if err != nil {
+			return nil, err
 		}
 
 		stagesCfgPerPlugin[plg] = append(stagesCfgPerPlugin[plg], &deployment.BuildPipelineSyncStagesRequest_StageConfig{