Plugin loader: refactor step evaluation (#85881)

pkg/plugins/manager/fakes/fakes.go

@@ -476,36 +476,36 @@ func (f *FakeDiscoverer) Discover(ctx context.Context, src plugins.PluginSource)
 }
 
 type FakeBootstrapper struct {
-	BootstrapFunc func(ctx context.Context, src plugins.PluginSource, bundles []*plugins.FoundBundle) ([]*plugins.Plugin, error)
+	BootstrapFunc func(ctx context.Context, src plugins.PluginSource, bundle *plugins.FoundBundle) ([]*plugins.Plugin, error)
 }
 
-func (f *FakeBootstrapper) Bootstrap(ctx context.Context, src plugins.PluginSource, bundles []*plugins.FoundBundle) ([]*plugins.Plugin, error) {
+func (f *FakeBootstrapper) Bootstrap(ctx context.Context, src plugins.PluginSource, bundle *plugins.FoundBundle) ([]*plugins.Plugin, error) {
 	if f.BootstrapFunc != nil {
-		return f.BootstrapFunc(ctx, src, bundles)
+		return f.BootstrapFunc(ctx, src, bundle)
 	}
 	return []*plugins.Plugin{}, nil
 }
 
 type FakeValidator struct {
-	ValidateFunc func(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error)
+	ValidateFunc func(ctx context.Context, ps *plugins.Plugin) error
 }
 
-func (f *FakeValidator) Validate(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error) {
+func (f *FakeValidator) Validate(ctx context.Context, ps *plugins.Plugin) error {
 	if f.ValidateFunc != nil {
 		return f.ValidateFunc(ctx, ps)
 	}
-	return []*plugins.Plugin{}, nil
+	return nil
 }
 
 type FakeInitializer struct {
-	IntializeFunc func(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error)
+	IntializeFunc func(ctx context.Context, ps *plugins.Plugin) (*plugins.Plugin, error)
 }
 
-func (f *FakeInitializer) Initialize(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error) {
+func (f *FakeInitializer) Initialize(ctx context.Context, ps *plugins.Plugin) (*plugins.Plugin, error) {
 	if f.IntializeFunc != nil {
 		return f.IntializeFunc(ctx, ps)
 	}
-	return []*plugins.Plugin{}, nil
+	return ps, nil
 }
 
 type FakeTerminator struct {

pkg/plugins/manager/loader/loader.go

@@ -44,19 +44,34 @@ func (l *Loader) Load(ctx context.Context, src plugins.PluginSource) ([]*plugins
 		return nil, err
 	}
 
-	bootstrappedPlugins, err := l.bootstrap.Bootstrap(ctx, src, discoveredPlugins)
-	if err != nil {
-		return nil, err
+	bootstrappedPlugins := []*plugins.Plugin{}
+	for _, foundBundle := range discoveredPlugins {
+		bootstrappedPlugin, err := l.bootstrap.Bootstrap(ctx, src, foundBundle)
+		if err != nil {
+			// TODO: Add error to registry
+			continue
+		}
+		bootstrappedPlugins = append(bootstrappedPlugins, bootstrappedPlugin...)
 	}
 
-	validatedPlugins, err := l.validation.Validate(ctx, bootstrappedPlugins)
-	if err != nil {
-		return nil, err
+	validatedPlugins := []*plugins.Plugin{}
+	for _, bootstrappedPlugin := range bootstrappedPlugins {
+		err := l.validation.Validate(ctx, bootstrappedPlugin)
+		if err != nil {
+			// TODO: Add error to registry
+			continue
+		}
+		validatedPlugins = append(validatedPlugins, bootstrappedPlugin)
 	}
 
-	initializedPlugins, err := l.initializer.Initialize(ctx, validatedPlugins)
-	if err != nil {
-		return nil, err
+	initializedPlugins := []*plugins.Plugin{}
+	for _, validatedPlugin := range validatedPlugins {
+		initializedPlugin, err := l.initializer.Initialize(ctx, validatedPlugin)
+		if err != nil {
+			// TODO: Add error to registry
+			continue
+		}
+		initializedPlugins = append(initializedPlugins, initializedPlugin)
 	}
 
 	end(initializedPlugins)

pkg/plugins/manager/loader/loader_test.go

@@ -450,24 +450,22 @@ func TestLoader_Load(t *testing.T) {
 					return []*plugins.FoundBundle{{Primary: plugins.FoundPlugin{JSONData: pluginJSON}}}, nil
 				},
 			}, &fakes.FakeBootstrapper{
-				BootstrapFunc: func(ctx context.Context, s plugins.PluginSource, b []*plugins.FoundBundle) ([]*plugins.Plugin, error) {
-					require.True(t, len(b) == 1)
-					require.Equal(t, b[0].Primary.JSONData, pluginJSON)
+				BootstrapFunc: func(ctx context.Context, s plugins.PluginSource, b *plugins.FoundBundle) ([]*plugins.Plugin, error) {
+					require.Equal(t, b.Primary.JSONData, pluginJSON)
 					require.Equal(t, src, s)
 
 					steps = append(steps, "bootstrap")
 					return []*plugins.Plugin{plugin}, nil
 				},
-			}, &fakes.FakeValidator{ValidateFunc: func(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error) {
-				require.Equal(t, []*plugins.Plugin{plugin}, ps)
+			}, &fakes.FakeValidator{ValidateFunc: func(ctx context.Context, ps *plugins.Plugin) error {
+				require.Equal(t, plugin, ps)
 
 				steps = append(steps, "validate")
-				return ps, nil
+				return nil
 			}},
 			&fakes.FakeInitializer{
-				IntializeFunc: func(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error) {
-					require.True(t, len(ps) == 1)
-					require.Equal(t, ps[0].JSONData, pluginJSON)
+				IntializeFunc: func(ctx context.Context, ps *plugins.Plugin) (*plugins.Plugin, error) {
+					require.Equal(t, ps.JSONData, pluginJSON)
 					steps = append(steps, "initialize")
 					return ps, nil
 				},

pkg/plugins/manager/pipeline/bootstrap/bootstrap.go

@@ -12,11 +12,11 @@ import (
 
 // Bootstrapper is responsible for the Bootstrap stage of the plugin loader pipeline.
 type Bootstrapper interface {
-	Bootstrap(ctx context.Context, src plugins.PluginSource, bundles []*plugins.FoundBundle) ([]*plugins.Plugin, error)
+	Bootstrap(ctx context.Context, src plugins.PluginSource, bundle *plugins.FoundBundle) ([]*plugins.Plugin, error)
 }
 
 // ConstructFunc is the function used for the Construct step of the Bootstrap stage.
-type ConstructFunc func(ctx context.Context, src plugins.PluginSource, bundles []*plugins.FoundBundle) ([]*plugins.Plugin, error)
+type ConstructFunc func(ctx context.Context, src plugins.PluginSource, bundle *plugins.FoundBundle) ([]*plugins.Plugin, error)
 
 // DecorateFunc is the function used for the Decorate step of the Bootstrap stage.
 type DecorateFunc func(ctx context.Context, p *plugins.Plugin) (*plugins.Plugin, error)
@@ -59,7 +59,7 @@ func New(cfg *config.PluginManagementCfg, opts Opts) *Bootstrap {
 }
 
 // Bootstrap will execute the Construct and Decorate steps of the Bootstrap stage.
-func (b *Bootstrap) Bootstrap(ctx context.Context, src plugins.PluginSource, found []*plugins.FoundBundle) ([]*plugins.Plugin, error) {
+func (b *Bootstrap) Bootstrap(ctx context.Context, src plugins.PluginSource, found *plugins.FoundBundle) ([]*plugins.Plugin, error) {
 	ps, err := b.constructStep(ctx, src, found)
 	if err != nil {
 		return nil, err
@@ -72,18 +72,14 @@ func (b *Bootstrap) Bootstrap(ctx context.Context, src plugins.PluginSource, fou
 	bootstrappedPlugins := make([]*plugins.Plugin, 0, len(ps))
 	for _, p := range ps {
 		var ip *plugins.Plugin
-		stepFailed := false
 		for _, decorate := range b.decorateSteps {
 			ip, err = decorate(ctx, p)
 			if err != nil {
-				stepFailed = true
 				b.log.Error("Could not decorate plugin", "pluginId", p.ID, "error", err)
-				break
+				return nil, err
 			}
 		}
-		if !stepFailed {
-			bootstrappedPlugins = append(bootstrappedPlugins, ip)
-		}
+		bootstrappedPlugins = append(bootstrappedPlugins, ip)
 	}
 
 	return bootstrappedPlugins, nil

pkg/plugins/manager/pipeline/bootstrap/steps.go

@@ -47,36 +47,34 @@ func NewDefaultConstructor(signatureCalculator plugins.SignatureCalculator, asse
 }
 
 // Construct will calculate the plugin's signature state and create the plugin using the pluginFactoryFunc.
-func (c *DefaultConstructor) Construct(ctx context.Context, src plugins.PluginSource, bundles []*plugins.FoundBundle) ([]*plugins.Plugin, error) {
-	res := make([]*plugins.Plugin, 0, len(bundles))
+func (c *DefaultConstructor) Construct(ctx context.Context, src plugins.PluginSource, bundle *plugins.FoundBundle) ([]*plugins.Plugin, error) {
+	res := []*plugins.Plugin{}
 
-	for _, bundle := range bundles {
-		sig, err := c.signatureCalculator.Calculate(ctx, src, bundle.Primary)
+	sig, err := c.signatureCalculator.Calculate(ctx, src, bundle.Primary)
+	if err != nil {
+		c.log.Warn("Could not calculate plugin signature state", "pluginId", bundle.Primary.JSONData.ID, "error", err)
+		return nil, err
+	}
+	plugin, err := c.pluginFactoryFunc(bundle.Primary, src.PluginClass(ctx), sig)
+	if err != nil {
+		c.log.Error("Could not create primary plugin base", "pluginId", bundle.Primary.JSONData.ID, "error", err)
+		return nil, err
+	}
+	res = append(res, plugin)
+
+	children := make([]*plugins.Plugin, 0, len(bundle.Children))
+	for _, child := range bundle.Children {
+		cp, err := c.pluginFactoryFunc(*child, plugin.Class, sig)
 		if err != nil {
-			c.log.Warn("Could not calculate plugin signature state", "pluginId", bundle.Primary.JSONData.ID, "error", err)
-			continue
+			c.log.Error("Could not create child plugin base", "pluginId", child.JSONData.ID, "error", err)
+			return nil, err
 		}
-		plugin, err := c.pluginFactoryFunc(bundle.Primary, src.PluginClass(ctx), sig)
-		if err != nil {
-			c.log.Error("Could not create primary plugin base", "pluginId", bundle.Primary.JSONData.ID, "error", err)
-			continue
-		}
-		res = append(res, plugin)
+		cp.Parent = plugin
+		plugin.Children = append(plugin.Children, cp)
 
-		children := make([]*plugins.Plugin, 0, len(bundle.Children))
-		for _, child := range bundle.Children {
-			cp, err := c.pluginFactoryFunc(*child, plugin.Class, sig)
-			if err != nil {
-				c.log.Error("Could not create child plugin base", "pluginId", child.JSONData.ID, "error", err)
-				continue
-			}
-			cp.Parent = plugin
-			plugin.Children = append(plugin.Children, cp)
-
-			children = append(children, cp)
-		}
-		res = append(res, children...)
+		children = append(children, cp)
 	}
+	res = append(res, children...)
 
 	return res, nil
 }

pkg/plugins/manager/pipeline/initialization/initialization.go

@@ -10,7 +10,7 @@ import (
 
 // Initializer is responsible for the Initialization stage of the plugin loader pipeline.
 type Initializer interface {
-	Initialize(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error)
+	Initialize(ctx context.Context, ps *plugins.Plugin) (*plugins.Plugin, error)
 }
 
 // InitializeFunc is the function used for the Initialize step of the Initialization stage.
@@ -40,28 +40,20 @@ func New(cfg *config.PluginManagementCfg, opts Opts) *Initialize {
 }
 
 // Initialize will execute the Initialize steps of the Initialization stage.
-func (i *Initialize) Initialize(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error) {
+func (i *Initialize) Initialize(ctx context.Context, ps *plugins.Plugin) (*plugins.Plugin, error) {
 	if len(i.initializeSteps) == 0 {
 		return ps, nil
 	}
 
 	var err error
-	initializedPlugins := make([]*plugins.Plugin, 0, len(ps))
-	for _, p := range ps {
-		var ip *plugins.Plugin
-		stepFailed := false
-		for _, init := range i.initializeSteps {
-			ip, err = init(ctx, p)
-			if err != nil {
-				stepFailed = true
-				i.log.Error("Could not initialize plugin", "pluginId", p.ID, "error", err)
-				break
-			}
-		}
-		if !stepFailed {
-			initializedPlugins = append(initializedPlugins, ip)
+	var ip *plugins.Plugin
+	for _, init := range i.initializeSteps {
+		ip, err = init(ctx, ps)
+		if err != nil {
+			i.log.Error("Could not initialize plugin", "pluginId", ps.ID, "error", err)
+			return nil, err
 		}
 	}
 
-	return initializedPlugins, nil
+	return ip, nil
 }

pkg/plugins/manager/pipeline/validation/validation.go

@@ -10,7 +10,7 @@ import (
 
 // Validator is responsible for the Validation stage of the plugin loader pipeline.
 type Validator interface {
-	Validate(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error)
+	Validate(ctx context.Context, ps *plugins.Plugin) error
 }
 
 // ValidateFunc is the function used for the Validate step of the Validation stage.
@@ -40,26 +40,18 @@ func New(cfg *config.PluginManagementCfg, opts Opts) *Validate {
 }
 
 // Validate will execute the Validate steps of the Validation stage.
-func (v *Validate) Validate(ctx context.Context, ps []*plugins.Plugin) ([]*plugins.Plugin, error) {
+func (v *Validate) Validate(ctx context.Context, ps *plugins.Plugin) error {
 	if len(v.validateSteps) == 0 {
-		return ps, nil
+		return nil
 	}
 
-	validatedPlugins := make([]*plugins.Plugin, 0, len(ps))
-	for _, p := range ps {
-		stepFailed := false
-		for _, validate := range v.validateSteps {
-			err := validate(ctx, p)
-			if err != nil {
-				stepFailed = true
-				v.log.Error("Plugin validation failed", "pluginId", p.ID, "error", err)
-				break
-			}
-		}
-		if !stepFailed {
-			validatedPlugins = append(validatedPlugins, p)
+	for _, validate := range v.validateSteps {
+		err := validate(ctx, ps)
+		if err != nil {
+			v.log.Error("Plugin validation failed", "pluginId", ps.ID, "error", err)
+			return err
 		}
 	}
 
-	return validatedPlugins, nil
+	return nil
 }