Only require default plugin configuration for task types with multiple registered plugins

go.mod

@@ -34,3 +34,5 @@ require (
 )
 
 replace github.com/aws/amazon-sagemaker-operator-for-k8s => github.com/aws/amazon-sagemaker-operator-for-k8s v1.0.1-0.20210303003444-0fb33b1fd49d
+
+replace github.com/flyteorg/flyteplugins => github.com/flyteorg/flyteplugins v0.10.10-0.20220225212740-cf5b4b938049

go.sum

@@ -241,6 +241,8 @@ github.com/flyteorg/flyteidl v0.22.0 h1:lVAXyacTCIX6Fl0qWoFzyto9Hfx0ADlyPhHPmOMi
 github.com/flyteorg/flyteidl v0.22.0/go.mod h1:576W2ViEyjTpT+kEVHAGbrTP3HARNUZ/eCwrNPmdx9U=
 github.com/flyteorg/flyteplugins v0.10.9 h1:Hj4kBc2pNAJOTUP14+KSd44RzSE+A6fMEnTMQXhE6r8=
 github.com/flyteorg/flyteplugins v0.10.9/go.mod h1:RXgHGGUGC1akEnAd0yi4cLuYP1BF1rVkxhGjzIrm6VU=
+github.com/flyteorg/flyteplugins v0.10.10-0.20220225212740-cf5b4b938049 h1:EQmrADz8qFLHyye41k2NHyQqr1XXs+1tdYj7cgmNCNY=
+github.com/flyteorg/flyteplugins v0.10.10-0.20220225212740-cf5b4b938049/go.mod h1:RXgHGGUGC1akEnAd0yi4cLuYP1BF1rVkxhGjzIrm6VU=
 github.com/flyteorg/flytestdlib v0.3.13/go.mod h1:Tz8JCECAbX6VWGwFT6cmEQ+RJpZ/6L9pswu3fzWs220=
 github.com/flyteorg/flytestdlib v0.4.7/go.mod h1:fv1ar34LJLMTaf0tbfetisLykUlARi7rP+NQTUn6QQs=
 github.com/flyteorg/flytestdlib v0.4.12 h1:e88Tcu+8ug6T2tzzxRltbsuBkcz4coAquF9xqwIrmLw=

pkg/controller/nodes/task/config/config.go

@@ -50,7 +50,7 @@ type BarrierConfig struct {
 }
 
 type TaskPluginConfig struct {
-	EnabledPlugins []string `json:"enabled-plugins" pflag:",deprecated"`
+	EnabledPlugins []string `json:"enabled-plugins" pflag:",Plugins enabled currently"`
 	// Maps task types to their plugin handler (by ID).
 	DefaultForTaskTypes map[string]string `json:"default-for-task-types" pflag:"-,"`
 }

pkg/controller/nodes/task/handler.go

