feat: scheduler (6/): add scheduler profile + plugin interfaces (Azur…

…e#381)

pkg/scheduler/framework/interface.go

@@ -0,0 +1,90 @@
+/*
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+*/
+
+package framework
+
+import (
+	"context"
+
+	fleetv1 "go.goms.io/fleet/apis/v1"
+)
+
+// Plugin is the interface which all scheduler plugins should implement.
+type Plugin interface {
+	Name() string
+	// TO-DO （chenyu1): add a method to help plugin set up the framework as needed.
+}
+
+// PostBatchPlugin is the interface which all plugins that would like to run at the PostBatch
+// extension point should implement.
+type PostBatchPlugin interface {
+	Plugin
+
+	// PostBatch runs after the scheduler has determined the number of bindings to create;
+	// a plugin which registers at this extension point must return one of the follows:
+	// * A Success status with a new batch size; or
+	// * A Skip status, if no changes in batch size is needed; or
+	// * An InternalError status, if an expected error has occurred
+	PostBatch(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot) (size int, status *Status)
+}
+
+// PreFilterPlugin is the interface which all plugins that would like to run at the PreFilter
+// extension point should implement.
+type PreFilterPlugin interface {
+	Plugin
+
+	// PreFilter runs before the scheduler enters the Filter stage; a plugin may perform
+	// some setup at this extension point, such as caching the results that will be used in
+	// following Filter calls, and/or run some checks to determine if it should be skipped in
+	// the Filter stage.
+	//
+	// A plugin which registers at this extension point must return one of the follows:
+	// * A Success status, if the plugin should run at the Filter stage; or
+	// * A Skip status, if the plugin should be skipped at the Filter stage; or
+	// * An InternalError status, if an expected error has occurred
+	PreFilter(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot) (status *Status)
+}
+
+// FilterPlugin is the interface which all plugins that would like to run at the Filter
+// extension point should implement.
+type FilterPlugin interface {
+	Plugin
+
+	// Filter runs at the Filter stage, to check if a placement can be bound to a specific cluster.
+	// A plugin which registers at this extension point must return one of the follows:
+	// * A Success status, if the placement can be bound to the cluster; or
+	// * A ClusterUnschedulable status, if the placement cannot be bound to the cluster; or
+	// * An InternalError status, if an expected error has occurred
+	Filter(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot, cluster *fleetv1.MemberCluster) (status *Status)
+}
+
+// PreScorePlugin is the interface which all plugins that would like to run at the PreScore
+// extension point should implement.
+type PreScorePlugin interface {
+	Plugin
+
+	// PreScore runs before the scheduler enters the Score stage; a plugin may perform
+	// some setup at this extension point, such as caching the results that will be used in
+	// following Score calls, and/or run some checks to determine if it should be skipped in
+	// the Filter stage.
+	//
+	// A plugin which registers at this extension point must return one of the follows:
+	// * A Success status, if the plugin should run at the Score stage; or
+	// * A Skip status, if the plugin should be skipped at the Score stage; or
+	// * An InternalError status, if an expected error has occurred
+	PreScore(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot) (status *Status)
+}
+
+// ScorePlugin is the interface which all plugins that would like to run at the Score
+// extension point should implement.
+type ScorePlugin interface {
+	Plugin
+
+	// Score runs at the Score stage, to score a cluster for a specific placement.
+	// A plugin which registers at this extension point must return one of the follows:
+	// * A Success status, with the score for the cluster; or
+	// * An InternalError status, if an expected error has occurred
+	Score(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot, cluster *fleetv1.MemberCluster) (score *ClusterScore, status *Status)
+}

pkg/scheduler/framework/profile.go

@@ -0,0 +1,70 @@
+/*
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+*/
+
+package framework
+
+// Profile specifies the scheduling profile a framework uses; it includes the plugins in use
+// by the framework at each extension point in order.
+//
+// At this moment, since Fleet does not support runtime profiles, all plugins are registered
+// directly to one universal profile, in their instantiated forms, rather than decoupled using
+// a factory registry and instantiated along with the profile's associated framework.
+type Profile struct {
+	name string
+
+	postBatchPlugins []PostBatchPlugin
+	preFilterPlugins []PreFilterPlugin
+	filterPlugins    []FilterPlugin
+	preScorePlugins  []PreScorePlugin
+	scorePlugins     []ScorePlugin
+
+	// RegisteredPlugins is a map of all plugins registered to the profile, keyed by their names.
+	// This helps to avoid setting up same plugin multiple times with the framework if the plugin
+	// registers at multiple extension points.
+	registeredPlugins map[string]Plugin
+}
+
+// WithPostBatchPlugin registers a PostBatchPlugin to the profile.
+func (profile *Profile) WithPostBatchPlugin(plugin PostBatchPlugin) *Profile {
+	profile.postBatchPlugins = append(profile.postBatchPlugins, plugin)
+	profile.registeredPlugins[plugin.Name()] = plugin
+	return profile
+}
+
+// WithPreFilterPlugin registers a PreFilterPlugin to the profile.
+func (profile *Profile) WithPreFilterPlugin(plugin PreFilterPlugin) *Profile {
+	profile.preFilterPlugins = append(profile.preFilterPlugins, plugin)
+	profile.registeredPlugins[plugin.Name()] = plugin
+	return profile
+}
+
+// WithFilterPlugin registers a FilterPlugin to the profile.
+func (profile *Profile) WithFilterPlugin(plugin FilterPlugin) *Profile {
+	profile.filterPlugins = append(profile.filterPlugins, plugin)
+	profile.registeredPlugins[plugin.Name()] = plugin
+	return profile
+}
+
+// WithPreScorePlugin registers a PreScorePlugin to the profile.
+func (profile *Profile) WithPreScorePlugin(plugin PreScorePlugin) *Profile {
+	profile.preScorePlugins = append(profile.preScorePlugins, plugin)
+	profile.registeredPlugins[plugin.Name()] = plugin
+	return profile
+}
+
+// WithScorePlugin registers a ScorePlugin to the profile.
+func (profile *Profile) WithScorePlugin(plugin ScorePlugin) *Profile {
+	profile.scorePlugins = append(profile.scorePlugins, plugin)
+	profile.registeredPlugins[plugin.Name()] = plugin
+	return profile
+}
+
+// NewProfile creates scheduling profile.
+func NewProfile(name string) *Profile {
+	return &Profile{
+		name:              name,
+		registeredPlugins: map[string]Plugin{},
+	}
+}

pkg/scheduler/framework/profile_test.go

@@ -0,0 +1,82 @@
+/*
+Copyright (c) Microsoft Corporation.
+Licensed under the MIT license.
+*/
+
+package framework
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+	fleetv1 "go.goms.io/fleet/apis/v1"
+)
+
+const (
+	dummyProfileName = "dummyProfile"
+	dummyPluginName  = "dummyAllPurposePlugin"
+)
+
+// A no-op, dummy plugin which connects to all extension points.
+type DummyAllPurposePlugin struct{}
+
+// Name returns the name of the dummy plugin.
+func (p *DummyAllPurposePlugin) Name() string {
+	return dummyPluginName
+}
+
+// PostBatch implements the PostBatch interface for the dummy plugin.
+func (p *DummyAllPurposePlugin) PostBatch(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot) (size int, status *Status) { //nolint:revive
+	return 1, nil
+}
+
+// PreFilter implements the PreFilter interface for the dummy plugin.
+func (p *DummyAllPurposePlugin) PreFilter(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot) (status *Status) { //nolint:revive
+	return nil
+}
+
+// Filter implements the Filter interface for the dummy plugin.
+func (p *DummyAllPurposePlugin) Filter(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot, cluster *fleetv1.MemberCluster) (status *Status) { //nolint:revive
+	return nil
+}
+
+// PreScore implements the PreScore interface for the dummy plugin.
+func (p *DummyAllPurposePlugin) PreScore(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot) (status *Status) { //nolint:revive
+	return nil
+}
+
+// Score implements the Score interface for the dummy plugin.
+func (p *DummyAllPurposePlugin) Score(ctx context.Context, state CycleStatePluginReadWriter, policy *fleetv1.ClusterPolicySnapshot, cluster *fleetv1.MemberCluster) (score *ClusterScore, status *Status) { //nolint:revive
+	return &ClusterScore{}, nil
+}
+
+// TestProfile tests the basic ops of a Profile.
+func TestProfile(t *testing.T) {
+	profile := NewProfile(dummyProfileName)
+
+	dummyAllPurposePlugin := &DummyAllPurposePlugin{}
+	dummyPlugin := Plugin(dummyAllPurposePlugin)
+
+	profile.WithPostBatchPlugin(dummyAllPurposePlugin)
+	profile.WithPreFilterPlugin(dummyAllPurposePlugin)
+	profile.WithFilterPlugin(dummyAllPurposePlugin)
+	profile.WithPreScorePlugin(dummyAllPurposePlugin)
+	profile.WithScorePlugin(dummyAllPurposePlugin)
+
+	wantProfile := &Profile{
+		name:             dummyProfileName,
+		postBatchPlugins: []PostBatchPlugin{dummyAllPurposePlugin},
+		preFilterPlugins: []PreFilterPlugin{dummyAllPurposePlugin},
+		filterPlugins:    []FilterPlugin{dummyAllPurposePlugin},
+		preScorePlugins:  []PreScorePlugin{dummyAllPurposePlugin},
+		scorePlugins:     []ScorePlugin{dummyAllPurposePlugin},
+		registeredPlugins: map[string]Plugin{
+			dummyPluginName: dummyPlugin,
+		},
+	}
+
+	if !cmp.Equal(profile, wantProfile, cmp.AllowUnexported(Profile{})) {
+		t.Fatalf("NewProfile() = %v, want %v", profile, wantProfile)
+	}
+}