diff --git a/pkg/scheduler/framework/interface.go b/pkg/scheduler/framework/interface.go index f416d9359fe..5852ce49b99 100644 --- a/pkg/scheduler/framework/interface.go +++ b/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 diff --git a/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources.go b/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources.go index 77c90014ff7..259748eadfc 100644 --- a/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources.go +++ b/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. diff --git a/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go b/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go index 0459af16db6..4ac6b5d6a11 100644 --- a/pkg/scheduler/framework/plugins/dynamicresources/dynamicresources_test.go +++ b/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) diff --git a/pkg/scheduler/framework/plugins/examples/multipoint/multipoint.go b/pkg/scheduler/framework/plugins/examples/multipoint/multipoint.go index 5d6b311c319..12179715835 100644 --- a/pkg/scheduler/framework/plugins/examples/multipoint/multipoint.go +++ b/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 { diff --git a/pkg/scheduler/framework/plugins/examples/prebind/prebind.go b/pkg/scheduler/framework/plugins/examples/prebind/prebind.go index a3247b5433f..8fffd90cb0d 100644 --- a/pkg/scheduler/framework/plugins/examples/prebind/prebind.go +++ b/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 { diff --git a/pkg/scheduler/framework/plugins/examples/stateful/stateful.go b/pkg/scheduler/framework/plugins/examples/stateful/stateful.go index dd0e6aae377..3607081e66e 100644 --- a/pkg/scheduler/framework/plugins/examples/stateful/stateful.go +++ b/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 { diff --git a/pkg/scheduler/framework/plugins/volumebinding/volume_binding.go b/pkg/scheduler/framework/plugins/volumebinding/volume_binding.go index 945c8e685d2..7a9b52bc70a 100644 --- a/pkg/scheduler/framework/plugins/volumebinding/volume_binding.go +++ b/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)) diff --git a/pkg/scheduler/framework/plugins/volumebinding/volume_binding_test.go b/pkg/scheduler/framework/plugins/volumebinding/volume_binding_test.go index fc9f60b8915..454bdb3fbbe 100644 --- a/pkg/scheduler/framework/plugins/volumebinding/volume_binding_test.go +++ b/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 diff --git a/pkg/scheduler/framework/runtime/framework_test.go b/pkg/scheduler/framework/runtime/framework_test.go index 012fe78b2a7..b3c91c57f2f 100644 --- a/pkg/scheduler/framework/runtime/framework_test.go +++ b/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) { } diff --git a/pkg/scheduler/testing/framework/fake_plugins.go b/pkg/scheduler/testing/framework/fake_plugins.go index 5e4d7eff61f..de8bd3ab018 100644 --- a/pkg/scheduler/testing/framework/fake_plugins.go +++ b/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 diff --git a/test/integration/scheduler/plugins/plugins_test.go b/test/integration/scheduler/plugins/plugins_test.go index 0e4f2031f63..ba6b70d32af 100644 --- a/test/integration/scheduler/plugins/plugins_test.go +++ b/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() diff --git a/test/integration/scheduler/preemption/preemption_test.go b/test/integration/scheduler/preemption/preemption_test.go index cb488f6d230..acac998f8c0 100644 --- a/test/integration/scheduler/preemption/preemption_test.go +++ b/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