Merge pull request #83080 from ahg-g/ahg-config-layer

Mapping Layer - 1. A mapping from predicates/priorities to plugin configuration

pkg/scheduler/apis/config/types.go

@@ -209,3 +209,37 @@ type PluginConfig struct {
 	// Args defines the arguments passed to the plugins at the time of initialization. Args can have arbitrary structure.
 	Args runtime.Unknown
 }
+
+///////////////////////////////////////////////////////////////////////////////
+// NOTE: The following methods are intentionally left out of the staging mirror.
+///////////////////////////////////////////////////////////////////////////////
+
+func appendPluginSet(dst *PluginSet, src *PluginSet) *PluginSet {
+	if dst == nil {
+		dst = &PluginSet{}
+	}
+	if src != nil {
+		dst.Enabled = append(dst.Enabled, src.Enabled...)
+		dst.Disabled = append(dst.Disabled, src.Disabled...)
+	}
+	return dst
+}
+
+// Append appends src Plugins to current Plugins. If a PluginSet is nil, it will
+// be created.
+func (p *Plugins) Append(src *Plugins) {
+	if p == nil || src == nil {
+		return
+	}
+	p.QueueSort = appendPluginSet(p.QueueSort, src.QueueSort)
+	p.PreFilter = appendPluginSet(p.PreFilter, src.PreFilter)
+	p.Filter = appendPluginSet(p.Filter, src.Filter)
+	p.PostFilter = appendPluginSet(p.PostFilter, src.PostFilter)
+	p.Score = appendPluginSet(p.Score, src.Score)
+	p.Reserve = appendPluginSet(p.Reserve, src.Reserve)
+	p.Permit = appendPluginSet(p.Permit, src.Permit)
+	p.PreBind = appendPluginSet(p.PreBind, src.PreBind)
+	p.Bind = appendPluginSet(p.Bind, src.Bind)
+	p.PostBind = appendPluginSet(p.PostBind, src.PostBind)
+	p.Unreserve = appendPluginSet(p.Unreserve, src.Unreserve)
+}

pkg/scheduler/framework/plugins/BUILD

@@ -1,4 +1,4 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
@@ -6,6 +6,7 @@ go_library(
     importpath = "k8s.io/kubernetes/pkg/scheduler/framework/plugins",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/scheduler/apis/config:go_default_library",
         "//pkg/scheduler/framework/plugins/noop:go_default_library",
         "//pkg/scheduler/framework/v1alpha1:go_default_library",
     ],
@@ -28,3 +29,13 @@ filegroup(
     tags = ["automanaged"],
     visibility = ["//visibility:public"],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["default_registry_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//pkg/scheduler/apis/config:go_default_library",
+        "//vendor/github.com/google/go-cmp/cmp:go_default_library",
+    ],
+)

pkg/scheduler/framework/plugins/default_registry.go

@@ -17,6 +17,9 @@ limitations under the License.
 package plugins
 
 import (
+	"fmt"
+
+	"k8s.io/kubernetes/pkg/scheduler/apis/config"
 	noop "k8s.io/kubernetes/pkg/scheduler/framework/plugins/noop"
 	framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 )
@@ -31,3 +34,47 @@ func NewDefaultRegistry() framework.Registry {
 		noop.Name: noop.New,
 	}
 }
+
+// ConfigProducerArgs contains arguments that are passed to the producer.
+// As we add more predicates/priorities to framework plugins mappings, more arguments
+// may be added here.
+type ConfigProducerArgs struct {
+	// Weight used for priority functions.
+	Weight int32
+}
+
+// ConfigProducer produces a framework's configuration.
+type ConfigProducer func(args ConfigProducerArgs) (config.Plugins, []config.PluginConfig)
+
+// ConfigProducerRegistry tracks mappings from predicates/priorities to framework config producers.
+type ConfigProducerRegistry struct {
+	// maps that associate predicates/priorities with framework plugin configurations.
+	PredicateToConfigProducer map[string]ConfigProducer
+	PriorityToConfigProducer  map[string]ConfigProducer
+}
+
+// NewConfigProducerRegistry creates a new producer registry.
+func NewConfigProducerRegistry() *ConfigProducerRegistry {
+	return &ConfigProducerRegistry{
+		PredicateToConfigProducer: make(map[string]ConfigProducer),
+		PriorityToConfigProducer:  make(map[string]ConfigProducer),
+	}
+}
+
+func registerProducer(name string, producer ConfigProducer, producersMap map[string]ConfigProducer) error {
+	if _, exist := producersMap[name]; exist {
+		return fmt.Errorf("already registered %q", name)
+	}
+	producersMap[name] = producer
+	return nil
+}
+
+// RegisterPredicate registers a config producer for a predicate.
+func (f *ConfigProducerRegistry) RegisterPredicate(name string, producer ConfigProducer) error {
+	return registerProducer(name, producer, f.PredicateToConfigProducer)
+}
+
+// RegisterPriority registers a framework config producer for a priority.
+func (f *ConfigProducerRegistry) RegisterPriority(name string, producer ConfigProducer) error {
+	return registerProducer(name, producer, f.PriorityToConfigProducer)
+}

