💫 Peer review with pierdipi on resolving the dependency hell

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

go.mod

@@ -39,10 +39,10 @@ require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/google/gofuzz v1.2.0
 	github.com/kedacore/keda/v2 v2.8.1
-	knative.dev/eventing v0.36.1-0.20230309084530-05f6d84ad43c
-	knative.dev/hack v0.0.0-20230228173453-3de51aff69a3
-	knative.dev/pkg v0.0.0-20230309013522-c5dd1d1264ba
-	knative.dev/reconciler-test v0.0.0-20230307080119-d52257e4e606
+	knative.dev/eventing v0.36.1-0.20230414134751-a5b4810a24b0
+	knative.dev/hack v0.0.0-20230412013450-4b3f2300c1ad
+	knative.dev/pkg v0.0.0-20230413131852-ce10b064f603
+	knative.dev/reconciler-test v0.0.0-20230413132853-06956b6259d6
 	sigs.k8s.io/controller-runtime v0.12.3
 )
 

go.sum

@@ -1518,14 +1518,14 @@ k8s.io/utils v0.0.0-20200912215256-4140de9c8800/go.mod h1:jPW/WVKK9YHAvNhRxK0md/
 k8s.io/utils v0.0.0-20210819203725-bdf08cb9a70a/go.mod h1:jPW/WVKK9YHAvNhRxK0md/EJ228hCsBRufyofKtW8HA=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2 h1:GfD9OzL11kvZN5iArC6oTS7RTj7oJOIfnislxYlqTj8=
 k8s.io/utils v0.0.0-20221108210102-8e77b1f39fe2/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-knative.dev/eventing v0.36.1-0.20230309084530-05f6d84ad43c h1:idpDTj8ipzWAy7kDZSq3uk4taQoOAAWJGKt84ScmqhE=
-knative.dev/eventing v0.36.1-0.20230309084530-05f6d84ad43c/go.mod h1:8mo69M3WWCQWNyWb2G+g2Jyeh8eDOiQud4Xd8jLr57Q=
-knative.dev/hack v0.0.0-20230228173453-3de51aff69a3 h1:Jt3n+AZsZHZaFhk/A1NnboAvgjV+hvvyeyyuIX/hUx0=
-knative.dev/hack v0.0.0-20230228173453-3de51aff69a3/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
-knative.dev/pkg v0.0.0-20230309013522-c5dd1d1264ba h1:THY9ox1maZNYcmvYBJ3hhtjqHp9Ci0fDsF1t+GIoiOo=
-knative.dev/pkg v0.0.0-20230309013522-c5dd1d1264ba/go.mod h1:S+KfTInuwEkZSTwvWqrWZV/TEw6ps51GUGaSC1Fnbe0=
-knative.dev/reconciler-test v0.0.0-20230307080119-d52257e4e606 h1:gJY32/uUublG1WSFH0QDstxE3aEQiU61JLr+t0uLp5c=
-knative.dev/reconciler-test v0.0.0-20230307080119-d52257e4e606/go.mod h1:2472v1Zg0stETnG6t2qpnyiJadgOJHb2iLaAgiOHa68=
+knative.dev/eventing v0.36.1-0.20230414134751-a5b4810a24b0 h1:fhVP6QyjYRyBVx7IUKEwWRcBOw6LYkuxJ7mgkt0fIRU=
+knative.dev/eventing v0.36.1-0.20230414134751-a5b4810a24b0/go.mod h1:vEQ32TrhYNoafRkuw4wxIyb5wSQ65yfejCfS611jmI8=
+knative.dev/hack v0.0.0-20230412013450-4b3f2300c1ad h1:+5MpC265m9pjmJl+popG9XO9G1l+Rq1py9ldqBHwA68=
+knative.dev/hack v0.0.0-20230412013450-4b3f2300c1ad/go.mod h1:yk2OjGDsbEnQjfxdm0/HJKS2WqTLEFg/N6nUs6Rqx3Q=
+knative.dev/pkg v0.0.0-20230413131852-ce10b064f603 h1:vBKLFZmoi5hniD9Fi5lVbxnrO8DLeIhBRDXgRB+cqAo=
+knative.dev/pkg v0.0.0-20230413131852-ce10b064f603/go.mod h1:Xa/jM3LpUnfQabS0kKR9sMTaDn4absCVvwkdNlwRwHc=
+knative.dev/reconciler-test v0.0.0-20230413132853-06956b6259d6 h1:zUBZkr9kTSzYBasHZw2WnMcBJy5COZS5Xau9ThmByFo=
+knative.dev/reconciler-test v0.0.0-20230413132853-06956b6259d6/go.mod h1:JwK7KUivj9TX7gJ6SAFfNxhmAfYc45kyASeRT8OG+pM=
 pgregory.net/rapid v0.3.3 h1:jCjBsY4ln4Atz78QoBWxUEvAHaFyNDQg9+WU62aCn1U=
 pgregory.net/rapid v0.3.3/go.mod h1:UYpPVyjFHzYBGHIxLFoupi8vwk6rXNzRY9OMvVxFIOU=
 rsc.io/binaryregexp v0.2.0/go.mod h1:qTv7/COck+e2FymRvadv62gMdZztPaShugOCi3I+8D8=

