Initial next gen scheduler

Signed-off-by: aavarghese <avarghese@us.ibm.com>

config/source/multi/600-config-scheduler.yaml

@@ -0,0 +1,36 @@
+# Copyright 2021 The Knative 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
+#
+# https://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.
+
+apiVersion: v1
+kind: ConfigMap
+metadata:
+ name: config-scheduler
+ namespace: knative-eventing
+ labels:
+ kafka.eventing.knative.dev/release: devel
+data:
+ predicates: |+
+ [
+ {"name" : "PodFitsResources"},
+ {"name" : "NoMaxResourceCount"},
+ {"name" : "EvenPodSpread"}
+ ]
+ priorities: |+
+ [
+ {"name" : "AvailabilityZonePriority", "weight" : 1},
+ {"name" : "AvailabilityNodePriority", "weight" : 1},
+ {"name" : "LowestOrdinalPriority", "weight" : 1}
+ ]
+
+ 

config/source/multi/deployments/controller.yaml

@@ -55,7 +55,10 @@ spec:
 
  # The scheduling policy type for placing vreplicas on pods (see type SchedulerPolicyType for enum list)
  - name: SCHEDULER_POLICY_TYPE
- value: 'EVENSPREAD_BYNODE'
+ value: ''
+
+ - name: CONFIG_SCHEDULER
+ value: config-scheduler
 
  resources:
  requests:

pkg/common/scheduler/factory/registry.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2021 The Knative 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 factory
+
+import (
+ "fmt"
+
+ state "knative.dev/eventing-kafka/pkg/common/scheduler/state"
+)
+
+// RegistryFP is a collection of all available filter plugins.
+type RegistryFP map[string]state.FilterPlugin
+
+// RegistrySP is a collection of all available scoring plugins.
+type RegistrySP map[string]state.ScorePlugin
+
+var (
+ FilterRegistry = make(RegistryFP)
+ ScoreRegistry = make(RegistrySP)
+)
+
+// Register adds a new plugin to the registry. If a plugin with the same name
+// exists, it returns an error.
+func RegisterFP(name string, factory state.FilterPlugin) error {
+ if _, ok := FilterRegistry[name]; ok {
+ return fmt.Errorf("a plugin named %v already exists", name)
+ }
+ FilterRegistry[name] = factory
+ return nil
+}
+
+// Unregister removes an existing plugin from the registry. If no plugin with
+// the provided name exists, it returns an error.
+func UnregisterFP(name string) error {
+ if _, ok := FilterRegistry[name]; !ok {
+ return fmt.Errorf("no plugin named %v exists", name)
+ }
+ delete(FilterRegistry, name)
+ return nil
+}
+
+func GetFilterPlugin(name string) (state.FilterPlugin, error) {
+ if f, exist := FilterRegistry[name]; exist {
+ return f, nil
+ }
+ return nil, fmt.Errorf("no plugin named %v exists", name)
+}
+
+// Register adds a new plugin to the registry. If a plugin with the same name
+// exists, it returns an error.
+func RegisterSP(name string, factory state.ScorePlugin) error {
+ if _, ok := ScoreRegistry[name]; ok {
+ return fmt.Errorf("a plugin named %v already exists", name)
+ }
+ ScoreRegistry[name] = factory
+ return nil
+}
+
+// Unregister removes an existing plugin from the registry. If no plugin with
+// the provided name exists, it returns an error.
+func UnregisterSP(name string) error {
+ if _, ok := ScoreRegistry[name]; !ok {
+ return fmt.Errorf("no plugin named %v exists", name)
+ }
+ delete(ScoreRegistry, name)
+ return nil
+}
+
+func GetScorePlugin(name string) (state.ScorePlugin, error) {
+ if f, exist := ScoreRegistry[name]; exist {
+ return f, nil
+ }
+ return nil, fmt.Errorf("no plugin named %v exists", name)
+}

pkg/common/scheduler/plugins/core/availabilitynodepriority/availability_node_priority.go