pkg/scheduler/framework/plugins/default_registry_test.go

@@ -0,0 +1,135 @@
+/*
+Copyright 2019 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package plugins
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/google/go-cmp/cmp"
+
+	"k8s.io/kubernetes/pkg/scheduler/apis/config"
+)
+
+func appendToPluginSet(pluginSet *config.PluginSet, name string, weight *int32) *config.PluginSet {
+	if pluginSet == nil {
+		pluginSet = &config.PluginSet{}
+	}
+	config := config.Plugin{Name: name}
+	if weight != nil {
+		config.Weight = *weight
+	}
+	pluginSet.Enabled = append(pluginSet.Enabled, config)
+	return pluginSet
+}
+
+func produceConfig(keys []string, producersMap map[string]ConfigProducer, args ConfigProducerArgs) (*config.Plugins, []config.PluginConfig, error) {
+	var plugins config.Plugins
+	var pluginConfig []config.PluginConfig
+	for _, k := range keys {
+		producer, exist := producersMap[k]
+		if !exist {
+			return nil, nil, fmt.Errorf("finding key %q", k)
+		}
+		p, pc := producer(args)
+		plugins.Append(&p)
+		pluginConfig = append(pluginConfig, pc...)
+	}
+	return &plugins, pluginConfig, nil
+}
+
+func TestRegisterConfigProducers(t *testing.T) {
+	registry := NewConfigProducerRegistry()
+	testPredicateName1 := "testPredicate1"
+	testFilterName1 := "testFilter1"
+	registry.RegisterPredicate(testPredicateName1,
+		func(_ ConfigProducerArgs) (plugins config.Plugins, pluginConfig []config.PluginConfig) {
+			plugins.Filter = appendToPluginSet(plugins.Filter, testFilterName1, nil)
+			return
+		})
+
+	testPredicateName2 := "testPredicate2"
+	testFilterName2 := "testFilter2"
+	registry.RegisterPredicate(testPredicateName2,
+		func(_ ConfigProducerArgs) (plugins config.Plugins, pluginConfig []config.PluginConfig) {
+			plugins.Filter = appendToPluginSet(plugins.Filter, testFilterName2, nil)
+			return
+		})
+
+	testPriorityName1 := "testPriority1"
+	testScoreName1 := "testScore1"
+	registry.RegisterPriority(testPriorityName1,
+		func(args ConfigProducerArgs) (plugins config.Plugins, pluginConfig []config.PluginConfig) {
+			plugins.Score = appendToPluginSet(plugins.Score, testScoreName1, &args.Weight)
+			return
+		})
+
+	testPriorityName2 := "testPriority2"
+	testScoreName2 := "testScore2"
+	registry.RegisterPriority(testPriorityName2,
+		func(args ConfigProducerArgs) (plugins config.Plugins, pluginConfig []config.PluginConfig) {
+			plugins.Score = appendToPluginSet(plugins.Score, testScoreName2, &args.Weight)
+			return
+		})
+
+	args := ConfigProducerArgs{Weight: 1}
+	predicatePlugins, _, err := produceConfig(
+		[]string{testPredicateName1, testPredicateName2}, registry.PredicateToConfigProducer, args)
+	if err != nil {
+		t.Fatalf("producing predicate framework configs: %v.", err)
+	}
+
+	priorityPlugins, _, err := produceConfig(
+		[]string{testPriorityName1, testPriorityName2}, registry.PriorityToConfigProducer, args)
+	if err != nil {
+		t.Fatalf("producing predicate framework configs: %v.", err)
+	}
+
+	// Verify that predicates and priorities are in the map and produce the expected score configurations.
+	var gotPlugins config.Plugins
+	gotPlugins.Append(predicatePlugins)
+	gotPlugins.Append(priorityPlugins)
+
+	// Verify the aggregated configuration.
+	wantPlugins := config.Plugins{
+		QueueSort: &config.PluginSet{},
+		PreFilter: &config.PluginSet{},
+		Filter: &config.PluginSet{
+			Enabled: []config.Plugin{
+				{Name: testFilterName1},
+				{Name: testFilterName2},
+			},
+		},
+		PostFilter: &config.PluginSet{},
+		Score: &config.PluginSet{
+			Enabled: []config.Plugin{
+				{Name: testScoreName1, Weight: 1},
+				{Name: testScoreName2, Weight: 1},
+			},
+		},
+		Reserve:   &config.PluginSet{},
+		Permit:    &config.PluginSet{},
+		PreBind:   &config.PluginSet{},
+		Bind:      &config.PluginSet{},
+		PostBind:  &config.PluginSet{},
+		Unreserve: &config.PluginSet{},
+	}
+
+	if diff := cmp.Diff(wantPlugins, gotPlugins); diff != "" {
+		t.Errorf("unexpected plugin configuration (-want, +got): %s", diff)
+	}
+}