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

Merge pull request #119444 from AxeZhan/scheduler_score

Browse Source 
revert "refactor: simplify RunScorePlugins for readability + performance"
This commit is contained in:
Kubernetes Prow Robot 2023-07-20 08:12:09 -07:00 committed by GitHub
commit f61a0e4107
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 35 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1017,56 +1017,62 @@ func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state *framework.Cy
allNodePluginScores := make([]framework.NodePluginScores, len(nodes)) allNodePluginScores := make([]framework.NodePluginScores, len(nodes))
numPlugins := len(f.scorePlugins) - state.SkipScorePlugins.Len() numPlugins := len(f.scorePlugins) - state.SkipScorePlugins.Len()
plugins := make([]framework.ScorePlugin, 0, numPlugins) plugins := make([]framework.ScorePlugin, 0, numPlugins)
pluginToNodeScores := make([]framework.NodeScoreList, numPlugins) pluginToNodeScores := make(map[string]framework.NodeScoreList, numPlugins)
for _, pl := range f.scorePlugins { for _, pl := range f.scorePlugins {
if state.SkipScorePlugins.Has(pl.Name()) { if state.SkipScorePlugins.Has(pl.Name()) {
continue continue
} }
plugins = append(plugins, pl) plugins = append(plugins, pl)
pluginToNodeScores[pl.Name()] = make(framework.NodeScoreList, len(nodes))
} }
ctx, cancel := context.WithCancel(ctx) ctx, cancel := context.WithCancel(ctx)
defer cancel() defer cancel()
errCh := parallelize.NewErrorChannel() errCh := parallelize.NewErrorChannel()
logger := klog.FromContext(ctx) if len(plugins) > 0 {
logger = klog.LoggerWithName(logger, "Score") logger := klog.FromContext(ctx)
// TODO(knelasevero): Remove duplicated keys from log entry calls logger = klog.LoggerWithName(logger, "Score")
// When contextualized logging hits GA // TODO(knelasevero): Remove duplicated keys from log entry calls
// https://github.com/kubernetes/kubernetes/issues/111672 // When contextualized logging hits GA
logger = klog.LoggerWithValues(logger, "pod", klog.KObj(pod)) // https://github.com/kubernetes/kubernetes/issues/111672
// Run Score method for each node in parallel. logger = klog.LoggerWithValues(logger, "pod", klog.KObj(pod))
f.Parallelizer().Until(ctx, len(plugins), func(i int) { // Run Score method for each node in parallel.
pl := plugins[i] f.Parallelizer().Until(ctx, len(nodes), func(index int) {
logger := klog.LoggerWithName(logger, pl.Name()) nodeName := nodes[index].Name
nodeScores := make(framework.NodeScoreList, len(nodes))
for index, node := range nodes {
nodeName := node.Name
logger := klog.LoggerWithValues(logger, "node", klog.ObjectRef{Name: nodeName}) logger := klog.LoggerWithValues(logger, "node", klog.ObjectRef{Name: nodeName})
ctx := klog.NewContext(ctx, logger) for _, pl := range plugins {
s, status := f.runScorePlugin(ctx, pl, state, pod, nodeName) logger := klog.LoggerWithName(logger, pl.Name())
if !status.IsSuccess() { ctx := klog.NewContext(ctx, logger)
err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError()) s, status := f.runScorePlugin(ctx, pl, state, pod, nodeName)
errCh.SendErrorWithCancel(err, cancel) if !status.IsSuccess() {
return err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
} errCh.SendErrorWithCancel(err, cancel)
nodeScores[index] = framework.NodeScore{ return
Name: nodeName, }
Score: s, pluginToNodeScores[pl.Name()][index] = framework.NodeScore{
Name: nodeName,
Score: s,
}
} }
}, metrics.Score)
if err := errCh.ReceiveError(); err != nil {
return nil, framework.AsStatus(fmt.Errorf("running Score plugins: %w", err))
} }
}
// Run NormalizeScore method for each ScorePlugin in parallel.
f.Parallelizer().Until(ctx, len(plugins), func(index int) {
pl := plugins[index]
if pl.ScoreExtensions() == nil { if pl.ScoreExtensions() == nil {
pluginToNodeScores[i] = nodeScores
return return
} }
nodeScoreList := pluginToNodeScores[pl.Name()]
status := f.runScoreExtension(ctx, pl, state, pod, nodeScores) status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList)
if !status.IsSuccess() { if !status.IsSuccess() {
err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError()) err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
errCh.SendErrorWithCancel(err, cancel) errCh.SendErrorWithCancel(err, cancel)
return return
} }
pluginToNodeScores[i] = nodeScores
}, metrics.Score) }, metrics.Score)
if err := errCh.ReceiveError(); err != nil { if err := errCh.ReceiveError(); err != nil {
return nil, framework.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err)) return nil, framework.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err))
@ -1082,7 +1088,7 @@ func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state *framework.Cy
for i, pl := range plugins { for i, pl := range plugins {
weight := f.scorePluginWeight[pl.Name()] weight := f.scorePluginWeight[pl.Name()]
nodeScoreList := pluginToNodeScores[i] nodeScoreList := pluginToNodeScores[pl.Name()]
score := nodeScoreList[index].Score score := nodeScoreList[index].Score
if score > framework.MaxNodeScore || score < framework.MinNodeScore { if score > framework.MaxNodeScore || score < framework.MinNodeScore {