Merge branch 'knative-extensions/main' into main

OWNERS_ALIASES

@@ -2,6 +2,9 @@
 # Do not modify this file, instead modify peribolos/knative-extensions.yaml
 
 aliases:
+  autoscaler-keda-approvers:
+  - ReToCode
+  - skonto
   backstage-plugins-reviewers:
   - aliok
   - cali0707
@@ -14,11 +17,9 @@ aliases:
   - pierdipi
   client-wg-leads:
   - dsimansk
-  - navidshaikh
   - rhuss
   client-writers:
   - dsimansk
-  - navidshaikh
   - rhuss
   discovery-approvers:
   - lberk
@@ -98,18 +99,21 @@ aliases:
   - lance
   - lkingland
   - matejvasek
+  - matzew
   - nainaz
   - salaboy
   function-runtime-approvers:
   - lance
   - lkingland
   - matejvasek
+  - matzew
   - nainaz
   - salaboy
   function-runtime-writers:
   - lance
   - lkingland
   - matejvasek
+  - matzew
   - nainaz
   - salaboy
   homebrew-kn-plugins-approvers:
@@ -118,13 +122,11 @@ aliases:
   - rhuss
   kn-plugin-admin-approvers:
   - maximilien
-  - navidshaikh
   - rhuss
   - zhanggbj
   kn-plugin-diag-approvers:
   - cdlliuy
   - maximilien
-  - navidshaikh
   kn-plugin-event-approvers:
   - cardil
   - rhuss
@@ -137,20 +139,17 @@ aliases:
   - rhuss
   kn-plugin-quickstart-approvers:
   - dsimansk
-  - navidshaikh
   - psschwei
   - rhuss
   kn-plugin-sample-approvers:
   - maximilien
-  - navidshaikh
   - rhuss
   kn-plugin-service-log-approvers:
   - rhuss
   kn-plugin-source-kafka-approvers:
   - daisy-ycguo
   - dsimansk
   - maximilien
-  - navidshaikh
   - rhuss
   kn-plugin-source-kamelet-approvers:
   - christophd
@@ -194,7 +193,6 @@ aliases:
   - knative-prow-robot
   - knative-prow-updater-robot
   - knative-test-reporter-robot
-  net-certmanager-approvers: []
   net-contour-approvers:
   - KauzClay
   - dprotaso
@@ -233,7 +231,7 @@ aliases:
   - davidhadas
   - evankanderson
   serving-approvers:
-  - psschwei
+  - ReToCode
   - skonto
   serving-progressive-rollout-approvers:
   - alexagriffith
@@ -251,8 +249,8 @@ aliases:
   serving-wg-leads:
   - dprotaso
   serving-writers:
+  - ReToCode
   - dprotaso
-  - psschwei
   - skonto
   steering-committee:
   - aliok

control-plane/cmd/post-install/main.go

@@ -16,54 +16,5 @@
 
 package main
 
-import (
-	"context"
-	"flag"
-	"fmt"
-	"log"
-
-	"k8s.io/client-go/kubernetes"
-
-	"knative.dev/pkg/environment"
-	"knative.dev/pkg/logging"
-	"knative.dev/pkg/signals"
-)
-
 func main() {
-	ctx := signals.NewContext()
-
-	config, err := logging.NewConfigFromMap(nil)
-	if err != nil {
-		log.Fatal("Failed to create logging config: ", err)
-	}
-
-	logger, _ := logging.NewLoggerFromConfig(config, "kafka-broker-post-install")
-	defer logger.Sync()
-
-	logging.WithLogger(ctx, logger)
-
-	if err := run(ctx); err != nil {
-		logger.Fatal(err)
-	}
-}
-
-func run(ctx context.Context) error {
-	env := environment.ClientConfig{}
-	env.InitFlags(flag.CommandLine)
-	flag.Parse()
-
-	config, err := env.GetRESTConfig()
-	if err != nil {
-		return fmt.Errorf("failed to get kubeconfig: %w", err)
-	}
-
-	deploymentDeleter := &kafkaDeploymentDeleter{
-		k8s: kubernetes.NewForConfigOrDie(config),
-	}
-
-	if err := deploymentDeleter.DeleteBrokerDeployments(ctx); err != nil {
-		return fmt.Errorf("broker migration failed: %v", err)
-	}
-
-	return nil
 }

control-plane/config/eventing-kafka-broker/100-channel/100-kafka-channel.yaml

@@ -147,6 +147,9 @@ spec:
                         description: Generation of the origin of the subscriber with uid:UID.
                         type: integer
                         format: int64
+                      name:
+                        description: The name of the subscription
+                        type: string
                       replyUri:
                         description: ReplyURI is the endpoint for the reply
                         type: string

control-plane/config/eventing-kafka-broker/200-controller/100-config-logging.yaml

@@ -27,7 +27,12 @@ data:
       <appender name="jsonConsoleAppender" class="ch.qos.logback.core.ConsoleAppender">
         <encoder class="net.logstash.logback.encoder.LogstashEncoder"/>
       </appender>
+      <appender name="async" class="ch.qos.logback.classic.AsyncAppender">
+          <appender-ref ref="jsonConsoleAppender" />
+          <neverBlock>true</neverBlock>
+          <maxFlushTime>1000</maxFlushTime>
+      </appender>
       <root level="INFO">
