[release-v1.17] Add missing backports (#1674)

* [release-1.17] Reduce mt-broker-controller memory usage with namespaced endpoint informer (knative#8421)

* Reduce mt-broker-controller memory usage with namespaced endpoint informer

Currently, the mt-broker-controller is using a cluster-wide endpoints
informer but it actually only uses endpoints in the "SYSTEM_NAMESPACE".

Using the namespaced informer factory ensures that the watcher
is only watching endpoints in the `knative-eventing` (also known as
`SYSTEM_NAMESPACE`) namespace.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* Start informer

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

---------

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>
Co-authored-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* [release-1.17] Scheduler: Resync reserved periodically to keep state consistent (knative#8453)

Scheduler: Resync reserved periodically to keep state consistent

Add resyncReserved removes deleted vPods from reserved to keep the
state consistent when leadership changes (Promote / Demote).

`initReserved` is not enough since the vPod lister can be stale.

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>
Co-authored-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* [release-1.17] Scheduler: log expected vreplicas by vpod (knative#8462)

Scheduler: log expected vreplicas by vpod

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>
Co-authored-by: Pierangelo Di Pilato <pierdipi@redhat.com>

* [release-1.17] Compare the entire PodTemplateSpec, instead of just its PodSpec (knative#8560)

Compare the entire PodTemplateSpec, instead of just its PodSpec

Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>
Co-authored-by: Matthias Wessendorf <mwessend@redhat.com>

* [release-1.17] Fix hardcoded knative-eventing namespace (knative#8577)

* dizzy: Remove hard-coded knative-eventing namespace and replace with system.Namespace() and touch test to get that properly injected

Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

* broom: Remove yet another hard-coded instance of knative-eventing

Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>

---------

Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>
Co-authored-by: Matthias Wessendorf <mwessend@redhat.com>

* [release-1.17] Guard reserved access with lock and create vpods in tests (knative#8504)

* Guard reserved access with lock in tests

* Create vpods in test

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>

---------

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>
Co-authored-by: Pierangelo Di Pilato <pierdipi@redhat.com>

---------

Signed-off-by: Pierangelo Di Pilato <pierdipi@redhat.com>
Signed-off-by: Matthias Wessendorf <mwessend@redhat.com>
Co-authored-by: Knative Prow Robot <automation+prow-robot@knative.team>
Co-authored-by: Pierangelo Di Pilato <pierdipi@redhat.com>
Co-authored-by: Matthias Wessendorf <mwessend@redhat.com>

Author: 4 people

pkg/apis/messaging/v1/in_memory_channel_validation.go

@@ -22,11 +22,13 @@ import (
 
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/kmp"
+	"knative.dev/pkg/system"
 
 	"knative.dev/eventing/pkg/apis/eventing"
+	_ "knative.dev/pkg/system/testing"
 )
 
-const eventingControllerSAName = "system:serviceaccount:knative-eventing:eventing-controller"
+var eventingControllerSAName = fmt.Sprintf("%s:%s:%s", "system:serviceaccount", system.Namespace(), "eventing-controller")
 
 func (imc *InMemoryChannel) Validate(ctx context.Context) *apis.FieldError {
 	errs := imc.Spec.Validate(ctx).ViaField("spec")

pkg/apis/messaging/v1/in_memory_channel_validation_test.go

@@ -17,8 +17,11 @@ limitations under the License.
 package v1
 
 import (
+	"fmt"
 	"testing"
 
+	"knative.dev/pkg/system"
+
 	"golang.org/x/net/context"
 	authenticationv1 "k8s.io/api/authentication/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -139,7 +142,7 @@ func TestInMemoryChannelValidation(t *testing.T) {
 		want: func() *apis.FieldError {
 			diff, _ := kmp.ShortDiff(validIMCSingleSubscriber.Spec.Subscribers, validIMCTwoSubscribers.Spec.Subscribers)
 			return &apis.FieldError{
-				Message: "Channel.Spec.Subscribers changed by user test-user which was not the system:serviceaccount:knative-eventing:eventing-controller service account",
+				Message: fmt.Sprintf("%s:%s:%s", "Channel.Spec.Subscribers changed by user test-user which was not the system:serviceaccount", system.Namespace(), "eventing-controller service account"),
 				Paths:   []string{"spec.subscribers"},
 				Details: diff,
 			}

pkg/reconciler/broker/controller.go

@@ -25,11 +25,11 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"knative.dev/pkg/apis"
 	configmapinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/configmap"
-	endpointsinformer "knative.dev/pkg/client/injection/kube/informers/core/v1/endpoints"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection/clients/dynamicclient"
 	secretinformer "knative.dev/pkg/injection/clients/namespacedkube/informers/core/v1/secret"
+	namespacedinformerfactory "knative.dev/pkg/injection/clients/namespacedkube/informers/factory"
 	"knative.dev/pkg/logging"
 	pkgreconciler "knative.dev/pkg/reconciler"
 	"knative.dev/pkg/resolver"
@@ -69,7 +69,12 @@ func NewController(
 	logger := logging.FromContext(ctx)
 	brokerInformer := brokerinformer.Get(ctx)
 	subscriptionInformer := subscriptioninformer.Get(ctx)
-	endpointsInformer := endpointsinformer.Get(ctx)
+
+	endpointsInformer := namespacedinformerfactory.Get(ctx).Core().V1().Endpoints()
+	if err := controller.StartInformers(ctx.Done(), endpointsInformer.Informer()); err != nil {
+		logger.Fatalw("Failed to start namespaced endpoints informer", zap.Error(err))
+	}
+
 	configmapInformer := configmapinformer.Get(ctx)
 	secretInformer := secretinformer.Get(ctx)
 	eventPolicyInformer := eventpolicyinformer.Get(ctx)

pkg/reconciler/broker/resources/eventpolicy.go

@@ -17,17 +17,21 @@ limitations under the License.
 package resources
 
 import (
+	"fmt"
+
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/utils/ptr"
 	eventingduckv1 "knative.dev/eventing/pkg/apis/duck/v1"
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	eventingv1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
 	"knative.dev/pkg/kmeta"
+	"knative.dev/pkg/system"
 )
 
+var OIDCBrokerSub = fmt.Sprintf("%s:%s:%s", "system:serviceaccount", system.Namespace(), "mt-broker-ingress-oidc")
+
 const (
 	BackingChannelEventPolicyLabelPrefix = "eventing.knative.dev/"
-	OIDCBrokerSub                        = "system:serviceaccount:knative-eventing:mt-broker-ingress-oidc"
 	brokerKind                           = "Broker"
 )
 

pkg/reconciler/containersource/containersource.go

@@ -109,8 +109,8 @@ func (r *Reconciler) reconcileReceiveAdapter(ctx context.Context, source *v1.Con
 		return nil, fmt.Errorf("getting Deployment: %v", err)
 	} else if !metav1.IsControlledBy(ra, source) {
 		return nil, fmt.Errorf("deployment %q is not owned by ContainerSource %q", ra.Name, source.Name)
-	} else if r.podSpecChanged(&ra.Spec.Template.Spec, &expected.Spec.Template.Spec) {
-		ra.Spec.Template.Spec = expected.Spec.Template.Spec
+	} else if r.podTemplateChanged(&ra.Spec.Template, &expected.Spec.Template) {
+		ra.Spec.Template = expected.Spec.Template
 		ra, err = r.kubeClientSet.AppsV1().Deployments(expected.Namespace).Update(ctx, ra, metav1.UpdateOptions{})
 		if err != nil {
 			return nil, fmt.Errorf("updating Deployment: %v", err)
@@ -159,6 +159,10 @@ func (r *Reconciler) podSpecChanged(have *corev1.PodSpec, want *corev1.PodSpec)
 	return !equality.Semantic.DeepDerivative(want, have)
 }
 
+func (r *Reconciler) podTemplateChanged(have *corev1.PodTemplateSpec, want *corev1.PodTemplateSpec) bool {
+	return !equality.Semantic.DeepDerivative(want, have)
+}
+
 func (r *Reconciler) sinkBindingSpecChanged(have *v1.SinkBindingSpec, want *v1.SinkBindingSpec) bool {
 	return !equality.Semantic.DeepDerivative(want, have)
 }

pkg/scheduler/scheduler.go

@@ -66,6 +66,8 @@ func (f SchedulerFunc) Schedule(ctx context.Context, vpod VPod) ([]duckv1alpha1.
 // VPod represents virtual replicas placed into real Kubernetes pods
 // The scheduler is responsible for placing VPods
 type VPod interface {
+	GetDeletionTimestamp() *metav1.Time
+
 	// GetKey returns the VPod key (namespace/name).
 	GetKey() types.NamespacedName
 

pkg/scheduler/state/state.go

@@ -278,23 +278,25 @@ func isPodUnschedulable(pod *v1.Pod) bool {
 func (s *State) MarshalJSON() ([]byte, error) {
 
 	type S struct {
-		FreeCap         []int32                     `json:"freeCap"`
-		SchedulablePods []int32                     `json:"schedulablePods"`
-		Capacity        int32                       `json:"capacity"`
-		Replicas        int32                       `json:"replicas"`
-		StatefulSetName string                      `json:"statefulSetName"`
-		PodSpread       map[string]map[string]int32 `json:"podSpread"`
-		Pending         map[string]int32            `json:"pending"`
+		FreeCap                []int32                     `json:"freeCap"`
+		SchedulablePods        []int32                     `json:"schedulablePods"`
+		Capacity               int32                       `json:"capacity"`
+		Replicas               int32                       `json:"replicas"`
+		StatefulSetName        string                      `json:"statefulSetName"`
+		PodSpread              map[string]map[string]int32 `json:"podSpread"`
+		Pending                map[string]int32            `json:"pending"`
+		ExpectedVReplicaByVPod map[string]int32            `json:"expectedVReplicaByVPod"`
 	}
 
 	sj := S{
-		FreeCap:         s.FreeCap,
-		SchedulablePods: s.SchedulablePods,
-		Capacity:        s.Capacity,
-		Replicas:        s.Replicas,
-		StatefulSetName: s.StatefulSetName,
-		PodSpread:       ToJSONable(s.PodSpread),
-		Pending:         toJSONablePending(s.Pending),
+		FreeCap:                s.FreeCap,
+		SchedulablePods:        s.SchedulablePods,
+		Capacity:               s.Capacity,
+		Replicas:               s.Replicas,
+		StatefulSetName:        s.StatefulSetName,
+		PodSpread:              ToJSONable(s.PodSpread),
+		Pending:                toJSONablePending(s.Pending),
+		ExpectedVReplicaByVPod: toJSONablePending(s.ExpectedVReplicaByVPod),
 	}
 
 	return json.Marshal(sj)

pkg/scheduler/statefulset/autoscaler.go

@@ -205,17 +205,19 @@ func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool) err
 		return err
 	}
 
-	logger.Debugw("checking adapter capacity",
-		zap.Int32("replicas", scale.Spec.Replicas),
-		zap.Any("state", state))
-
 	newReplicas := integer.Int32Max(int32(math.Ceil(float64(state.TotalExpectedVReplicas())/float64(state.Capacity))), a.minReplicas)
 
 	// Only scale down if permitted
 	if !attemptScaleDown && newReplicas < scale.Spec.Replicas {
 		newReplicas = scale.Spec.Replicas
 	}
 
+	logger.Debugw("checking adapter capacity",
+		zap.Bool("attemptScaleDown", attemptScaleDown),
+		zap.Int32("replicas", scale.Spec.Replicas),
+		zap.Int32("newReplicas", newReplicas),
+		zap.Any("state", state))
+
 	if newReplicas != scale.Spec.Replicas {
 		scale.Spec.Replicas = newReplicas
 		logger.Infow("updating adapter replicas", zap.Int32("replicas", scale.Spec.Replicas))

pkg/scheduler/statefulset/scheduler.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"sort"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.uber.org/zap"
@@ -114,6 +115,11 @@ type StatefulSetScheduler struct {
 	// replicas is the (cached) number of statefulset replicas.
 	replicas int32
 
+	// isLeader signals whether a given Scheduler instance is leader or not.
+	// The autoscaler is considered the leader when ephemeralLeaderElectionObject is in a
+	// bucket where we've been promoted.
+	isLeader atomic.Bool
+
 	// 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
@@ -130,6 +136,9 @@ func (s *StatefulSetScheduler) Promote(b reconciler.Bucket, enq func(reconciler.
 	if !b.Has(ephemeralLeaderElectionObject) {
 		return nil
 	}
+	// The demoted bucket has the ephemeralLeaderElectionObject, so we are not leader anymore.
+	// Flip the flag after running initReserved.
+	defer s.isLeader.Store(true)
 
 	if v, ok := s.autoscaler.(reconciler.LeaderAware); ok {
 		return v.Promote(b, enq)
@@ -151,6 +160,9 @@ func (s *StatefulSetScheduler) initReserved() error {
 
 	s.reserved = make(map[types.NamespacedName]map[string]int32, len(vPods))
 	for _, vPod := range vPods {
+		if !vPod.GetDeletionTimestamp().IsZero() {
+			continue
+		}
 		s.reserved[vPod.GetKey()] = make(map[string]int32, len(vPod.GetPlacements()))
 		for _, placement := range vPod.GetPlacements() {
 			s.reserved[vPod.GetKey()][placement.PodName] += placement.VReplicas
@@ -159,8 +171,41 @@ func (s *StatefulSetScheduler) initReserved() error {
 	return nil
 }
 
+// resyncReserved removes deleted vPods from reserved to keep the state consistent when leadership
+// changes (Promote / Demote).
+// initReserved is not enough since the vPod lister can be stale.
+func (s *StatefulSetScheduler) resyncReserved() error {
+	if !s.isLeader.Load() {
+		return nil
+	}
+
+	vPods, err := s.vpodLister()
+	if err != nil {
+		return fmt.Errorf("failed to list vPods during reserved resync: %w", err)
+	}
+	vPodsByK := vPodsByKey(vPods)
+
+	s.reservedMu.Lock()
+	defer s.reservedMu.Unlock()
+
+	for key := range s.reserved {
+		vPod, ok := vPodsByK[key]
+		if !ok || vPod == nil {
+			delete(s.reserved, key)
+		}
+	}
+
+	return nil
+}
+
 // Demote implements reconciler.LeaderAware.
 func (s *StatefulSetScheduler) Demote(b reconciler.Bucket) {
+	if !b.Has(ephemeralLeaderElectionObject) {
+		return
+	}
+	// The demoted bucket has the ephemeralLeaderElectionObject, so we are not leader anymore.
+	defer s.isLeader.Store(false)
+
 	if v, ok := s.autoscaler.(reconciler.LeaderAware); ok {
 		v.Demote(b)
 	}
@@ -208,6 +253,17 @@ func newStatefulSetScheduler(ctx context.Context,
 		sif.Shutdown()
 	}()
 
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-time.After(cfg.RefreshPeriod * 3):
+				_ = s.resyncReserved()
+			}
+		}
+	}()
+
 	return s
 }
 
@@ -561,3 +617,11 @@ func upsertPlacements(placements []duckv1alpha1.Placement, placement duckv1alpha
 	}
 	return placements
 }
+
+func vPodsByKey(vPods []scheduler.VPod) map[types.NamespacedName]scheduler.VPod {
+	r := make(map[types.NamespacedName]scheduler.VPod, len(vPods))
+	for _, vPod := range vPods {
+		r[vPod.GetKey()] = vPod
+	}
+	return r
+}

pkg/scheduler/statefulset/scheduler_test.go

@@ -708,6 +708,21 @@ func TestStatefulsetScheduler(t *testing.T) {
 
 			vpodClient := tscheduler.NewVPodClient()
 			vpod := vpodClient.Create(vpodNamespace, vpodName, tc.vreplicas, tc.placements)
+			for vPodKey, res := range tc.initialReserved {
+				if vPodKey.Namespace == vpodNamespace && vPodKey.Name == vpodName {
+					continue
+				}
+				var placements []duckv1alpha1.Placement
+				count := int32(0)
+				for pod, vReplicas := range res {
+					count += vReplicas
+					placements = append(placements, duckv1alpha1.Placement{
+						PodName:   pod,
+						VReplicas: vReplicas,
+					})
+				}
+				vpodClient.Create(vPodKey.Namespace, vPodKey.Name, count, placements)
+			}
 
 			for i := int32(0); i < tc.replicas; i++ {
 				nodeName := "node" + fmt.Sprint(i)
@@ -745,7 +760,9 @@ func TestStatefulsetScheduler(t *testing.T) {
 				t.Fatal("unexpected error", err)
 			}
 			if tc.initialReserved != nil {
+				s.reservedMu.Lock()
 				s.reserved = tc.initialReserved
+				s.reservedMu.Unlock()
 			}
 
 			// Give some time for the informer to notify the scheduler and set the number of replicas