github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-27 21:47:07 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #83365 from notpad/feature/83244

Browse Source 
A consistent interface for plugin extended functionality
This commit is contained in:
Kubernetes Prow Robot 2019-10-01 13:01:11 -07:00 committed by GitHub
commit dc0013a506
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 85 additions and 72 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
pkg/scheduler/core/generic_scheduler.go
View File

@ -1120,7 +1120,7 @@ func (g *genericScheduler) selectVictimsOnNode(
return err return err
} }
} }
status := g.framework.RunPreFilterUpdaterRemovePod(pluginContext, pod, rp, nodeInfoCopy) status := g.framework.RunPreFilterExtensionRemovePod(pluginContext, pod, rp, nodeInfoCopy)
if !status.IsSuccess() { if !status.IsSuccess() {
return status.AsError() return status.AsError()
} }
@ -1133,7 +1133,7 @@ func (g *genericScheduler) selectVictimsOnNode(
return err return err
} }
} }
status := g.framework.RunPreFilterUpdaterAddPod(pluginContext, pod, ap, nodeInfoCopy) status := g.framework.RunPreFilterExtensionAddPod(pluginContext, pod, ap, nodeInfoCopy)
if !status.IsSuccess() { if !status.IsSuccess() {
return status.AsError() return status.AsError()
} }

2
pkg/scheduler/factory/factory_test.go
View File

