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,116 @@
+/*
+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"
+
+ "k8s.io/apimachinery/pkg/types"
+ "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
+
+const (
+ // When zone information is present, give 1/3 of the weighting to zone spreading, 2/3 to node spreading
+ zoneWeighting float64 = 1.0 / 3.0
+ MaxSkew = 2 //TODO: Move to predicate argument
+ ErrReasonNegativeScore = "computed score is negative"
+ ErrReasonNoResource = "node or zone does not exist"
+)
+
+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. The "score" returned in this function is higher for pods with lower ordinals and higher free capacity
+// It is normalized later with node info
+func (pl *AvailabilityNodePriority) Score(ctx context.Context, states *state.State, key types.NamespacedName, podID int32) (int64, *state.Status) {
+ logger := logging.FromContext(ctx).With("Score", pl.Name())
+ var score int64 = 0
+ var skew int32
+
+ if states.LastOrdinal > 0 {
+ _, nodeName, err := states.GetPodInfo(state.PodNameFromOrdinal(states.StatefulSetName, podID))
+ if err != nil {
+ return score, state.NewStatus(state.Error, ErrReasonNoResource)
+ }
+
+ currentReps := states.NodeSpread[key][nodeName] //get #vreps on this node
+ var totalvreps int32 = currentReps
+ for otherNodeName, _ := range states.NodeToZoneMap { //compare with #vreps on other nodes
+ if otherNodeName != nodeName {
+ otherReps := states.NodeSpread[key][otherNodeName]
+ totalvreps = totalvreps + otherReps
+ if skew = (currentReps + 1) - otherReps; skew < 0 {
+ skew = skew * int32(-1)
+ }
+
+ logger.Infof("Current Node %v with %d and Other Node %v with %d causing skew %d", nodeName, currentReps, otherNodeName, otherReps, skew)
+ if skew > MaxSkew {
+ logger.Infof("Pod %d will cause an uneven node spread", podID)
+ score = score + int64(totalvreps-skew) //lesser skews get higher score
+ } else {
+ score = score + int64(totalvreps) //temporary max score
+ }
+ }
+ }
+
+ if score < 0 {
+ return 0, state.NewStatus(state.Error, ErrReasonNegativeScore)
+ }
+ }
+
+ 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, states *state.State, scores state.PodScoreList) *state.Status {
+ logger := logging.FromContext(ctx).With("Normalize Score", pl.Name())
+
+ if len(scores) == 0 {
+ logger.Error("empty score list")
+ return state.NewStatus(state.Error, ErrReasonNoResource)
+ }
+
+ for range scores {
+ }
+ return nil
+}