-        <appender-ref ref="jsonConsoleAppender"/>
+        <appender-ref ref="async"/>
       </root>
     </configuration>

control-plane/config/post-install/200-controller-cluster-role.yaml

@@ -20,20 +20,4 @@ metadata:
   name: knative-kafka-controller-post-install
   labels:
     app.kubernetes.io/version: devel
-rules:
-  # we need to be able to delete old deployments
-  - apiGroups:
-      - "apps"
-    resources:
-      - "deployments"
-    verbs:
-      - "delete"
-  # we need to get statefulsets
-  - apiGroups:
-      - "apps"
-    resources:
-      - "statefulsets"
-    verbs:
-      - "get"
-      - "list"
-
+rules: []

control-plane/pkg/apis/config/features.go

@@ -29,6 +29,7 @@ import (
 	"knative.dev/eventing/pkg/apis/feature"
 	"knative.dev/pkg/configmap"
 	"knative.dev/pkg/logging"
+	"knative.dev/pkg/reconciler"
 )
 
 const (
@@ -217,3 +218,12 @@ func executeTemplateToString(template template.Template, metadata v1.ObjectMeta,
 
 	return result.String(), nil
 }
+
+type Stores []reconciler.ConfigStore
+
+func (css Stores) ToContext(ctx context.Context) context.Context {
+	for _, cs := range css {
+		ctx = cs.ToContext(ctx)
+	}
+	return ctx
+}

control-plane/pkg/reconciler/broker/namespaced_broker.go

@@ -72,6 +72,7 @@ type NamespacedReconciler struct {
 	ServiceAccountLister     corelisters.ServiceAccountLister
 	ServiceLister            corelisters.ServiceLister
 	ClusterRoleBindingLister rbaclisters.ClusterRoleBindingLister
+	DeploymentLister         appslisters.DeploymentLister
 	StatefulSetLister        appslisters.StatefulSetLister
 	BrokerLister             eventinglisters.BrokerLister
 
@@ -316,12 +317,18 @@ func (r *NamespacedReconciler) getManifestFromSystemNamespace(broker *eventing.B
 	}
 	resources = append(resources, additionalConfigMaps...)
 
-	additionalDeployments, err := r.statefulSetsFromSystemNamespace(broker)
+	additionalDeployments, err := r.deploymentsFromSystemNamespace(broker)
 	if err != nil {
 		return mf.Manifest{}, err
 	}
 	resources = append(resources, additionalDeployments...)
 
+	additionalStatefulsets, err := r.statefulSetsFromSystemNamespace(broker)
+	if err != nil {
+		return mf.Manifest{}, err
+	}
+	resources = append(resources, additionalStatefulsets...)
+
 	additionalServiceAccounts, err := r.serviceAccountsFromSystemNamespace(broker)
 	if err != nil {
 		return mf.Manifest{}, err
@@ -368,7 +375,6 @@ func (r *NamespacedReconciler) getManifestFromAdditionalResources(broker *eventi
 
 func (r *NamespacedReconciler) statefulSetsFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
 	deployments := []string{
-		"kafka-broker-receiver",
 		"kafka-broker-dispatcher",
 	}
 	resources := make([]unstructured.Unstructured, 0, len(deployments))
@@ -382,6 +388,21 @@ func (r *NamespacedReconciler) statefulSetsFromSystemNamespace(broker *eventing.
 	return resources, nil
 }
 
+func (r *NamespacedReconciler) deploymentsFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
+	deployments := []string{
+		"kafka-broker-receiver",
+	}
+	resources := make([]unstructured.Unstructured, 0, len(deployments))
+	for _, name := range deployments {
+		resource, err := r.createManifestFromSystemDeployment(broker, name)
+		if err != nil {
+			return nil, err
+		}
+		resources = append(resources, resource)
+	}
+	return resources, nil
+}
+
 func (r *NamespacedReconciler) configMapsFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
 	configMaps := []string{
 		"config-kafka-broker-data-plane",
@@ -472,6 +493,25 @@ func (r *NamespacedReconciler) createManifestFromSystemStatefulSet(broker *event
 	return unstructuredFromObject(cm)
 }
 
+func (r *NamespacedReconciler) createManifestFromSystemDeployment(broker *eventing.Broker, name string) (unstructured.Unstructured, error) {
+	sysDeployment, err := r.DeploymentLister.Deployments(r.SystemNamespace).Get(name)
+	if err != nil {
+		return unstructured.Unstructured{}, fmt.Errorf("failed to get Deployment %s/%s: %w", r.SystemNamespace, name, err)
+	}
+
+	cm := &appsv1.Deployment{
+		TypeMeta: metav1.TypeMeta{Kind: "Deployment", APIVersion: appsv1.SchemeGroupVersion.String()},
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace:   broker.GetNamespace(),
+			Name:        sysDeployment.Name,
+			Labels:      sysDeployment.Labels,
+			Annotations: sysDeployment.Annotations,
+		},
+		Spec: sysDeployment.Spec,
+	}
+	return unstructuredFromObject(cm)
+}
+
 func (r *NamespacedReconciler) serviceAccountsFromSystemNamespace(broker *eventing.Broker) ([]unstructured.Unstructured, error) {
 	serviceAccounts := []string{
 		"knative-kafka-broker-data-plane",