@ -635,7 +635,7 @@ func (t *TestPlugin) Score(pc *framework.PluginContext, p *v1.Pod, nodeName stri
return 1, nil return 1, nil
} }
func (t *TestPlugin) NormalizeScore(pc *framework.PluginContext, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status { func (t *TestPlugin) Extensions() framework.ScoreExtensions {
return nil return nil
} }

30
pkg/scheduler/framework/v1alpha1/framework.go
View File

@ -297,42 +297,42 @@ func (f *framework) RunPreFilterPlugins(
return nil return nil
} }
// RunPreFilterUpdaterAddPod calls the AddPod interface for the set of configured // RunPreFilterExtensionAddPod calls the AddPod interface for the set of configured
// PreFilter plugins. It returns directly if any of the plugins return any // PreFilter plugins. It returns directly if any of the plugins return any
// status other than Success. // status other than Success.
func (f *framework) RunPreFilterUpdaterAddPod(pc *PluginContext, podToSchedule *v1.Pod, func (f *framework) RunPreFilterExtensionAddPod(pc *PluginContext, podToSchedule *v1.Pod,
podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status { podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status {
for _, pl := range f.preFilterPlugins { for _, pl := range f.preFilterPlugins {
if updater := pl.Updater(); updater != nil { if pl.Extensions() == nil {
status := updater.AddPod(pc, podToSchedule, podToAdd, nodeInfo) continue
if !status.IsSuccess() { }
if status := pl.Extensions().AddPod(pc, podToSchedule, podToAdd, nodeInfo); !status.IsSuccess() {
msg := fmt.Sprintf("error while running AddPod for plugin %q while scheduling pod %q: %v", msg := fmt.Sprintf("error while running AddPod for plugin %q while scheduling pod %q: %v",
pl.Name(), podToSchedule.Name, status.Message()) pl.Name(), podToSchedule.Name, status.Message())
klog.Error(msg) klog.Error(msg)
return NewStatus(Error, msg) return NewStatus(Error, msg)
} }
} }
}
return nil return nil
} }
// RunPreFilterUpdaterRemovePod calls the RemovePod interface for the set of configured // RunPreFilterExtensionRemovePod calls the RemovePod interface for the set of configured
// PreFilter plugins. It returns directly if any of the plugins return any // PreFilter plugins. It returns directly if any of the plugins return any
// status other than Success. // status other than Success.
func (f *framework) RunPreFilterUpdaterRemovePod(pc *PluginContext, podToSchedule *v1.Pod, func (f *framework) RunPreFilterExtensionRemovePod(pc *PluginContext, podToSchedule *v1.Pod,
podToRemove *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status { podToRemove *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status {
for _, pl := range f.preFilterPlugins { for _, pl := range f.preFilterPlugins {
if updater := pl.Updater(); updater != nil { if pl.Extensions() == nil {
status := updater.RemovePod(pc, podToSchedule, podToRemove, nodeInfo) continue
if !status.IsSuccess() { }
if status := pl.Extensions().RemovePod(pc, podToSchedule, podToRemove, nodeInfo); !status.IsSuccess() {
msg := fmt.Sprintf("error while running RemovePod for plugin %q while scheduling pod %q: %v", msg := fmt.Sprintf("error while running RemovePod for plugin %q while scheduling pod %q: %v",
pl.Name(), podToSchedule.Name, status.Message()) pl.Name(), podToSchedule.Name, status.Message())
klog.Error(msg) klog.Error(msg)
return NewStatus(Error, msg) return NewStatus(Error, msg)
} }
} }
}
return nil return nil
} }
@ -417,8 +417,10 @@ func (f *framework) RunScorePlugins(pc *PluginContext, pod *v1.Pod, nodes []*v1.
workqueue.ParallelizeUntil(ctx, 16, len(f.scorePlugins), func(index int) { workqueue.ParallelizeUntil(ctx, 16, len(f.scorePlugins), func(index int) {
pl := f.scorePlugins[index] pl := f.scorePlugins[index]
nodeScoreList := pluginToNodeScores[pl.Name()] nodeScoreList := pluginToNodeScores[pl.Name()]
status := pl.NormalizeScore(pc, pod, nodeScoreList) if pl.Extensions() == nil {
if !status.IsSuccess() { return
}
if status := pl.Extensions().NormalizeScore(pc, pod, nodeScoreList); !status.IsSuccess() {
err := fmt.Errorf("normalize score plugin %q failed with error %v", pl.Name(), status.Message()) err := fmt.Errorf("normalize score plugin %q failed with error %v", pl.Name(), status.Message())
errCh.SendErrorWithCancel(err, cancel) errCh.SendErrorWithCancel(err, cancel)
return return

36
pkg/scheduler/framework/v1alpha1/framework_test.go
View File

@ -34,7 +34,7 @@ const (
scorePlugin1 = "score-plugin-1" scorePlugin1 = "score-plugin-1"
pluginNotImplementingScore = "plugin-not-implementing-score" pluginNotImplementingScore = "plugin-not-implementing-score"
preFilterPluginName = "prefilter-plugin" preFilterPluginName = "prefilter-plugin"
preFilterWithUpdaterPluginName = "prefilter-with-updater-plugin" preFilterWithExtensionsPluginName = "prefilter-with-extensions-plugin"
) )
// TestScoreWithNormalizePlugin implements ScoreWithNormalizePlugin interface. // TestScoreWithNormalizePlugin implements ScoreWithNormalizePlugin interface.
@ -87,6 +87,10 @@ func (pl *TestScoreWithNormalizePlugin) Score(pc *PluginContext, p *v1.Pod, node
return setScoreRes(pl.inj) return setScoreRes(pl.inj)
} }
func (pl *TestScoreWithNormalizePlugin) Extensions() ScoreExtensions {
return pl
}
// TestScorePlugin only implements ScorePlugin interface. // TestScorePlugin only implements ScorePlugin interface.
type TestScorePlugin struct { type TestScorePlugin struct {
name string name string
@ -101,7 +105,7 @@ func (pl *TestScorePlugin) Score(pc *PluginContext, p *v1.Pod, nodeName string)
return setScoreRes(pl.inj) return setScoreRes(pl.inj)
} }
func (pl *TestScorePlugin) NormalizeScore(pc *PluginContext, pod *v1.Pod, scores NodeScoreList) *Status { func (pl *TestScorePlugin) Extensions() ScoreExtensions {
return nil return nil
} }
@ -126,39 +130,39 @@ func (pl *TestPreFilterPlugin) PreFilter(pc *PluginContext, p *v1.Pod) *Status {
return nil return nil
} }
func (pl *TestPreFilterPlugin) Updater() Updater { func (pl *TestPreFilterPlugin) Extensions() PreFilterExtensions {
return nil return nil
} }
// TestPreFilterWithUpdatePlugin implements Add/Remove interfaces. // TestPreFilterWithExtensionsPlugin implements Add/Remove interfaces.
type TestPreFilterWithUpdaterPlugin struct { type TestPreFilterWithExtensionsPlugin struct {
PreFilterCalled int PreFilterCalled int
AddCalled int AddCalled int
RemoveCalled int RemoveCalled int
} }
func (pl *TestPreFilterWithUpdaterPlugin) Name() string { func (pl *TestPreFilterWithExtensionsPlugin) Name() string {
return preFilterWithUpdaterPluginName return preFilterWithExtensionsPluginName
} }
func (pl *TestPreFilterWithUpdaterPlugin) PreFilter(pc *PluginContext, p *v1.Pod) *Status { func (pl *TestPreFilterWithExtensionsPlugin) PreFilter(pc *PluginContext, p *v1.Pod) *Status {
pl.PreFilterCalled++ pl.PreFilterCalled++
return nil return nil
} }
func (pl *TestPreFilterWithUpdaterPlugin) AddPod(pc *PluginContext, podToSchedule *v1.Pod, func (pl *TestPreFilterWithExtensionsPlugin) AddPod(pc *PluginContext, podToSchedule *v1.Pod,
podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status { podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status {
pl.AddCalled++ pl.AddCalled++
return nil return nil
} }
func (pl *TestPreFilterWithUpdaterPlugin) RemovePod(pc *PluginContext, podToSchedule *v1.Pod, func (pl *TestPreFilterWithExtensionsPlugin) RemovePod(pc *PluginContext, podToSchedule *v1.Pod,
podToRemove *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status { podToRemove *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status {
pl.RemoveCalled++ pl.RemoveCalled++
return nil return nil
} }
func (pl *TestPreFilterWithUpdaterPlugin) Updater() Updater { func (pl *TestPreFilterWithExtensionsPlugin) Extensions() PreFilterExtensions {
return pl return pl
} }
@ -424,25 +428,25 @@ func TestRunScorePlugins(t *testing.T) {
func TestPreFilterPlugins(t *testing.T) { func TestPreFilterPlugins(t *testing.T) {
preFilter1 := &TestPreFilterPlugin{} preFilter1 := &TestPreFilterPlugin{}
preFilter2 := &TestPreFilterWithUpdaterPlugin{} preFilter2 := &TestPreFilterWithExtensionsPlugin{}
r := make(Registry) r := make(Registry)
r.Register(preFilterPluginName, r.Register(preFilterPluginName,
func(_ *runtime.Unknown, fh FrameworkHandle) (Plugin, error) { func(_ *runtime.Unknown, fh FrameworkHandle) (Plugin, error) {
return preFilter1, nil return preFilter1, nil
}) })
r.Register(preFilterWithUpdaterPluginName, r.Register(preFilterWithExtensionsPluginName,
func(_ *runtime.Unknown, fh FrameworkHandle) (Plugin, error) { func(_ *runtime.Unknown, fh FrameworkHandle) (Plugin, error) {
return preFilter2, nil return preFilter2, nil
}) })
plugins := &config.Plugins{PreFilter: &config.PluginSet{Enabled: []config.Plugin{{Name: preFilterWithUpdaterPluginName}, {Name: preFilterPluginName}}}} plugins := &config.Plugins{PreFilter: &config.PluginSet{Enabled: []config.Plugin{{Name: preFilterWithExtensionsPluginName}, {Name: preFilterPluginName}}}}
t.Run("TestPreFilterPlugin", func(t *testing.T) { t.Run("TestPreFilterPlugin", func(t *testing.T) {
f, err := NewFramework(r, plugins, emptyArgs) f, err := NewFramework(r, plugins, emptyArgs)
if err != nil { if err != nil {
t.Fatalf("Failed to create framework for testing: %v", err) t.Fatalf("Failed to create framework for testing: %v", err)
} }
f.RunPreFilterPlugins(nil, nil) f.RunPreFilterPlugins(nil, nil)
f.RunPreFilterUpdaterAddPod(nil, nil, nil, nil) f.RunPreFilterExtensionAddPod(nil, nil, nil, nil)
f.RunPreFilterUpdaterRemovePod(nil, nil, nil, nil) f.RunPreFilterExtensionRemovePod(nil, nil, nil, nil)
if preFilter1.PreFilterCalled != 1 { if preFilter1.PreFilterCalled != 1 {
t.Errorf("preFilter1 called %v, expected: 1", preFilter1.PreFilterCalled) t.Errorf("preFilter1 called %v, expected: 1", preFilter1.PreFilterCalled)

43
pkg/scheduler/framework/v1alpha1/interface.go
View File

@ -165,10 +165,10 @@ type QueueSortPlugin interface {
Less(*PodInfo, *PodInfo) bool Less(*PodInfo, *PodInfo) bool
} }
// Updater is an interface that is included in plugins that allow specifying // PreFilterExtensions is an interface that is included in plugins that allow specifying
// callbacks to make incremental updates to its supposedly pre-calculated // callbacks to make incremental updates to its supposedly pre-calculated
// state. // state.
type Updater interface { type PreFilterExtensions interface {
// AddPod is called by the framework while trying to evaluate the impact // AddPod is called by the framework while trying to evaluate the impact
// of adding podToAdd to the node while scheduling podToSchedule. // of adding podToAdd to the node while scheduling podToSchedule.
AddPod(pc *PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status AddPod(pc *PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status
@ -184,13 +184,13 @@ type PreFilterPlugin interface {
// PreFilter is called at the beginning of the scheduling cycle. All PreFilter // PreFilter is called at the beginning of the scheduling cycle. All PreFilter
// plugins must return success or the pod will be rejected. // plugins must return success or the pod will be rejected.
PreFilter(pc *PluginContext, p *v1.Pod) *Status PreFilter(pc *PluginContext, p *v1.Pod) *Status
// Updater returns an updater if the plugin implements one, or nil if it // Extensions returns a PreFilterExtensions interface if the plugin implements one,
// does not. A Pre-filter plugin can provide an updater to incrementally // or nil if it does not. A Pre-filter plugin can provide extensions to incrementally
// modify its pre-processed info. The framework guarantees that the updater // modify its pre-processed info. The framework guarantees that the extensions
// AddPod/RemovePod functions will only be called after PreFilter, // AddPod/RemovePod will only be called after PreFilter, possibly on a cloned
// possibly on a cloned PluginContext, and may call those functions more than // PluginContext, and may call those functions more than once before calling
// once before calling Filter again on a specific node. // Filter again on a specific node.
Updater() Updater Extensions() PreFilterExtensions
} }
// FilterPlugin is an interface for Filter plugins. These plugins are called at the // FilterPlugin is an interface for Filter plugins. These plugins are called at the
@ -230,6 +230,14 @@ type PostFilterPlugin interface {
PostFilter(pc *PluginContext, pod *v1.Pod, nodes []*v1.Node, filteredNodesStatuses NodeToStatusMap) *Status PostFilter(pc *PluginContext, pod *v1.Pod, nodes []*v1.Node, filteredNodesStatuses NodeToStatusMap) *Status
} }
// ScoreExtensions is an interface for Score extended functionality.
type ScoreExtensions interface {
// NormalizeScore is called for all node scores produced by the same plugin's "Score"
// method. A successful run of NormalizeScore will update the scores list and return
// a success status.
NormalizeScore(pc *PluginContext, p *v1.Pod, scores NodeScoreList) *Status
}
// ScorePlugin is an interface that must be implemented by "score" plugins to rank // ScorePlugin is an interface that must be implemented by "score" plugins to rank
// nodes that passed the filtering phase. // nodes that passed the filtering phase.
type ScorePlugin interface { type ScorePlugin interface {
@ -239,13 +247,8 @@ type ScorePlugin interface {
// the pod will be rejected. // the pod will be rejected.
Score(pc *PluginContext, p *v1.Pod, nodeName string) (int, *Status) Score(pc *PluginContext, p *v1.Pod, nodeName string) (int, *Status)
// NormalizeScore is called for all node scores produced by the same plugin's "Score" // Extensions returns a ScoreExtensions interface if it implements one, or nil if does not.
// method. A successful run of NormalizeScore will update the scores list and return Extensions() ScoreExtensions
// a success status.
//
// NOTE: This function is optional, and you could implement it as a no-op by simply
// returning nil.
NormalizeScore(pc *PluginContext, p *v1.Pod, scores NodeScoreList) *Status
} }
// ReservePlugin is an interface for Reserve plugins. These plugins are called // ReservePlugin is an interface for Reserve plugins. These plugins are called
@ -342,15 +345,15 @@ type Framework interface {
// schedule the target pod. // schedule the target pod.
RunFilterPlugins(pc *PluginContext, pod *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status RunFilterPlugins(pc *PluginContext, pod *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status
// RunPreFilterUpdaterAddPod calls the AddPod interface for the set of configured // RunPreFilterExtensionAddPod calls the AddPod interface for the set of configured
// PreFilter plugins. It returns directly if any of the plugins return any // PreFilter plugins. It returns directly if any of the plugins return any
// status other than Success. // status other than Success.
RunPreFilterUpdaterAddPod(pc *PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status RunPreFilterExtensionAddPod(pc *PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status
// RunPreFilterUpdaterRemovePod calls the RemovePod interface for the set of configured // RunPreFilterExtensionRemovePod calls the RemovePod interface for the set of configured
// PreFilter plugins. It returns directly if any of the plugins return any // PreFilter plugins. It returns directly if any of the plugins return any
// status other than Success. // status other than Success.
RunPreFilterUpdaterRemovePod(pc *PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status RunPreFilterExtensionRemovePod(pc *PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *Status
// RunPostFilterPlugins runs the set of configured post-filter plugins. If any // RunPostFilterPlugins runs the set of configured post-filter plugins. If any
// of these plugins returns any status other than "Success", the given node is // of these plugins returns any status other than "Success", the given node is

4
pkg/scheduler/internal/queue/scheduling_queue_test.go
View File

@ -175,11 +175,11 @@ func (*fakeFramework) RunFilterPlugins(pc *framework.PluginContext, pod *v1.Pod,
return nil return nil
} }
func (*fakeFramework) RunPreFilterUpdaterAddPod(pc *framework.PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *framework.Status { func (*fakeFramework) RunPreFilterExtensionAddPod(pc *framework.PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *framework.Status {
return nil return nil
} }
func (*fakeFramework) RunPreFilterUpdaterRemovePod(pc *framework.PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *framework.Status { func (*fakeFramework) RunPreFilterExtensionRemovePod(pc *framework.PluginContext, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *schedulernodeinfo.NodeInfo) *framework.Status {
return nil return nil
} }

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

@ -162,7 +162,7 @@ func (sp *ScorePlugin) Score(pc *framework.PluginContext, p *v1.Pod, nodeName st
return score, nil return score, nil
} }
func (sp *ScorePlugin) NormalizeScore(pc *framework.PluginContext, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status { func (sp *ScorePlugin) Extensions() framework.ScoreExtensions {
return nil return nil
} }
@ -189,6 +189,10 @@ func (sp *ScoreWithNormalizePlugin) NormalizeScore(pc *framework.PluginContext,
return nil return nil
} }
func (sp *ScoreWithNormalizePlugin) Extensions() framework.ScoreExtensions {
return sp
}
// Name returns name of the plugin. // Name returns name of the plugin.
func (fp *FilterPlugin) Name() string { func (fp *FilterPlugin) Name() string {
return filterPluginName return filterPluginName
@ -330,8 +334,8 @@ func (pp *PreFilterPlugin) Name() string {
return prefilterPluginName return prefilterPluginName
} }
// Updater returns the updater interface. // Extensions returns the PreFilterExtensions interface.
func (pp *PreFilterPlugin) Updater() framework.Updater { func (pp *PreFilterPlugin) Extensions() framework.PreFilterExtensions {
return nil return nil
} }

2
test/integration/scheduler/preemption_test.go
View File

@ -111,7 +111,7 @@ func (fp *tokenFilter) RemovePod(pc *framework.PluginContext, podToSchedule *v1.
return nil return nil
} }
func (fp *tokenFilter) Updater() framework.Updater { func (fp *tokenFilter) Extensions() framework.PreFilterExtensions {
return fp return fp
} }