github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-22 03:11:40 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #99498 from gavinfish/sched-preempthandle

Browse Source 
Scheduler: unroll PreemptHandle to Handle
This commit is contained in:
Kubernetes Prow Robot 2021-03-01 13:58:28 -08:00 committed by GitHub
commit 0ced9d2854
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 28 additions and 53 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/scheduler/core/generic_scheduler_test.go
View File

@ -1173,7 +1173,7 @@ func TestFindFitPredicateCallCounts(t *testing.T) {
if err := scheduler.cache.UpdateSnapshot(scheduler.nodeInfoSnapshot); err != nil {
t.Fatal(err)
}
fwk.PreemptHandle().AddNominatedPod(framework.NewPodInfo(&v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "nominated"}, Spec: v1.PodSpec{Priority: &midPriority}}), "1")
fwk.AddNominatedPod(framework.NewPodInfo(&v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: "nominated"}, Spec: v1.PodSpec{Priority: &midPriority}}), "1")
_, _, err = scheduler.findNodesThatFitPod(context.Background(), fwk, framework.NewCycleState(), test.pod)

18
pkg/scheduler/framework/interface.go
View File

@ -552,6 +552,10 @@ type Framework interface {
// passed to the plugin factories at the time of plugin initialization. Plugins
// must store and use this handle to call framework functions.
type Handle interface {
// PodNominator abstracts operations to maintain nominated Pods.
PodNominator
// PluginsRunner abstracts operations to run some plugins.
PluginsRunner
// SnapshotSharedLister returns listers from the latest NodeInfo Snapshot. The snapshot
// is taken at the beginning of a scheduling cycle and remains unchanged until
// a pod finishes "Permit" point. There is no guarantee that the information
@ -581,8 +585,8 @@ type Handle interface {
// RunFilterPluginsWithNominatedPods runs the set of configured filter plugins for nominated pod on the given node.
RunFilterPluginsWithNominatedPods(ctx context.Context, state *CycleState, pod *v1.Pod, info *NodeInfo) *Status
// TODO: unroll the wrapped interfaces to Handle.
PreemptHandle() PreemptHandle
// Extenders returns registered scheduler extenders.
Extenders() []Extender
}
// PostFilterResult wraps needed info for scheduler framework to act upon PostFilter phase.
@ -590,16 +594,6 @@ type PostFilterResult struct {
NominatedNodeName string
}
// PreemptHandle incorporates all needed logic to run preemption logic.
type PreemptHandle interface {
// PodNominator abstracts operations to maintain nominated Pods.
PodNominator
// PluginsRunner abstracts operations to run some plugins.
PluginsRunner
// Extenders returns registered scheduler extenders.
Extenders() []Extender
}
// PodNominator abstracts operations to maintain nominated Pods.
type PodNominator interface {
// AddNominatedPod adds the given pod to the nominated pod map or

10
pkg/scheduler/framework/plugins/defaultpreemption/default_preemption.go
View File

@ -116,7 +116,6 @@ func (pl *DefaultPreemption) PostFilter(ctx context.Context, state *framework.Cy
// before it is retried after many other pending pods.
func (pl *DefaultPreemption) preempt(ctx context.Context, state *framework.CycleState, pod *v1.Pod, m framework.NodeToStatusMap) (string, *framework.Status) {
cs := pl.fh.ClientSet()
ph := pl.fh.PreemptHandle()
nodeLister := pl.fh.SnapshotSharedLister().NodeInfos()
// 0) Fetch the latest version of <pod>.
@ -156,7 +155,7 @@ func (pl *DefaultPreemption) preempt(ctx context.Context, state *framework.Cycle
}
// 3) Interact with registered Extenders to filter out some candidates if needed.
candidates, status = CallExtenders(ph.Extenders(), pod, nodeLister, candidates)
candidates, status = CallExtenders(pl.fh.Extenders(), pod, nodeLister, candidates)
if !status.IsSuccess() {
return "", status
}
@ -606,12 +605,11 @@ func selectVictimsOnNode(
pdbs []*policy.PodDisruptionBudget,
) ([]*v1.Pod, int, *framework.Status) {
var potentialVictims []*framework.PodInfo
ph := fh.PreemptHandle()
removePod := func(rpi *framework.PodInfo) error {
if err := nodeInfo.RemovePod(rpi.Pod); err != nil {
return err
}
status := ph.RunPreFilterExtensionRemovePod(ctx, state, pod, rpi, nodeInfo)
status := fh.RunPreFilterExtensionRemovePod(ctx, state, pod, rpi, nodeInfo)
if !status.IsSuccess() {
return status.AsError()
}
@ -619,7 +617,7 @@ func selectVictimsOnNode(
}
addPod := func(api *framework.PodInfo) error {
nodeInfo.AddPodInfo(api)
status := ph.RunPreFilterExtensionAddPod(ctx, state, pod, api, nodeInfo)
status := fh.RunPreFilterExtensionAddPod(ctx, state, pod, api, nodeInfo)
if !status.IsSuccess() {
return status.AsError()
}
@ -714,7 +712,7 @@ func PrepareCandidate(c Candidate, fh framework.Handle, cs kubernetes.Interface,
// this node. So, we should remove their nomination. Removing their
// nomination updates these pods and moves them to the active queue. It
// lets scheduler find another place for them.
nominatedPods := getLowerPriorityNominatedPods(fh.PreemptHandle(), pod, c.Name())
nominatedPods := getLowerPriorityNominatedPods(fh, pod, c.Name())
if err := util.ClearNominatedNodeName(cs, nominatedPods...); err != nil {
klog.ErrorS(err, "cannot clear 'NominatedNodeName' field")
// We do not return as this error is not critical.

46
pkg/scheduler/framework/runtime/framework.go
View File

@ -88,7 +88,8 @@ type frameworkImpl struct {
metricsRecorder *metricsRecorder
profileName string
preemptHandle framework.PreemptHandle
extenders []framework.Extender
framework.PodNominator
// Indicates that RunFilterPlugins should accumulate all failed statuses and not return
// after the first failure.
@ -122,6 +123,11 @@ func (f *frameworkImpl) getExtensionPoints(plugins *config.Plugins) []extensionP
}
}
// Extenders returns the registered extenders.
func (f *frameworkImpl) Extenders() []framework.Extender {
return f.extenders
}
type frameworkOptions struct {
clientSet clientset.Interface
eventRecorder events.EventRecorder
@ -218,19 +224,6 @@ func defaultFrameworkOptions() frameworkOptions {
}
}
var _ framework.PreemptHandle = &preemptHandle{}
type preemptHandle struct {
extenders []framework.Extender
framework.PodNominator
framework.PluginsRunner
}
// Extenders returns the registered extenders.
func (ph *preemptHandle) Extenders() []framework.Extender {
return ph.extenders
}
var _ framework.Framework = &frameworkImpl{}
// NewFramework initializes plugins given the configuration and the registry.
@ -251,11 +244,8 @@ func NewFramework(r Registry, plugins *config.Plugins, args []config.PluginConfi
metricsRecorder: options.metricsRecorder,
profileName: options.profileName,
runAllFilters: options.runAllFilters,
}
f.preemptHandle = &preemptHandle{
extenders: options.extenders,
PodNominator: options.podNominator,
PluginsRunner: f,
extenders: options.extenders,
PodNominator: options.podNominator,
}
if plugins == nil {
return f, nil
@ -605,7 +595,6 @@ func (f *frameworkImpl) runPostFilterPlugin(ctx context.Context, pl framework.Po
func (f *frameworkImpl) RunFilterPluginsWithNominatedPods(ctx context.Context, state *framework.CycleState, pod *v1.Pod, info *framework.NodeInfo) *framework.Status {
var status *framework.Status
ph := f.PreemptHandle()
podsAdded := false
// We run filters twice in some cases. If the node has greater or equal priority
// nominated pods, we run them when those pods are added to PreFilter state and nodeInfo.
@ -630,7 +619,7 @@ func (f *frameworkImpl) RunFilterPluginsWithNominatedPods(ctx context.Context, s
nodeInfoToUse := info
if i == 0 {
var err error
podsAdded, stateToUse, nodeInfoToUse, err = addNominatedPods(ctx, ph, pod, state, info)
podsAdded, stateToUse, nodeInfoToUse, err = addNominatedPods(ctx, f, pod, state, info)
if err != nil {
return framework.AsStatus(err)
}
@ -638,7 +627,7 @@ func (f *frameworkImpl) RunFilterPluginsWithNominatedPods(ctx context.Context, s
break
}
statusMap := ph.RunFilterPlugins(ctx, stateToUse, pod, nodeInfoToUse)
statusMap := f.RunFilterPlugins(ctx, stateToUse, pod, nodeInfoToUse)
status = statusMap.Merge()
if !status.IsSuccess() && !status.IsUnschedulable() {
return status
@ -651,12 +640,12 @@ func (f *frameworkImpl) RunFilterPluginsWithNominatedPods(ctx context.Context, s
// addNominatedPods adds pods with equal or greater priority which are nominated
// to run on the node. It returns 1) whether any pod was added, 2) augmented cycleState,
// 3) augmented nodeInfo.
func addNominatedPods(ctx context.Context, ph framework.PreemptHandle, pod *v1.Pod, state *framework.CycleState, nodeInfo *framework.NodeInfo) (bool, *framework.CycleState, *framework.NodeInfo, error) {
if ph == nil || nodeInfo.Node() == nil {
func addNominatedPods(ctx context.Context, fh framework.Handle, pod *v1.Pod, state *framework.CycleState, nodeInfo *framework.NodeInfo) (bool, *framework.CycleState, *framework.NodeInfo, error) {
if fh == nil || nodeInfo.Node() == nil {
// This may happen only in tests.
return false, state, nodeInfo, nil
}
nominatedPodInfos := ph.NominatedPodsForNode(nodeInfo.Node().Name)
nominatedPodInfos := fh.NominatedPodsForNode(nodeInfo.Node().Name)
if len(nominatedPodInfos) == 0 {
return false, state, nodeInfo, nil
}
@ -666,7 +655,7 @@ func addNominatedPods(ctx context.Context, ph framework.PreemptHandle, pod *v1.P
for _, pi := range nominatedPodInfos {
if corev1.PodPriority(pi.Pod) >= corev1.PodPriority(pod) && pi.Pod.UID != pod.UID {
nodeInfoOut.AddPodInfo(pi)
status := ph.RunPreFilterExtensionAddPod(ctx, stateOut, pod, pi, nodeInfoOut)
status := fh.RunPreFilterExtensionAddPod(ctx, stateOut, pod, pi, nodeInfoOut)
if !status.IsSuccess() {
return false, state, nodeInfo, status.AsError()
}
@ -1133,11 +1122,6 @@ func (f *frameworkImpl) pluginsNeeded(plugins *config.Plugins) map[string]config
return pgMap
}
// PreemptHandle returns the internal preemptHandle object.
func (f *frameworkImpl) PreemptHandle() framework.PreemptHandle {
return f.preemptHandle
}
// ProfileName returns the profile name associated to this framework.
func (f *frameworkImpl) ProfileName() string {
return f.profileName

5
test/integration/scheduler/framework_test.go
View File

@ -409,15 +409,14 @@ func (pp *PostFilterPlugin) PostFilter(ctx context.Context, state *framework.Cyc
return nil, framework.NewStatus(framework.Error, err.Error())
}
ph := pp.fh.PreemptHandle()
for _, nodeInfo := range nodeInfos {
ph.RunFilterPlugins(ctx, state, pod, nodeInfo)
pp.fh.RunFilterPlugins(ctx, state, pod, nodeInfo)
}
var nodes []*v1.Node
for _, nodeInfo := range nodeInfos {
nodes = append(nodes, nodeInfo.Node())
}
ph.RunScorePlugins(ctx, state, pod, nodes)
pp.fh.RunScorePlugins(ctx, state, pod, nodes)
if pp.failPostFilter {
return nil, framework.NewStatus(framework.Error, fmt.Sprintf("injecting failure for pod %v", pod.Name))