@@ -221,7 +221,7 @@ func (t *Handler) Setup(ctx context.Context, sCtx handler.SetupContext) error {
 
 	// Create the resource negotiator here
 	// and then convert it to proxies later and pass them to plugins
-	enabledPlugins, err := WranglePluginsAndGenerateFinalList(ctx, &t.cfg.TaskPlugins, t.pluginRegistry)
+	enabledPlugins, defaultForTaskTypes, err := WranglePluginsAndGenerateFinalList(ctx, &t.cfg.TaskPlugins, t.pluginRegistry)
 	if err != nil {
 		logger.Errorf(ctx, "Failed to finalize enabled plugins. Error: %s", err)
 		return err
@@ -244,7 +244,7 @@ func (t *Handler) Setup(ctx context.Context, sCtx handler.SetupContext) error {
 		// For every default plugin for a task type specified in flytepropeller config we validate that the plugin's
 		// static definition includes that task type as something it is registered to handle.
 		for _, tt := range p.RegisteredTaskTypes {
-			for _, defaultTaskType := range p.DefaultForTaskTypes {
+			for _, defaultTaskType := range defaultForTaskTypes[cp.GetID()] {
 				if defaultTaskType == tt {
 					if existingHandler, alreadyDefaulted := t.defaultPlugins[tt]; alreadyDefaulted && existingHandler.GetID() != cp.GetID() {
 						logger.Errorf(ctx, "TaskType [%s] has multiple default handlers specified: [%s] and [%s]",

pkg/controller/nodes/task/plugin_config.go

@@ -13,23 +13,21 @@ import (
 	"github.com/flyteorg/flytepropeller/pkg/controller/nodes/task/k8s"
 )
 
-func WranglePluginsAndGenerateFinalList(ctx context.Context, cfg *config.TaskPluginConfig, pr PluginRegistryIface) ([]core.PluginEntry, error) {
+func WranglePluginsAndGenerateFinalList(ctx context.Context, cfg *config.TaskPluginConfig, pr PluginRegistryIface) (enabledPlugins []core.PluginEntry, defaultForTaskTypes map[pluginID][]taskType, err error) {
 	allPluginsEnabled := false
 	pluginsConfigMeta := config.PluginsConfigMeta{
-		AllDefaultForTaskTypes: map[pluginID][]taskType{},
+		AllDefaultForTaskTypes: defaultForTaskTypes,
 	}
-	var err error
 	if cfg != nil {
 		pluginsConfigMeta, err = cfg.GetEnabledPlugins()
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 	}
 	if pluginsConfigMeta.EnabledPlugins.Len() == 0 {
 		allPluginsEnabled = true
 	}
 
-	var finalizedPlugins []core.PluginEntry
 	logger.Infof(ctx, "Enabled plugins: %v", pluginsConfigMeta.EnabledPlugins.List())
 	logger.Infof(ctx, "Loading core Plugins, plugin configuration [all plugins enabled: %v]", allPluginsEnabled)
 	for _, cpe := range pr.GetCorePlugins() {
@@ -38,10 +36,7 @@ func WranglePluginsAndGenerateFinalList(ctx context.Context, cfg *config.TaskPlu
 			logger.Infof(ctx, "Plugin [%s] is DISABLED (not found in enabled plugins list).", id)
 		} else {
 			logger.Infof(ctx, "Plugin [%s] ENABLED", id)
-			if defaults, ok := pluginsConfigMeta.AllDefaultForTaskTypes[id]; ok {
-				cpe.DefaultForTaskTypes = defaults
-			}
-			finalizedPlugins = append(finalizedPlugins, cpe)
+			enabledPlugins = append(enabledPlugins, cpe)
 		}
 	}
 
@@ -65,11 +60,10 @@ func WranglePluginsAndGenerateFinalList(ctx context.Context, cfg *config.TaskPlu
 				LoadPlugin: func(ctx context.Context, iCtx core.SetupContext) (plugin core.Plugin, e error) {
 					return k8s.NewPluginManagerWithBackOff(ctx, iCtx, kpe, backOffController, monitorIndex)
 				},
-				IsDefault:           kpe.IsDefault,
-				DefaultForTaskTypes: pluginsConfigMeta.AllDefaultForTaskTypes[id],
+				IsDefault: kpe.IsDefault,
 			}
-			finalizedPlugins = append(finalizedPlugins, plugin)
+			enabledPlugins = append(enabledPlugins, plugin)
 		}
 	}
-	return finalizedPlugins, nil
+	return enabledPlugins, pluginsConfigMeta.AllDefaultForTaskTypes, nil
 }

pkg/controller/nodes/task/plugin_config_test.go

@@ -61,7 +61,7 @@ func TestWranglePluginsAndGenerateFinalList(t *testing.T) {
 				core: tt.args.corePlugins,
 				k8s:  tt.args.k8sPlugins,
 			}
-			got, err := WranglePluginsAndGenerateFinalList(context.TODO(), tt.args.cfg, pr)
+			got, _, err := WranglePluginsAndGenerateFinalList(context.TODO(), tt.args.cfg, pr)
 			if (err != nil) != tt.want.err {
 				t.Errorf("WranglePluginsAndGenerateFinalList() error = %v, wantErr %v", err, tt.want.err)
 				return