hack/tools.go

@@ -29,7 +29,6 @@ import (
 
 	// Test images from eventing
 	_ "knative.dev/eventing/test/test_images/event-sender"
-	_ "knative.dev/eventing/test/test_images/heartbeats"
 	_ "knative.dev/eventing/test/test_images/performance"
 	_ "knative.dev/eventing/test/test_images/print"
 	_ "knative.dev/eventing/test/test_images/recordevents"

test/e2e_new_channel/tracing_test.go

@@ -29,7 +29,6 @@ import (
 	tracinghelper "knative.dev/eventing/test/conformance/helpers/tracing"
 	"knative.dev/eventing/test/rekt/resources/channel_impl"
 	"knative.dev/eventing/test/rekt/resources/subscription"
-	"knative.dev/eventing/test/rekt/resources/svc"
 	"knative.dev/pkg/system"
 	_ "knative.dev/pkg/system/testing"
 	"knative.dev/reconciler-test/pkg/environment"
@@ -38,6 +37,7 @@ import (
 	"knative.dev/reconciler-test/pkg/feature"
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
+	"knative.dev/reconciler-test/pkg/resources/service"
 
 	"knative.dev/eventing-kafka-broker/test/pkg/tracing"
 )
@@ -72,7 +72,7 @@ func eventWithTraceExported() *feature.Feature {
 	f.Setup("install channel", channel_impl.Install(channelName))
 	f.Setup("install subscription", subscription.Install(subName,
 		subscription.WithChannel(channel_impl.AsRef(channelName)),
-		subscription.WithSubscriber(svc.AsRef(sinkName), ""),
+		subscription.WithSubscriber(service.AsKReference(sinkName), ""),
 	))
 
 	f.Setup("subscription is ready", subscription.IsReady(subName))

vendor/knative.dev/eventing/pkg/apis/feature/features.go

@@ -34,6 +34,17 @@ const (
 	// Allowed neither explicitly disables or enables a behavior.
 	// eg. allow a client to control behavior with an annotation or allow a new value through validation.
 	Allowed Flag = "Allowed"
+	// Strict is only applicable to the TransportEncryption feature.
+	// The following applies:
+	// - Addressables must not accept events to non-HTTPS endpoints
+	// - Addressables must only advertise HTTPS endpoints
+	Strict Flag = "Strict"
+	// Permissive is only applicable to the TransportEncryption feature.
+	// The following applies:
+	// - Addressables should accept events at both HTTP and HTTPS endpoints
+	// - Addressables should advertise both HTTP and HTTPS endpoints
+	// - Producers should prefer to send events to HTTPS endpoints, if available
+	Permissive Flag = "Permissive"
 )
 
 // Flags is a map containing all the enabled/disabled flags for the experimental features.
@@ -50,6 +61,16 @@ func (e Flags) IsAllowed(featureName string) bool {
 	return e.IsEnabled(featureName) || (e != nil && e[featureName] == Allowed)
 }
 
+// IsPermissiveTransportEncryption returns true if the TransportEncryption feature is in Permissive mode.
+func (e Flags) IsPermissiveTransportEncryption() bool {
+	return e != nil && e[TransportEncryption] == Permissive
+}
+
+// IsStrictTransportEncryption returns true if the TransportEncryption feature is in Strict mode.
+func (e Flags) IsStrictTransportEncryption() bool {
+	return e != nil && e[TransportEncryption] == Strict
+}
+
 // NewFlagsConfigFromMap creates a Flags from the supplied Map
 func NewFlagsConfigFromMap(data map[string]string) (Flags, error) {
 	flags := Flags{}
@@ -66,6 +87,10 @@ func NewFlagsConfigFromMap(data map[string]string) (Flags, error) {
 			flags[sanitizedKey] = Disabled
 		} else if strings.EqualFold(v, string(Enabled)) {
 			flags[sanitizedKey] = Enabled
+		} else if strings.EqualFold(v, string(Permissive)) {
+			flags[sanitizedKey] = Permissive
+		} else if strings.EqualFold(v, string(Strict)) {
+			flags[sanitizedKey] = Strict
 		} else {
 			return Flags{}, fmt.Errorf("cannot parse the boolean flag '%s' = '%s'. Allowed values: [true, false]", k, v)
 		}

vendor/knative.dev/eventing/pkg/apis/feature/flag_names.go

@@ -17,9 +17,10 @@ limitations under the License.
 package feature
 
 const (
-	KReferenceGroup    = "kreference-group"
-	DeliveryRetryAfter = "delivery-retryafter"
-	DeliveryTimeout    = "delivery-timeout"
-	KReferenceMapping  = "kreference-mapping"
-	NewTriggerFilters  = "new-trigger-filters"
+	KReferenceGroup     = "kreference-group"
+	DeliveryRetryAfter  = "delivery-retryafter"
+	DeliveryTimeout     = "delivery-timeout"
+	KReferenceMapping   = "kreference-mapping"
+	NewTriggerFilters   = "new-trigger-filters"
+	TransportEncryption = "transport-encryption"
 )

vendor/knative.dev/eventing/pkg/kncloudevents/retries.go

@@ -83,7 +83,7 @@ func RetryConfigFromDeliverySpec(spec v1.DeliverySpec) (RetryConfig, error) {
 
 	retryConfig := NoRetries()
 
-	retryConfig.CheckRetry = RetryIfGreaterThan300
+	retryConfig.CheckRetry = SelectiveRetry
 
 	if spec.Retry != nil {
 		retryConfig.RetryMax = int(*spec.Retry)
@@ -131,11 +131,6 @@ func RetryConfigFromDeliverySpec(spec v1.DeliverySpec) (RetryConfig, error) {
 	return retryConfig, nil
 }
 
-// RetryIfGreaterThan300 is a simple default implementation
-func RetryIfGreaterThan300(_ context.Context, response *http.Response, err error) (bool, error) {
-	return !(response != nil && (response.StatusCode < 300 && response.StatusCode != -1)), err
-}
-
 // SelectiveRetry is an alternative function to determine whether to retry based on response
 //
 // Note - Returning true indicates a retry should occur.  Returning an error will result in that

vendor/knative.dev/eventing/pkg/scheduler/statefulset/autoscaler.go

@@ -20,13 +20,16 @@ import (
 	"context"
 	"math"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.uber.org/zap"
 	v1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/wait"
 	clientappsv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
+	"knative.dev/pkg/reconciler"
 
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	"knative.dev/pkg/logging"
@@ -35,6 +38,16 @@ import (
 	st "knative.dev/eventing/pkg/scheduler/state"
 )
 
+var (
+	// ephemeralLeaderElectionObject is the key used to check whether a given autoscaler instance
+	// is leader or not.
+	// This is an ephemeral key and must be kept stable and unmodified across releases.
+	ephemeralLeaderElectionObject = types.NamespacedName{
+		Namespace: "knative-eventing",
+		Name:      "autoscaler-ephemeral",
+	}
+)
+
 type Autoscaler interface {
 	// Start runs the autoscaler until cancelled.
 	Start(ctx context.Context)
@@ -59,9 +72,35 @@ type autoscaler struct {
 	// refreshPeriod is how often the autoscaler tries to scale down the statefulset
 	refreshPeriod time.Duration
 	lock          sync.Locker
+
+	// isLeader signals whether a given autoscaler 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
+}
+
+var (
+	_ reconciler.LeaderAware = &autoscaler{}
+)
+
+// Promote implements reconciler.LeaderAware.
+func (a *autoscaler) Promote(b reconciler.Bucket, _ func(reconciler.Bucket, types.NamespacedName)) error {
+	if b.Has(ephemeralLeaderElectionObject) {
+		// The promoted bucket has the ephemeralLeaderElectionObject, so we are leader.
+		a.isLeader.Store(true)
+	}
+	return nil
 }
 
-func newAutoscaler(ctx context.Context, cfg *Config, stateAccessor st.StateAccessor) Autoscaler {
+// Demote implements reconciler.LeaderAware.
+func (a *autoscaler) Demote(b reconciler.Bucket) {
+	if b.Has(ephemeralLeaderElectionObject) {
+		// The demoted bucket has the ephemeralLeaderElectionObject, so we are not leader anymore.
+		a.isLeader.Store(false)
+	}
+}
+
+func newAutoscaler(ctx context.Context, cfg *Config, stateAccessor st.StateAccessor) *autoscaler {
 	return &autoscaler{
 		logger:            logging.FromContext(ctx),
 		statefulSetClient: kubeclient.Get(ctx).AppsV1().StatefulSets(cfg.StatefulSetNamespace),
@@ -73,6 +112,7 @@ func newAutoscaler(ctx context.Context, cfg *Config, stateAccessor st.StateAcces
 		capacity:          cfg.PodCapacity,
 		refreshPeriod:     cfg.RefreshPeriod,
 		lock:              new(sync.Mutex),
+		isLeader:          atomic.Bool{},
 	}
 }
 
@@ -111,6 +151,9 @@ func (a *autoscaler) syncAutoscale(ctx context.Context, attemptScaleDown bool, p
 }
 
 func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool, pending int32) error {
+	if !a.isLeader.Load() {
+		return nil
+	}
 	state, err := a.stateAccessor.State(nil)
 	if err != nil {
 		a.logger.Info("error while refreshing scheduler state (will retry)", zap.Error(err))

vendor/knative.dev/eventing/pkg/scheduler/statefulset/scheduler.go

@@ -32,6 +32,7 @@ import (
 	corev1listers "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/utils/integer"
+	"knative.dev/pkg/reconciler"
 
 	kubeclient "knative.dev/pkg/client/injection/kube/client"
 	statefulsetinformer "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset"
@@ -145,11 +146,31 @@ type StatefulSetScheduler struct {
 	reserved map[types.NamespacedName]map[string]int32
 }
 
+var (
+	_ reconciler.LeaderAware = &StatefulSetScheduler{}
+	_ scheduler.Scheduler    = &StatefulSetScheduler{}
+)
+
+// Promote implements reconciler.LeaderAware.
+func (s *StatefulSetScheduler) Promote(b reconciler.Bucket, enq func(reconciler.Bucket, types.NamespacedName)) error {
+	if v, ok := s.autoscaler.(reconciler.LeaderAware); ok {
+		return v.Promote(b, enq)
+	}
+	return nil
+}
+
+// Demote implements reconciler.LeaderAware.
+func (s *StatefulSetScheduler) Demote(b reconciler.Bucket) {
+	if v, ok := s.autoscaler.(reconciler.LeaderAware); ok {
+		v.Demote(b)
+	}
+}
+
 func newStatefulSetScheduler(ctx context.Context,
 	cfg *Config,
 	stateAccessor st.StateAccessor,
 	autoscaler Autoscaler,
-	podlister corev1listers.PodNamespaceLister) scheduler.Scheduler {
+	podlister corev1listers.PodNamespaceLister) *StatefulSetScheduler {
 
 	scheduler := &StatefulSetScheduler{
 		ctx:                  ctx,

vendor/knative.dev/eventing/test/README.md

@@ -136,7 +136,7 @@ build and push the test images used by the e2e tests. It requires:
   to be set
 - You to be
   [authenticated with your `KO_DOCKER_REPO`](https://github.com/knative/serving/blob/main/DEVELOPMENT.md#environment-setup)
-- [`docker`](https://docs.docker.com/install/) to be installed
+- [`docker`](https://docs.docker.com/get-docker/) to be installed
 
 `PLATFORM` environment variable is optional. If it is specified, test images
 will be built for specific hardware architecture, according to its value (for

vendor/knative.dev/eventing/test/e2e-common.sh

@@ -65,6 +65,9 @@ export SYSTEM_NAMESPACE
 SYSTEM_NAMESPACE="${SYSTEM_NAMESPACE:-"knative-eventing-"$(head -c 128 < \
   /dev/urandom | LC_CTYPE=C tr -dc 'a-z0-9' | fold -w 10 | head -n 1)}"
 
+CERT_MANAGER_NAMESPACE="cert-manager"
+export CERT_MANAGER_NAMESPACE
+
 # Latest release. If user does not supply this as a flag, the latest
 # tagged release on the current branch will be used.
 readonly LATEST_RELEASE_VERSION=$(latest_version)
@@ -82,6 +85,8 @@ function knative_setup() {
   enable_sugar || fail_test "Could not enable Sugar Controller Injection"
 
   unleash_duck || fail_test "Could not unleash the chaos duck"
+
+  install_cert_manager || fail_test "Could not install Cert Manager"
 }
 
 function scale_controlplane() {
@@ -372,3 +377,10 @@ function wait_for_file() {
   done
   return 0
 }
+
+function install_cert_manager() {
+  kubectl apply -f third_party/cert-manager/01-cert-manager.crds.yaml
+  kubectl apply -f third_party/cert-manager/02-cert-manager.yaml
+
+  wait_until_pods_running "$CERT_MANAGER_NAMESPACE" || fail_test "Failed to install cert manager"
+}

vendor/knative.dev/eventing/test/rekt/resources/containersource/containersource.yaml

@@ -49,7 +49,7 @@ spec:
     spec:
       containers:
       - name: heartbeats
-        image: ko://knative.dev/eventing/test/test_images/heartbeats
+        image: ko://knative.dev/eventing/cmd/heartbeats
         imagePullPolicy: IfNotPresent
         args:
         - {{ .args }}