@@ -0,0 +1,71 @@
+/*
+Copyright 2021 The Knative 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 availabilitynodepriority
+
+import (
+ "context"
+ "fmt"
+
+ "knative.dev/eventing-kafka/pkg/common/scheduler/factory"
+ state "knative.dev/eventing-kafka/pkg/common/scheduler/state"
+ "knative.dev/pkg/logging"
+)
+
+// AvailabilityNodePriority is a score plugin that favors pods that create an even spread of resources across nodes for HA
+type AvailabilityNodePriority struct {
+}
+
+// Verify AvailabilityNodePriority Implements ScorePlugin Interface
+var _ state.ScorePlugin = &AvailabilityNodePriority{}
+
+// Name of the plugin
+const Name = state.AvailabilityNodePriority
+
+func init() {
+ factory.RegisterSP(Name, &AvailabilityNodePriority{}) //TODO: Not working
+ fmt.Println("AvailabilityNodePriority has been registered")
+}
+
+// Name returns name of the plugin
+func (pl *AvailabilityNodePriority) Name() string {
+ return Name
+}
+
+// Score invoked at the score extension point.
+func (pl *AvailabilityNodePriority) Score(ctx context.Context, states *state.State, podID int32) (int64, *state.Status) {
+ logger := logging.FromContext(ctx).With("Score", pl.Name())
+
+ score := calculatePriority()
+
+ logger.Infof("Pod %q scored by %q priority successfully", podID, pl.Name())
+ return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *AvailabilityNodePriority) ScoreExtensions() state.ScoreExtensions {
+ return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *AvailabilityNodePriority) NormalizeScore(ctx context.Context, state *state.State, scores state.PodScoreList) *state.Status {
+ return nil
+}
+
+// calculatePriority returns the priority of a pod. Given the ...
+func calculatePriority() int64 {
+ return 1
+}

pkg/common/scheduler/plugins/core/availabilitynodepriority/availability_node_priority_test.go

@@ -0,0 +1,17 @@
+/*
+Copyright 2021 The Knative 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 availabilitynodepriority

pkg/common/scheduler/plugins/core/availabilityzonepriority/availability_zone_priority.go

@@ -0,0 +1,70 @@
+/*
+Copyright 2021 The Knative 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 availabilityzonepriority
+
+import (
+ "context"
+
+ "knative.dev/eventing-kafka/pkg/common/scheduler/factory"
+ state "knative.dev/eventing-kafka/pkg/common/scheduler/state"
+ "knative.dev/pkg/logging"
+)
+
+// AvailabilityZonePriority is a score plugin that favors pods that create an even spread of resources across zones for HA
+type AvailabilityZonePriority struct {
+}
+
+// Verify AvailabilityZonePriority Implements ScorePlugin Interface
+var _ state.ScorePlugin = &AvailabilityZonePriority{}
+
+// Name of the plugin
+const Name = state.AvailabilityZonePriority
+
+func init() {
+ factory.RegisterSP(Name, &AvailabilityZonePriority{})
+ //fmt.Println("AvailabilityZonePriority plugin has been registered")
+}
+
+// Name returns name of the plugin
+func (pl *AvailabilityZonePriority) Name() string {
+ return Name
+}
+
+// Ssts invoked at the ssts extension point.
+func (pl *AvailabilityZonePriority) Score(ctx context.Context, states *state.State, podID int32) (int64, *state.Status) {
+ logger := logging.FromContext(ctx).With("Score", pl.Name())
+
+ score := calculatePriority()
+
+ logger.Infof("Pod %q scored by %q priority successfully", podID, pl.Name())
+ return score, state.NewStatus(state.Success)
+}
+
+// ScoreExtensions of the Score plugin.
+func (pl *AvailabilityZonePriority) ScoreExtensions() state.ScoreExtensions {
+ return pl
+}
+
+// NormalizeScore invoked after scoring all pods.
+func (pl *AvailabilityZonePriority) NormalizeScore(ctx context.Context, state *state.State, scores state.PodScoreList) *state.Status {
+ return nil
+}
+
+// calculatePriority returns the priority of a pod. Given the ...
+func calculatePriority() int64 {
+ return 1
+}

pkg/common/scheduler/plugins/core/availabilityzonepriority/availability_zone_priority_test.go

@@ -0,0 +1,17 @@
+/*
+Copyright 2021 The Knative 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 availabilityzonepriority

pkg/common/scheduler/plugins/core/evenpodspread/even_pod_spread.go

@@ -0,0 +1,53 @@
+/*
+Copyright 2021 The Knative 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 evenpodspread
+
+import (
+ "context"
+
+ "knative.dev/eventing-kafka/pkg/common/scheduler/factory"
+ state "knative.dev/eventing-kafka/pkg/common/scheduler/state"
+ "knative.dev/pkg/logging"
+)
+
+// EvenPodSpread is a filter plugin that eliminates pods that do not create an equal spread of resources across pods
+type EvenPodSpread struct {
+}
+
+// Verify EvenPodSpread Implements FilterPlugin Interface
+var _ state.FilterPlugin = &EvenPodSpread{}
+
+// Name of the plugin
+const Name = state.EvenPodSpread
+
+func init() {
+ factory.RegisterFP(Name, &EvenPodSpread{})
+ //fmt.Println("EvenPodSpread plugin has been registered")
+}
+
+// Name returns name of the plugin
+func (pl *EvenPodSpread) Name() string {
+ return Name
+}
+
+// Filter invoked at the filter extension point.
+func (pl *EvenPodSpread) Filter(ctx context.Context, states *state.State, podID int32) *state.Status {
+ logger := logging.FromContext(ctx).With("Filter", pl.Name())
+
+ logger.Infof("Pod %q passed %q predicate successfully", podID, pl.Name())
+ return state.NewStatus(state.Success)
+}

pkg/common/scheduler/plugins/core/evenpodspread/even_pod_spread_test.go

@@ -0,0 +1,17 @@
+/*
+Copyright 2021 The Knative 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 evenpodspread