Repository: knative-sandbox/eventing-kafka
Pull request #733 (2021-06-23 18:51:38)

[main] Update community files
Cron -knative-prow-robot

/cc knative-sandbox/channel-wg-leads
/assign knative-sandbox/channel-wg-leads

Produced by: knative-sandbox/knobots/actions/update-community

Details:
```

```

knative-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka/pull/733#" title="Author self-approved">knative-automation</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **aliok** after the PR has been reviewed.
You can assign the PR to them by writing `/assign @aliok` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka/blob/main/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["aliok"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/733?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#733](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/733?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (a921a95) into [main](https://codecov.io/gh/knative-sandbox/eventing-kafka/commit/1ee93d6e8cdfb00888d83e656f977ee97bfd70c2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (1ee93d6) will **not change** coverage.
> The diff coverage is `n/a`.

[![Impacted file tree graph](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/733/graphs/tree.svg?width=650&height=150&src=pr&token=W765CT93QC&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/733?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

```diff
@@           Coverage Diff           @@
##             main     #733   +/-   ##
=======================================
  Coverage   73.78%   73.78%           
=======================================
  Files         133      133           
  Lines        5852     5852           
=======================================
  Hits         4318     4318           
  Misses       1312     1312           
  Partials      222      222           
```



------

[Continue to review full report at Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/733?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/733?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Last update [1ee93d6...a921a95](https://codecov.io/gh/knative-sandbox/eventing-kafka/pull/733?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox).


Patch:
diff --git a/OWNERS_ALIASES b/OWNERS_ALIASES
index 0d120a04c..24c126c6f 100644
--- a/OWNERS_ALIASES
+++ b/OWNERS_ALIASES
@@ -1,4 +1,5 @@
-# This file is auto-generated from peribolos
+# This file is auto-generated from peribolos.
+# Do not modify this file, instead modify peribolos/knative-sandbox.yaml
 
 aliases:
   api-core-wg-leads:
diff --git a/pkg/common/scheduler/statefulset/autoscaler.go b/pkg/common/scheduler/statefulset/autoscaler.go
index 6de54d327..3a5f0885c 100644
--- a/pkg/common/scheduler/statefulset/autoscaler.go
+++ b/pkg/common/scheduler/statefulset/autoscaler.go
@@ -36,7 +36,7 @@ type Autoscaler interface {
 	Start(ctx context.Context)
 
 	// Autoscale is used to immediately trigger the autoscaler with the hint
-	// that pending number of vreplicas coudn't be scheduled.
+	// that pending number of vreplicas couldn't be scheduled.
 	Autoscale(pending int32)
 }
 
@@ -106,7 +106,7 @@ func (a *autoscaler) Autoscale(pending int32) {
 }
 
 func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool, pending int32) error {
-	state, err := a.stateAccessor.State()
+	state, err := a.stateAccessor.State(nil)
 	if err != nil {
 		a.logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
 		return err
@@ -128,19 +128,8 @@ func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool, pen
 
 	// Take into account pending replicas
 	if pending > 0 {
-		// The number of replicas may be lower than the last ordinal, for instance
-		// when the statefulset is manually scaled down. In that case, replicas above
-		// scale.Spec.Replicas have not been considered when scheduling vreplicas.
-		// Adjust accordingly (applicable only for MAXFILLUP scheduling policy and not for HA)
-		if state.schedulerPolicy != EVENSPREAD {
-			pending -= state.freeCapacity()
-		}
-
-		// Still need more?
-		if pending > 0 {
-			// Make sure to allocate enough pods for holding all pending replicas.
-			newreplicas += int32(math.Ceil(float64(pending) / float64(a.capacity)))
-		}
+		// Make sure to allocate enough pods for holding all pending replicas.
+		newreplicas += int32(math.Ceil(float64(pending) / float64(a.capacity)))
 	}
 
 	// Make sure to never scale down past the last ordinal
diff --git a/pkg/common/scheduler/statefulset/scheduler.go b/pkg/common/scheduler/statefulset/scheduler.go
index 01c316b04..e740a72f3 100644
--- a/pkg/common/scheduler/statefulset/scheduler.go
+++ b/pkg/common/scheduler/statefulset/scheduler.go
@@ -89,9 +89,14 @@ type StatefulSetScheduler struct {
 	// replicas is the (cached) number of statefulset replicas.
 	replicas int32
 
-	// pending tracks the number of virtual replicas that haven't been scheduled yet.
-	// Used by the autoscaler
+	// pending tracks the number of virtual replicas that haven't been scheduled yet
+	// because there wasn't enough free capacity.
+	// The autoscaler uses
 	pending map[types.NamespacedName]int32
+
+	// reserved tracks vreplicas that have been placed (ie. scheduled) but haven't been
+	// committed yet (ie. not appearing in vpodLister)
+	reserved map[types.NamespacedName]map[string]int32
 }
 
 func NewStatefulSetScheduler(ctx context.Context,
@@ -109,6 +114,7 @@ func NewStatefulSetScheduler(ctx context.Context,
 		pending:           make(map[types.NamespacedName]int32),
 		lock:              new(sync.Mutex),
 		stateAccessor:     stateAccessor,
+		reserved:          make(map[types.NamespacedName]map[string]int32),
 		autoscaler:        autoscaler,
 	}
 
@@ -123,26 +129,31 @@ func NewStatefulSetScheduler(ctx context.Context,
 }
 
 func (s *StatefulSetScheduler) Schedule(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
 	placements, err := s.scheduleVPod(vpod)
 	if placements == nil {
 		return placements, err
 	}
+
 	sort.SliceStable(placements, func(i int, j int) bool {
 		return ordinalFromPodName(placements[i].PodName) < ordinalFromPodName(placements[j].PodName)
 	})
+
+	// Reserve new placements until they are committed to the vpod.
+	s.reservePlacements(vpod, placements)
+
 	return placements, err
 }
 
 func (s *StatefulSetScheduler) scheduleVPod(vpod scheduler.VPod) ([]duckv1alpha1.Placement, error) {
-	s.lock.Lock()
-	defer s.lock.Unlock()
-
 	logger := s.logger.With("key", vpod.GetKey())
 	logger.Info("scheduling")
 
 	// Get the current placements state
 	// Quite an expensive operation but safe and simple.
-	state, err := s.stateAccessor.State()
+	state, err := s.stateAccessor.State(s.reserved)
 	if err != nil {
 		logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))
 		return nil, err
@@ -197,6 +208,7 @@ func (s *StatefulSetScheduler) scheduleVPod(vpod scheduler.VPod) ([]duckv1alpha1
 	} else {
 		placements, left = s.addReplicas(state, vpod.GetVReplicas()-tr, placements)
 	}
+
 	if left > 0 {
 		// Give time for the autoscaler to do its job
 		logger.Info("scheduling failed (not enough pod replicas)", zap.Any("placement", placements), zap.Int32("left", left))
@@ -290,7 +302,6 @@ func (s *StatefulSetScheduler) removeReplicasEvenSpread(diff int32, placements [
 
 func (s *StatefulSetScheduler) addReplicas(state *state, diff int32, placements []duckv1alpha1.Placement) ([]duckv1alpha1.Placement, int32) {
 	// Pod affinity algorithm: prefer adding replicas to existing pods before considering other replicas
-	// In the future, we might want to spread replicas across pods in different regions.
 	newPlacements := make([]duckv1alpha1.Placement, 0, len(placements))
 
 	// Add to existing
@@ -487,3 +498,29 @@ func (s *StatefulSetScheduler) updateStatefulset(obj interface{}) {
 		s.logger.Infow("statefulset replicas updated", zap.Int32("replicas", s.replicas))
 	}
 }
+
+func (s *StatefulSetScheduler) reservePlacements(vpod scheduler.VPod, placements []duckv1alpha1.Placement) {
+	existing := vpod.GetPlacements()
+	for _, p := range placements {
+		placed := int32(0)
+		for _, e := range existing {
+			if e.PodName == p.PodName {
+				placed = e.VReplicas
+				break
+			}
+		}
+
+		// Only record placements exceeding existing ones, since the
+		// goal is to prevent pods to be overcommitted.
+		if p.VReplicas > placed {
+			if _, ok := s.reserved[vpod.GetKey()]; !ok {
+				s.reserved[vpod.GetKey()] = make(map[string]int32)
+			}
+
+			// note: track all vreplicas, not only the new ones since
+			// the next time `state()` is called some vreplicas might
+			// have been committed.
+			s.reserved[vpod.GetKey()][p.PodName] = p.VReplicas
+		}
+	}
+}
diff --git a/pkg/common/scheduler/statefulset/state.go b/pkg/common/scheduler/statefulset/state.go
index 975cfad9a..f448c3018 100644
--- a/pkg/common/scheduler/statefulset/state.go
+++ b/pkg/common/scheduler/statefulset/state.go
@@ -21,14 +21,17 @@ import (
 
 	"go.uber.org/zap"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
 	corev1 "k8s.io/client-go/listers/core/v1"
 	"knative.dev/eventing-kafka/pkg/common/scheduler"
 	"knative.dev/pkg/logging"
 )
 
 type stateAccessor interface {
-	// State returns the current state about placed vpods get
-	State() (*state, error)
+	// State returns the current state (snapshot) about placed vpods
+	// Take into account reserved vreplicas and update `reserved` to reflect
+	// the current state.
+	State(reserved map[types.NamespacedName]map[string]int32) (*state, error)
 }
 
 // state provides information about the current scheduling of all vpods
@@ -100,7 +103,7 @@ func newStateBuilder(ctx context.Context, lister scheduler.VPodLister, podCapaci
 	}
 }
 
-func (s *stateBuilder) State() (*state, error) {
+func (s *stateBuilder) State(reserved map[types.NamespacedName]map[string]int32) (*state, error) {
 	vpods, err := s.vpodLister()
 	if err != nil {
 		return nil, err
@@ -109,27 +112,38 @@ func (s *stateBuilder) State() (*state, error) {
 	free := make([]int32, 0, 256)
 	last := int32(-1)
 
+	// keep track of (vpod key, podname) pairs with existing placements
+	withPlacement := make(map[types.NamespacedName]map[string]bool)
+
 	for _, vpod := range vpods {
 		ps := vpod.GetPlacements()
 
+		withPlacement[vpod.GetKey()] = make(map[string]bool)
+
 		for i := 0; i < len(ps); i++ {
 			podName := ps[i].PodName
 			vreplicas := ps[i].VReplicas
 
-			ordinal := ordinalFromPodName(podName)
+			// Account for reserved vreplicas
+			vreplicas = withReserved(vpod.GetKey(), podName, vreplicas, reserved)
 
-			free = grow(free, ordinal, s.capacity)
+			free, last = s.updateFreeCapacity(free, last, podName, vreplicas)
 
-			free[ordinal] -= vreplicas
+			withPlacement[vpod.GetKey()][podName] = true
+		}
+	}
 
-			if free[ordinal] < 0 {
-				// Over committed. The autoscaler will fix it.
-				s.logger.Infow("pod is overcommitted", zap.String("podName", podName))
+	// Account for reserved vreplicas with no prior placements
+	for key, ps := range reserved {
+		for podName, rvreplicas := range ps {
+			if wp, ok := withPlacement[key]; ok {
+				if _, ok := wp[podName]; ok {
+					// already accounted for
+					break
+				}
 			}
 
-			if ordinal > last && free[ordinal] != s.capacity {
-				last = ordinal
-			}
+			free, last = s.updateFreeCapacity(free, last, podName, rvreplicas)
 		}
 	}
 
@@ -159,6 +173,25 @@ func (s *stateBuilder) State() (*state, error) {
 	return &state{free: free, lastOrdinal: last, capacity: s.capacity, schedulerPolicy: s.schedulerPolicy}, nil
 }
 
+func (s *stateBuilder) updateFreeCapacity(free []int32, last int32, podName string, vreplicas int32) ([]int32, int32) {
+	ordinal := ordinalFromPodName(podName)
+	free = grow(free, ordinal, s.capacity)
+
+	free[ordinal] -= vreplicas
+
+	// Assert the pod is not overcommitted
+	if free[ordinal] < 0 {
+		// This should not happen anymore. Log as an error but do not interrupt the current scheduling.
+		s.logger.Errorw("pod is overcommitted", zap.String("podName", podName), zap.Int32("free", free[ordinal]))
+	}
+
+	if ordinal > last && free[ordinal] != s.capacity {
+		last = ordinal
+	}
+
+	return free, last
+}
+
 func grow(slice []int32, ordinal int32, def int32) []int32 {
 	l := int32(len(slice))
 	diff := ordinal - l + 1
@@ -172,3 +205,27 @@ func grow(slice []int32, ordinal int32, def int32) []int32 {
 	}
 	return slice
 }
+
+func withReserved(key types.NamespacedName, podName string, committed int32, reserved map[types.NamespacedName]map[string]int32) int32 {
+	if reserved != nil {
+		if rps, ok := reserved[key]; ok {
+			if rvreplicas, ok := rps[podName]; ok {
+				if committed < rvreplicas {
+					// new placement hasn't been committed yet. Adjust locally
+					return rvreplicas
+				} else if committed == rvreplicas {
+					// new placement has been committed.
+					delete(rps, podName)
+					if len(rps) == 0 {
+						delete(reserved, key)
+					}
+				}
+				// else {
+				// 	// the number of vreplicas has been reduced but not committed yet.
+				// 	// do nothing.
+				// }
+			}
+		}
+	}
+	return committed
+}


Test patch:
diff --git a/pkg/common/scheduler/statefulset/autoscaler_test.go b/pkg/common/scheduler/statefulset/autoscaler_test.go
index 924e8f840..2d009b5cc 100644
--- a/pkg/common/scheduler/statefulset/autoscaler_test.go
+++ b/pkg/common/scheduler/statefulset/autoscaler_test.go
@@ -17,7 +17,6 @@ limitations under the License.
 package statefulset
 
 import (
-	"fmt"
 	"testing"
 	"time"
 
@@ -87,7 +86,7 @@ func TestAutoscaler(t *testing.T) {
 					{PodName: "pod-1", VReplicas: int32(7)}}),
 			},
 			pendings:     int32(3),
-			wantReplicas: int32(2),
+			wantReplicas: int32(3),
 		},
 		{
 			name:     "no replicas, with placements, with pending, not enough capacity",
@@ -171,7 +170,7 @@ func TestAutoscaler(t *testing.T) {
 					{PodName: "pod-1", VReplicas: int32(7)}}),
 			},
 			pendings:     int32(3),
-			wantReplicas: int32(2),
+			wantReplicas: int32(3),
 		},
 		{
 			name:     "with replicas, with placements, with pending, not enough capacity",
@@ -414,11 +413,10 @@ func TestCompactor(t *testing.T) {
 				vpodClient.Append(vpod)
 			}
 
-			state, err := stateAccessor.State()
+			state, err := stateAccessor.State(nil)
 			if err != nil {
 				t.Fatalf("unexpected error: %v", err)
 			}
-			fmt.Println(state)
 
 			autoscaler.mayCompact(state)
 
diff --git a/pkg/common/scheduler/statefulset/state_test.go b/pkg/common/scheduler/statefulset/state_test.go
index e6bcdf361..aba39ad63 100644
--- a/pkg/common/scheduler/statefulset/state_test.go
+++ b/pkg/common/scheduler/statefulset/state_test.go
@@ -24,6 +24,7 @@ import (
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
 	tscheduler "knative.dev/eventing-kafka/pkg/common/scheduler/testing"
 	listers "knative.dev/eventing/pkg/reconciler/testing/v1"
@@ -37,6 +38,7 @@ func TestStateBuilder(t *testing.T) {
 		expected        state
 		freec           int32
 		schedulerPolicy SchedulerPolicyType
+		reserved        map[types.NamespacedName]map[string]int32
 		nodes           []*v1.Node
 		err             error
 	}{
@@ -76,6 +78,62 @@ func TestStateBuilder(t *testing.T) {
 			freec:           int32(24),
 			schedulerPolicy: MAXFILLUP,
 		},
+		{
+			name: "many vpods, with gaps and reserved vreplicas",
+			vpods: [][]duckv1alpha1.Placement{
+				{{PodName: "statefulset-name-0", VReplicas: 1}, {PodName: "statefulset-name-2", VReplicas: 5}},
+				{{PodName: "statefulset-name-1", VReplicas: 0}},
+				{{PodName: "statefulset-name-1", VReplicas: 0}, {PodName: "statefulset-name-3", VReplicas: 0}},
+			},
+			expected: state{capacity: 10, free: []int32{int32(4), int32(10), int32(5), int32(10)}, lastOrdinal: 2, schedulerPolicy: MAXFILLUP},
+			freec:    int32(19),
+			reserved: map[types.NamespacedName]map[string]int32{
+				{Name: "vpod-name-3", Namespace: testNs}: {
+					"statefulset-name-0": 5,
+				},
+			},
+			schedulerPolicy: MAXFILLUP,
+		},
+		{
+			name: "many vpods, with gaps and reserved vreplicas on existing and new placements, fully committed",
+			vpods: [][]duckv1alpha1.Placement{
+				{{PodName: "statefulset-name-0", VReplicas: 1}, {PodName: "statefulset-name-2", VReplicas: 5}},
+				{{PodName: "statefulset-name-1", VReplicas: 0}},
+				{{PodName: "statefulset-name-1", VReplicas: 0}, {PodName: "statefulset-name-3", VReplicas: 0}},
+			},
+			expected: state{capacity: 10, free: []int32{int32(4), int32(7), int32(5), int32(10), int32(5)}, lastOrdinal: 4, schedulerPolicy: MAXFILLUP},
+			freec:    int32(31),
+			reserved: map[types.NamespacedName]map[string]int32{
+				{Name: "vpod-name-3", Namespace: "vpod-ns-3"}: {
+					"statefulset-name-4": 5,
+				},
+				{Name: "vpod-name-4", Namespace: "vpod-ns-4"}: {
+					"statefulset-name-0": 5,
+					"statefulset-name-1": 3,
+				},
+			},
+			schedulerPolicy: MAXFILLUP,
+		},
+		{
+			name: "many vpods, with gaps and reserved vreplicas on existing and new placements, partially committed",
+			vpods: [][]duckv1alpha1.Placement{
+				{{PodName: "statefulset-name-0", VReplicas: 1}, {PodName: "statefulset-name-2", VReplicas: 5}},
+				{{PodName: "statefulset-name-1", VReplicas: 0}},
+				{{PodName: "statefulset-name-1", VReplicas: 0}, {PodName: "statefulset-name-3", VReplicas: 0}},
+			},
+			expected: state{capacity: 10, free: []int32{int32(4), int32(7), int32(2), int32(10)}, lastOrdinal: 2, schedulerPolicy: MAXFILLUP},
+			freec:    int32(13),
+			reserved: map[types.NamespacedName]map[string]int32{
+				{Name: "vpod-name-0", Namespace: "vpod-ns-0"}: {
+					"statefulset-name-2": 8,
+				},
+				{Name: "vpod-name-4", Namespace: "vpod-ns-4"}: {
+					"statefulset-name-0": 5,
+					"statefulset-name-1": 3,
+				},
+			},
+			schedulerPolicy: MAXFILLUP,
+		},
 		{
 			name:            "no vpods, all nodes with zone labels",
 			vpods:           [][]duckv1alpha1.Placement{},
@@ -119,7 +177,7 @@ func TestStateBuilder(t *testing.T) {
 
 			ls := listers.NewListers(nodelist)
 			stateBuilder := newStateBuilder(ctx, vpodClient.List, int32(10), tc.schedulerPolicy, ls.GetNodeLister())
-			state, err := stateBuilder.State()
+			state, err := stateBuilder.State(tc.reserved)
 			if err != nil {
 				t.Fatal("unexpected error", err)
 			}