feat: add PreBindPreFlight and implement in in-tree plugins

pkg/scheduler/framework/interface.go

@@ -432,6 +432,12 @@ type ReservePlugin interface {
 // These plugins are called before a pod being scheduled.
 type PreBindPlugin interface {
 	Plugin
+	// PreBindPreFlight is called before PreBind, and the plugin is supposed to return Success, Skip, or Error status.
+	// If it returns Success, it means this PreBind plugin will handle this pod.
+	// If it returns Skip, it means this PreBind plugin has nothing to do with the pod, and PreBind will be skipped.
+	// This function should be lightweight, and shouldn't do any actual operation, e.g., creating a volume etc.
+	PreBindPreFlight(ctx context.Context, state fwk.CycleState, p *v1.Pod, nodeName string) *fwk.Status
+
 	// PreBind is called before binding a pod. All prebind plugins must return
 	// success or the pod will be rejected and won't be sent for binding.
 	PreBind(ctx context.Context, state fwk.CycleState, p *v1.Pod, nodeName string) *fwk.Status

pkg/scheduler/framework/plugins/dynamicresources/dynamicresources.go

@@ -797,6 +797,22 @@ func (pl *DynamicResources) PreBind(ctx context.Context, cs fwk.CycleState, pod
 	return nil
 }
 
+// PreBindPreFlight is called before PreBind, and determines whether PreBind is going to do something for this pod, or not.
+// It just checks state.claims to determine whether there are any claims and hence the plugin has to handle them at PreBind.
+func (pl *DynamicResources) PreBindPreFlight(ctx context.Context, cs fwk.CycleState, p *v1.Pod, nodeName string) *fwk.Status {
+	if !pl.enabled {
+		return fwk.NewStatus(fwk.Skip)
+	}
+	state, err := getStateData(cs)
+	if err != nil {
+		return statusError(klog.FromContext(ctx), err)
+	}
+	if len(state.claims) == 0 {
+		return fwk.NewStatus(fwk.Skip)
+	}
+	return nil
+}
+
 // bindClaim gets called by PreBind for claim which is not reserved for the pod yet.
 // It might not even be allocated. bindClaim then ensures that the allocation
 // and reservation are recorded. This finishes the work started in Reserve.

pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go

@@ -310,6 +310,7 @@ type want struct {
 	prescore         result
 	reserve          result
 	unreserve        result
+	prebindPreFlight *fwk.Status
 	prebind          result
 	postbind         result
 	postFilterResult *framework.PostFilterResult
@@ -370,6 +371,7 @@ func TestPlugin(t *testing.T) {
 				postfilter: result{
 					status: fwk.NewStatus(fwk.Unschedulable),
 				},
+				prebindPreFlight: fwk.NewStatus(fwk.Skip),
 			},
 		},
 		"claim-reference": {
@@ -907,6 +909,7 @@ func TestPlugin(t *testing.T) {
 				postfilter: result{
 					status: fwk.NewStatus(fwk.Unschedulable, `plugin disabled`),
 				},
+				prebindPreFlight: fwk.NewStatus(fwk.Skip),
 			},
 			disableDRA: true,
 		},
@@ -1007,7 +1010,7 @@ func TestPlugin(t *testing.T) {
 				assert.Equal(t, tc.want.preFilterResult, result)
 				testCtx.verify(t, tc.want.prefilter, initialObjects, result, status)
 			})
-			unschedulable := status.Code() != fwk.Success
+			unschedulable := status.IsRejected()
 
 			var potentialNodes []*framework.NodeInfo
 
@@ -1074,11 +1077,14 @@ func TestPlugin(t *testing.T) {
 
 					initialObjects = testCtx.listAll(t)
 					initialObjects = testCtx.updateAPIServer(t, initialObjects, tc.prepare.prebind)
-					status := testCtx.p.PreBind(testCtx.ctx, testCtx.state, tc.pod, selectedNode.Node().Name)
-					t.Run("prebind", func(t *testing.T) {
-						testCtx.verify(t, tc.want.prebind, initialObjects, nil, status)
+					preBindPreFlightStatus := testCtx.p.PreBindPreFlight(testCtx.ctx, testCtx.state, tc.pod, selectedNode.Node().Name)
+					t.Run("prebindPreFlight", func(t *testing.T) {
+						assert.Equal(t, tc.want.prebindPreFlight, preBindPreFlightStatus)
+					})
+					preBindStatus := testCtx.p.PreBind(testCtx.ctx, testCtx.state, tc.pod, selectedNode.Node().Name)
+					t.Run("prebind", func(t *testing.T) {
+						testCtx.verify(t, tc.want.prebind, initialObjects, nil, preBindStatus)
 					})
-
 					if tc.want.unreserveAfterBindFailure != nil {
 						initialObjects = testCtx.listAll(t)
 						testCtx.p.Unreserve(testCtx.ctx, testCtx.state, tc.pod, selectedNode.Node().Name)

pkg/scheduler/framework/plugins/examples/multipoint/multipoint.go

@@ -73,6 +73,10 @@ func (mc CommunicatingPlugin) Unreserve(ctx context.Context, state fwk.CycleStat
 	}
 }
 
+func (mc CommunicatingPlugin) PreBindPreFlight(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {
+	return nil
+}
+
 // PreBind is the function invoked by the framework at "prebind" extension point.
 func (mc CommunicatingPlugin) PreBind(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {
 	if pod == nil {

pkg/scheduler/framework/plugins/examples/prebind/prebind.go

@@ -39,6 +39,10 @@ func (sr StatelessPreBindExample) Name() string {
 	return Name
 }
 
+func (mc StatelessPreBindExample) PreBindPreFlight(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {
+	return nil
+}
+
 // PreBind is the functions invoked by the framework at "prebind" extension point.
 func (sr StatelessPreBindExample) PreBind(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {
 	if pod == nil {

pkg/scheduler/framework/plugins/examples/stateful/stateful.go

@@ -69,6 +69,10 @@ func (mp *MultipointExample) Unreserve(ctx context.Context, state fwk.CycleState
 	mp.executionPoints = nil
 }
 
+func (mp *MultipointExample) PreBindPreFlight(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {
+	return nil
+}
+
 // PreBind is the function invoked by the framework at "prebind" extension
 // point.
 func (mp *MultipointExample) PreBind(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {

pkg/scheduler/framework/plugins/volumebinding/volume_binding.go

@@ -541,6 +541,26 @@ func (pl *VolumeBinding) Reserve(ctx context.Context, cs fwk.CycleState, pod *v1
 	return nil
 }
 
+var errNoPodVolumeForNode = fmt.Errorf("no pod volume found for node")
+
+// PreBindPreFlight is called before PreBind, and determines whether PreBind is going to do something for this pod, or not.
+// It checks state.podVolumesByNode to determine whether there are any pod volumes for the node and hence the plugin has to handle them at PreBind.
+func (pl *VolumeBinding) PreBindPreFlight(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {
+	s, err := getStateData(state)
+	if err != nil {
+		return fwk.AsStatus(err)
+	}
+	if s.allBound {
+		// no need to bind volumes
+		return fwk.NewStatus(fwk.Skip)
+	}
+
+	if _, ok := s.podVolumesByNode[nodeName]; !ok {
+		return fwk.AsStatus(fmt.Errorf("%w %q", errNoPodVolumeForNode, nodeName))
+	}
+	return nil
+}
+
 // PreBind will make the API update with the assumed bindings and wait until
 // the PV controller has completely finished the binding operation.
 //
@@ -558,7 +578,7 @@ func (pl *VolumeBinding) PreBind(ctx context.Context, cs fwk.CycleState, pod *v1
 	// we don't need to hold the lock as only one node will be pre-bound for the given pod
 	podVolumes, ok := s.podVolumesByNode[nodeName]
 	if !ok {
-		return fwk.AsStatus(fmt.Errorf("no pod volumes found for node %q", nodeName))
+		return fwk.AsStatus(fmt.Errorf("%w %q", errNoPodVolumeForNode, nodeName))
 	}
 	logger := klog.FromContext(ctx)
 	logger.V(5).Info("Trying to bind volumes for pod", "pod", klog.KObj(pod))

pkg/scheduler/framework/plugins/volumebinding/volume_binding_test.go

@@ -1131,6 +1131,64 @@ func TestVolumeBinding(t *testing.T) {
 	}
 }
 
+func Test_PreBindPreFlight(t *testing.T) {
+	table := []struct {
+		name     string
+		nodeName string
+		state    *stateData
+		want     *fwk.Status
+	}{
+		{
+			name:     "all bound",
+			nodeName: "node-a",
+			state: &stateData{
+				allBound: true,
+			},
+			want: fwk.NewStatus(fwk.Skip),
+		},
+		{
+			name:     "volume to be bound",
+			nodeName: "node-a",
+			state: &stateData{
+				podVolumesByNode: map[string]*PodVolumes{
+					"node-a": {},
+				},
+			},
+			want: fwk.NewStatus(fwk.Success),
+		},
+		{
+			name:     "error: state is nil",
+			nodeName: "node-a",
+			want:     fwk.AsStatus(fwk.ErrNotFound),
+		},
+		{
+			name:     "error: node is not found in podVolumesByNode",
+			nodeName: "node-a",
+			state: &stateData{
+				podVolumesByNode: map[string]*PodVolumes{
+					"node-b": {},
+				},
+			},
+			want: fwk.AsStatus(errNoPodVolumeForNode),
+		},
+	}
+
+	for _, item := range table {
+		t.Run(item.name, func(t *testing.T) {
+			pl := &VolumeBinding{}
+			_, ctx := ktesting.NewTestContext(t)
+			state := framework.NewCycleState()
+			if item.state != nil {
+				state.Write(stateKey, item.state)
+			}
+			status := pl.PreBindPreFlight(ctx, state, &v1.Pod{}, item.nodeName)
+			if !status.Equal(item.want) {
+				t.Errorf("PreBindPreFlight failed - got: %v, want: %v", status, item.want)
+			}
+		})
+	}
+}
+
 func TestIsSchedulableAfterCSINodeChange(t *testing.T) {
 	table := []struct {
 		name   string

pkg/scheduler/framework/runtime/framework_test.go

@@ -239,6 +239,10 @@ func (pl *TestPlugin) PreBind(ctx context.Context, state fwk.CycleState, p *v1.P
 	return fwk.NewStatus(fwk.Code(pl.inj.PreBindStatus), injectReason)
 }
 
+func (pl *TestPlugin) PreBindPreFlight(ctx context.Context, state fwk.CycleState, p *v1.Pod, nodeName string) *fwk.Status {
+	return nil
+}
+
 func (pl *TestPlugin) PostBind(ctx context.Context, state fwk.CycleState, p *v1.Pod, nodeName string) {
 }
 

pkg/scheduler/testing/framework/fake_plugins.go

@@ -207,6 +207,11 @@ func (pl *FakePreBindPlugin) Name() string {
 	return "FakePreBind"
 }
 
+// PreBindPreFlight invoked at the PreBind extension point.
+func (pl *FakePreBindPlugin) PreBindPreFlight(_ context.Context, _ fwk.CycleState, _ *v1.Pod, _ string) *fwk.Status {
+	return pl.Status
+}
+
 // PreBind invoked at the PreBind extension point.
 func (pl *FakePreBindPlugin) PreBind(_ context.Context, _ fwk.CycleState, _ *v1.Pod, _ string) *fwk.Status {
 	return pl.Status

test/integration/scheduler/plugins/plugins_test.go

@@ -492,6 +492,11 @@ func (pp *PreBindPlugin) Name() string {
 	return preBindPluginName
 }
 
+// PreBindPreFlight is a test function that returns nil for testing.
+func (pp *PreBindPlugin) PreBindPreFlight(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {
+	return nil
+}
+
 // PreBind is a test function that returns (true, nil) or errors for testing.
 func (pp *PreBindPlugin) PreBind(ctx context.Context, state fwk.CycleState, pod *v1.Pod, nodeName string) *fwk.Status {
 	pp.mutex.Lock()

test/integration/scheduler/preemption/preemption_test.go

@@ -1512,6 +1512,10 @@ func (af *alwaysFail) Name() string {
 	return alwaysFailPlugin
 }
 
+func (af *alwaysFail) PreBindPreFlight(_ context.Context, _ fwk.CycleState, p *v1.Pod, _ string) *fwk.Status {
+	return nil
+}
+
 func (af *alwaysFail) PreBind(_ context.Context, _ fwk.CycleState, p *v1.Pod, _ string) *fwk.Status {
 	if strings.Contains(p.Name, doNotFailMe) {
 		return nil