Merge pull request #77567 from wgliang/features/scheduling-framework-post-bind

Add Post-bind extension point to the scheduling framework

pkg/scheduler/framework/v1alpha1/framework.go

@@ -38,6 +38,7 @@ type framework struct {
 	queueSortPlugins []QueueSortPlugin
 	reservePlugins   []ReservePlugin
 	prebindPlugins   []PrebindPlugin
+	postbindPlugins  []PostbindPlugin
 	unreservePlugins []UnreservePlugin
 	permitPlugins    []PermitPlugin
 }
@@ -112,6 +113,20 @@ func NewFramework(r Registry, plugins *config.Plugins, args []config.PluginConfi
 		}
 	}
 
+	if plugins.PostBind != nil {
+		for _, pb := range plugins.PostBind.Enabled {
+			if pg, ok := f.plugins[pb.Name]; ok {
+				p, ok := pg.(PostbindPlugin)
+				if !ok {
+					return nil, fmt.Errorf("plugin %v does not extend postbind plugin", pb.Name)
+				}
+				f.postbindPlugins = append(f.postbindPlugins, p)
+			} else {
+				return nil, fmt.Errorf("postbind plugin %v does not exist", pb.Name)
+			}
+		}
+	}
+
 	if plugins.Unreserve != nil {
 		for _, ur := range plugins.Unreserve.Enabled {
 			if pg, ok := f.plugins[ur.Name]; ok {
@@ -191,6 +206,14 @@ func (f *framework) RunPrebindPlugins(
 	return nil
 }
 
+// RunPostbindPlugins runs the set of configured postbind plugins.
+func (f *framework) RunPostbindPlugins(
+	pc *PluginContext, pod *v1.Pod, nodeName string) {
+	for _, pl := range f.postbindPlugins {
+		pl.Postbind(pc, pod, nodeName)
+	}
+}
+
 // RunReservePlugins runs the set of configured reserve plugins. If any of these
 // plugins returns an error, it does not continue running the remaining ones and
 // returns the error. In such case, pod will not be scheduled.

pkg/scheduler/framework/v1alpha1/interface.go

@@ -148,6 +148,17 @@ type PrebindPlugin interface {
 	Prebind(pc *PluginContext, p *v1.Pod, nodeName string) *Status
 }
 
+// PostbindPlugin is an interface that must be implemented by "postbind" plugins.
+// These plugins are called after a pod is successfully bound to a node.
+type PostbindPlugin interface {
+	Plugin
+	// Postbind is called after a pod is successfully bound. These plugins are
+	// informational. A common application of this extension point is for cleaning
+	// up. If a plugin needs to clean-up its state after a pod is scheduled and
+	// bound, Postbind is the extension point that it should register.
+	Postbind(pc *PluginContext, p *v1.Pod, nodeName string)
+}
+
 // UnreservePlugin is an interface for Unreserve plugins. This is an informational
 // extension point. If a pod was reserved and then rejected in a later phase, then
 // un-reserve plugins will be notified. Un-reserve plugins should clean up state
@@ -186,6 +197,9 @@ type Framework interface {
 	// internal error. In either case the pod is not going to be bound.
 	RunPrebindPlugins(pc *PluginContext, pod *v1.Pod, nodeName string) *Status
 
+	// RunPostbindPlugins runs the set of configured postbind plugins.
+	RunPostbindPlugins(pc *PluginContext, pod *v1.Pod, nodeName string)
+
 	// RunReservePlugins runs the set of configured reserve plugins. If any of these
 	// plugins returns an error, it does not continue running the remaining ones and
 	// returns the error. In such case, pod will not be scheduled.

pkg/scheduler/internal/queue/scheduling_queue_test.go

@@ -171,6 +171,8 @@ func (*fakeFramework) RunPrebindPlugins(pc *framework.PluginContext, pod *v1.Pod
 	return nil
 }
 
+func (*fakeFramework) RunPostbindPlugins(pc *framework.PluginContext, pod *v1.Pod, nodeName string) {}
+
 func (*fakeFramework) RunReservePlugins(pc *framework.PluginContext, pod *v1.Pod, nodeName string) *framework.Status {
 	return nil
 }

pkg/scheduler/scheduler.go

@@ -592,6 +592,9 @@ func (sched *Scheduler) scheduleOne() {
 		} else {
 			klog.V(2).Infof("pod %v/%v is bound successfully on node %v, %d nodes evaluated, %d nodes were found feasible", assumedPod.Namespace, assumedPod.Name, scheduleResult.SuggestedHost, scheduleResult.EvaluatedNodes, scheduleResult.FeasibleNodes)
 			metrics.PodScheduleSuccesses.Inc()
+
+			// Run "postbind" plugins.
+			fwk.RunPostbindPlugins(pluginContext, assumedPod, scheduleResult.SuggestedHost)
 		}
 	}()
 }

test/integration/scheduler/framework_test.go

@@ -33,6 +33,7 @@ import (
 type TesterPlugin struct {
 	numReserveCalled    int
 	numPrebindCalled    int
+	numPostbindCalled   int
 	numUnreserveCalled  int
 	failReserve         bool
 	failPrebind         bool
@@ -53,6 +54,10 @@ type PrebindPlugin struct {
 	TesterPlugin
 }
 
+type PostbindPlugin struct {
+	TesterPlugin
+}
+
 type UnreservePlugin struct {
 	TesterPlugin
 }
@@ -66,11 +71,13 @@ const (
 	reservePluginName   = "reserve-plugin"
 	prebindPluginName   = "prebind-plugin"
 	unreservePluginName = "unreserve-plugin"
+	postbindPluginName  = "postbind-plugin"
 	permitPluginName    = "permit-plugin"
 )
 
 var _ = framework.ReservePlugin(&ReservePlugin{})
 var _ = framework.PrebindPlugin(&PrebindPlugin{})
+var _ = framework.PostbindPlugin(&PostbindPlugin{})
 var _ = framework.UnreservePlugin(&UnreservePlugin{})
 var _ = framework.PermitPlugin(&PermitPlugin{})
 
@@ -125,6 +132,28 @@ func NewPrebindPlugin(_ *runtime.Unknown, _ framework.FrameworkHandle) (framewor
 	return pbdPlugin, nil
 }
 
+var ptbdPlugin = &PostbindPlugin{}
+
+// Name returns name of the plugin.
+func (pp *PostbindPlugin) Name() string {
+	return postbindPluginName
+}
+
+// Postbind is a test function, which counts the number of times called.
+func (pp *PostbindPlugin) Postbind(pc *framework.PluginContext, pod *v1.Pod, nodeName string) {
+	pp.numPostbindCalled++
+}
+
+// reset used to reset numPostbindCalled.
+func (pp *PostbindPlugin) reset() {
+	pp.numPostbindCalled = 0
+}
+
+// NewPostbindPlugin is the factory for postbind plugin.
+func NewPostbindPlugin(_ *runtime.Unknown, _ framework.FrameworkHandle) (framework.Plugin, error) {
+	return ptbdPlugin, nil
+}
+
 var unresPlugin = &UnreservePlugin{}
 
 // Name returns name of the plugin.
@@ -459,6 +488,120 @@ func TestUnreservePlugin(t *testing.T) {
 	}
 }
 
+// TestPostbindPlugin tests invocation of postbind plugins.
+func TestPostbindPlugin(t *testing.T) {
+	// Create a plugin registry for testing. Register a prebind and a postbind plugin.
+	registry := framework.Registry{
+		prebindPluginName:  NewPrebindPlugin,
+		postbindPluginName: NewPostbindPlugin,
+	}
+
+	// Setup initial prebind and postbind plugin for testing.
+	plugins := &schedulerconfig.Plugins{
+		PreBind: &schedulerconfig.PluginSet{
+			Enabled: []schedulerconfig.Plugin{
+				{
+					Name: prebindPluginName,
+				},
+			},
+		},
+		PostBind: &schedulerconfig.PluginSet{
+			Enabled: []schedulerconfig.Plugin{
+				{
+					Name: postbindPluginName,
+				},
+			},
+		},
+	}
+	// Set reserve prebind and postbind config for testing
+	pluginConfig := []schedulerconfig.PluginConfig{
+		{
+			Name: prebindPluginName,
+			Args: runtime.Unknown{},
+		},
+		{
+			Name: postbindPluginName,
+			Args: runtime.Unknown{},
+		},
+	}
+
+	// Create the master and the scheduler with the test plugin set.
+	context := initTestSchedulerWithOptions(t,
+		initTestMaster(t, "postbind-plugin", nil),
+		false, nil, registry, plugins, pluginConfig, false, time.Second)
+	defer cleanupTest(t, context)
+
+	cs := context.clientSet
+	// Add a few nodes.
+	_, err := createNodes(cs, "test-node", nil, 2)
+	if err != nil {
+		t.Fatalf("Cannot create nodes: %v", err)
+	}
+
+	tests := []struct {
+		prebindFail   bool
+		prebindReject bool
+	}{
+		{
+			prebindFail:   false,
+			prebindReject: false,
+		},
+		{
+			prebindFail:   true,
+			prebindReject: false,
+		},
+		{
+			prebindFail:   false,
+			prebindReject: true,
+		},
+		{
+			prebindFail:   true,
+			prebindReject: true,
+		},
+	}
+
+	for i, test := range tests {
+		pbdPlugin.failPrebind = test.prebindFail
+		pbdPlugin.rejectPrebind = test.prebindReject
+
+		// Create a best effort pod.
+		pod, err := createPausePod(cs,
+			initPausePod(cs, &pausePodConfig{Name: "test-pod", Namespace: context.ns.Name}))
+		if err != nil {
+			t.Errorf("Error while creating a test pod: %v", err)
+		}
+
+		if test.prebindFail {
+			if err = wait.Poll(10*time.Millisecond, 30*time.Second, podSchedulingError(cs, pod.Namespace, pod.Name)); err != nil {
+				t.Errorf("test #%v: Expected a scheduling error, but didn't get it. error: %v", i, err)
+			}
+			if ptbdPlugin.numPostbindCalled > 0 {
+				t.Errorf("test #%v: Didn't expected the postbind plugin to be called %d times.", i, ptbdPlugin.numPostbindCalled)
+			}
+		} else {
+			if test.prebindReject {
+				if err = waitForPodUnschedulable(cs, pod); err != nil {
+					t.Errorf("test #%v: Didn't expected the pod to be scheduled. error: %v", i, err)
+				}
+				if ptbdPlugin.numPostbindCalled > 0 {
+					t.Errorf("test #%v: Didn't expected the postbind plugin to be called %d times.", i, ptbdPlugin.numPostbindCalled)
+				}
+			} else {
+				if err = waitForPodToSchedule(cs, pod); err != nil {
+					t.Errorf("test #%v: Expected the pod to be scheduled. error: %v", i, err)
+				}
+				if ptbdPlugin.numPostbindCalled == 0 {
+					t.Errorf("test #%v: Expected the postbind plugin to be called, was called %d times.", i, ptbdPlugin.numPostbindCalled)
+				}
+			}
+		}
+
+		ptbdPlugin.reset()
+		pbdPlugin.reset()
+		cleanupPods(cs, t, []*v1.Pod{pod})
+	}
+}
+
 // TestPermitPlugin tests invocation of permit plugins.
 func TestPermitPlugin(t *testing.T) {
 	// Create a plugin registry for testing. Register only a permit plugin.