Add handler to auto create Event Types (#7034)

Fixes #6909 

<!-- Please include the 'why' behind your changes if no issue exists -->

I still need to add unit tests, but I'd like to get any feedback wrt/
code structure and overall approach. Thx.
/cc @matzew @pierDipi 

## Proposed Changes

<!-- Please categorize your changes:
- 🎁 Add new feature
- 🐛 Fix bug
- 🧹 Update or clean up current behavior
- 🗑️ Remove feature or internal logic
-->

- Add handler to auto create Event Types
- Add feature flag to gate it

### Pre-review Checklist

<!-- If these boxes are not checked, you will be asked to complete these
requirements or explain why they do not apply to your PR. -->

- [ ] **At least 80% unit test coverage**
- [ ] **E2E tests** for any new behavior
- [ ] **Docs PR** for any user-facing impact
- [ ] **Spec PR** for any new API feature
- [ ] **Conformance test** for any change to the spec

**Release Note**

<!--
📄 If this change has user-visible impact, write a release
note in the block
below. Include the string "action required" if additional action is
required of
users switching to the new release, for example in case of a breaking
change.

Write as if you are speaking to users, not other Knative contributors.
If this
change has no user-visible impact, no release note is needed.
-->

```release-note
Even Type auto-create feature:
- Feature flag to enable: `eventtype-auto-create` in `configmap/config-features`
- Based on CloudEvents processed in a broker corresponding `EventType` resources are created in the namespace  

```


**Docs**

The doc issue to track: knative/docs#5612

<!--
:book: If this change has user-visible impact, link to an issue or PR in
https://github.com/knative/docs.
-->

---------

Co-authored-by: Matthias Wessendorf <mwessend@redhat.com>
Co-authored-by: Pierangelo Di Pilato <pierangelodipilato@gmail.com>

cmd/broker/ingress/main.go

@@ -41,9 +41,12 @@ import (
 	tracingconfig "knative.dev/pkg/tracing/config"
 
 	cmdbroker "knative.dev/eventing/cmd/broker"
+	"knative.dev/eventing/pkg/apis/feature"
 	broker "knative.dev/eventing/pkg/broker"
 	"knative.dev/eventing/pkg/broker/ingress"
+	eventingclient "knative.dev/eventing/pkg/client/injection/client"
 	brokerinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1/broker"
+	eventtypeinformer "knative.dev/eventing/pkg/client/injection/informers/eventing/v1beta2/eventtype"
 	"knative.dev/eventing/pkg/kncloudevents"
 	"knative.dev/eventing/pkg/reconciler/names"
 )
@@ -136,6 +139,9 @@ func main() {
 		logger.Fatal("Unable to create message sender", zap.Error(err))
 	}
 
+	featureStore := feature.NewStore(logging.FromContext(ctx).Named("feature-config-store"))
+	featureStore.WatchConfigs(configMapWatcher)
+
 	reporter := ingress.NewStatsReporter(env.ContainerName, kmeta.ChildName(env.PodName, uuid.New().String()))
 
 	h := &ingress.Handler{
@@ -152,6 +158,17 @@ func main() {
 		logger.Warn("Failed to start ConfigMap watcher", zap.Error(err))
 	}
 
+	// Init auto-create only if enabled, after ConfigMap watcher is started
+	if featureStore.IsEnabled(feature.EvenTypeAutoCreate) {
+		autoCreate := &broker.EventTypeAutoHandler{
+			EventTypeLister: eventtypeinformer.Get(ctx).Lister(),
+			EventingClient:  eventingclient.Get(ctx).EventingV1beta2(),
+			FeatureStore:    featureStore,
+			Logger:          logger,
+		}
+		h.EvenTypeHandler = autoCreate
+	}
+
 	// Start all of the informers and wait for them to sync.
 	logger.Info("Starting informers.")
 	if err := controller.StartInformers(ctx.Done(), informers...); err != nil {

config/brokers/mt-channel-broker/roles/ingress-clusterrole.yaml

@@ -19,6 +19,15 @@ metadata:
     app.kubernetes.io/version: devel
     app.kubernetes.io/name: knative-eventing
 rules:
+  - apiGroups:
+      - eventing.knative.dev
+    resources:
+      - eventtypes
+    verbs:
+      - create
+      - get
+      - list
+      - watch
   - apiGroups:
       - eventing.knative.dev
     resources:

config/core/configmaps/features.yaml

@@ -47,3 +47,7 @@ data:
   # ALPHA feature: The transport-encryption flag allows you to encrypt events in transit using the transport layer security (TLS) protocol.
   # For more details: https://github.com/knative/eventing/issues/5957
   transport-encryption: "disabled"
+
+  # ALPHA feature: The eventtype-auto-create flag allows automatic creation of Even Type instances based on Event's type being processed.
+  # For more details: https://github.com/knative/eventing/issues/6909
+  eventtype-auto-create: "disabled"

pkg/apis/feature/flag_names.go

@@ -23,4 +23,5 @@ const (
 	KReferenceMapping   = "kreference-mapping"
 	NewTriggerFilters   = "new-trigger-filters"
 	TransportEncryption = "transport-encryption"
+	EvenTypeAutoCreate  = "eventtype-auto-create"
 )

pkg/broker/eventtypes.go

@@ -0,0 +1,106 @@
+/*
+Copyright 2023 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 broker
+
+import (
+	"context"
+	"encoding/base64"
+	"fmt"
+
+	"github.com/cloudevents/sdk-go/v2/event"
+	"go.uber.org/zap"
+
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	"knative.dev/eventing/pkg/apis/eventing/v1beta2"
+	"knative.dev/eventing/pkg/apis/feature"
+	eventingv1beta2 "knative.dev/eventing/pkg/client/clientset/versioned/typed/eventing/v1beta2"
+	v1beta22 "knative.dev/eventing/pkg/client/listers/eventing/v1beta2"
+	"knative.dev/eventing/pkg/utils"
+	"knative.dev/pkg/apis"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+)
+
+type EventTypeAutoHandler struct {
+	EventTypeLister v1beta22.EventTypeLister
+	EventingClient  eventingv1beta2.EventingV1beta2Interface
+	FeatureStore    *feature.Store
+	Logger          *zap.Logger
+}
+
+// generateEventTypeName is a pseudo unique name for EvenType object based on on the input params
+func generateEventTypeName(name, namespace, eventType, eventSource string) string {
+	suffixParts := name + namespace + eventType + eventSource
+	suffix := base64.StdEncoding.EncodeToString([]byte(suffixParts))[:10]
+	return utils.ToDNS1123Subdomain(fmt.Sprintf("%s-%s-%s", "et", name, suffix))
+}
+
+// AutoCreateEventType creates EventType object based on processed events's types from addressable KReference objects
+func (h *EventTypeAutoHandler) AutoCreateEventType(ctx context.Context, event *event.Event, addressable *duckv1.KReference, ownerUID types.UID) error {
+	// Feature flag gate
+	if !h.FeatureStore.IsEnabled(feature.EvenTypeAutoCreate) {
+		h.Logger.Debug("Event Type auto creation is disabled")
+		return nil
+	}
+	h.Logger.Debug("Event Types auto creation is enabled")
+
+	eventTypeName := generateEventTypeName(addressable.Name, addressable.Namespace, event.Type(), event.Source())
+
+	exists, err := h.EventTypeLister.EventTypes(addressable.Namespace).Get(eventTypeName)
+	if err != nil && !apierrs.IsNotFound(err) {
+		h.Logger.Error("Failed to retrieve Even Type", zap.Error(err))
+		return err
+	}
+	if exists != nil {
+		return nil
+	}
+
+	source, _ := apis.ParseURL(event.Source())
+	schema, _ := apis.ParseURL(event.DataSchema())
+
+	et := &v1beta2.EventType{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      eventTypeName,
+			Namespace: addressable.Namespace,
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion: addressable.APIVersion,
+					Kind:       addressable.Kind,
+					Name:       addressable.Name,
+					UID:        ownerUID,
+				},
+			},
+		},
+		Spec: v1beta2.EventTypeSpec{
+			Type:        event.Type(),
+			Source:      source,
+			Schema:      schema,
+			SchemaData:  event.DataSchema(),
+			Reference:   addressable,
+			Description: "Event Type auto-created by controller",
+		},
+	}
+
+	_, err = h.EventingClient.EventTypes(et.Namespace).Create(ctx, et, metav1.CreateOptions{})
+	if err != nil && !apierrs.IsAlreadyExists(err) {
+		h.Logger.Error("Failed to create Event Type", zap.Error(err))
+		return err
+	}
+	return nil
+}

pkg/broker/eventtypes_test.go

@@ -0,0 +1,192 @@
+/*
+Copyright 2023 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 broker
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	v2 "github.com/cloudevents/sdk-go/v2"
+	"github.com/cloudevents/sdk-go/v2/event"
+	"go.uber.org/zap"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
+	"knative.dev/eventing/pkg/apis/eventing/v1beta2"
+	"knative.dev/eventing/pkg/apis/feature"
+	fakeeventingclientset "knative.dev/eventing/pkg/client/clientset/versioned/fake"
+	reconcilertestingv1beta2 "knative.dev/eventing/pkg/reconciler/testing/v1beta2"
+	"knative.dev/eventing/test/lib/resources"
+	duckv1 "knative.dev/pkg/apis/duck/v1"
+	logtesting "knative.dev/pkg/logging/testing"
+)
+
+func TestEventTypeAutoHandler_AutoCreateEventType(t *testing.T) {
+	testCases := []struct {
+		name              string
+		featureFlag       string
+		addressable       *duckv1.KReference
+		events            []v2.Event
+		expectedEventType []v1beta2.EventType
+		expectedError     error
+	}{
+		{
+			name:        "With 1 broker and 1 type",
+			featureFlag: "enabled",
+			addressable: &duckv1.KReference{
+				APIVersion: "eventing.knative.dev/v1",
+				Kind:       "Broker",
+				Namespace:  "default",
+				Name:       "broker"},
+			events: []v2.Event{initEvent("")},
+			expectedEventType: []v1beta2.EventType{
+				initEventTypeObject(),
+				initEventTypeObject()},
+			expectedError: nil,
+		},
+		{
+			name:        "With 1 broker and multiple types",
+			featureFlag: "enabled",
+			addressable: &duckv1.KReference{
+				APIVersion: "eventing.knative.dev/v1",
+				Kind:       "Broker",
+				Namespace:  "default",
+				Name:       "broker"},
+			events: []v2.Event{
+				initEvent("foo.type"),
+				initEvent("bar.type")},
+			expectedEventType: []v1beta2.EventType{
+				initEventTypeObject(),
+				initEventTypeObject()},
+			expectedError: nil,
+		},
+	}
+	for _, tc := range testCases {
+		ctx := context.TODO()
+		eventtypes := make([]runtime.Object, 0, 10)
+		listers := reconcilertestingv1beta2.NewListers(eventtypes)
+		eventingClient := fakeeventingclientset.NewSimpleClientset()
+		logger := zap.NewNop()
+
+		handler := &EventTypeAutoHandler{
+			EventTypeLister: listers.GetEventTypeLister(),
+			EventingClient:  eventingClient.EventingV1beta2(),
+			FeatureStore:    initFeatureStore(t, tc.featureFlag),
+			Logger:          logger,
+		}
+
+		ownerUID := types.UID("owner-uid")
+
+		for i, event := range tc.events {
+
+			err := handler.AutoCreateEventType(ctx, &event, tc.addressable, ownerUID)
+			if err != nil {
+				if tc.expectedError == err {
+					t.Errorf("test case '%s', expected '%s', got '%s'", tc.name, tc.expectedError, err)
+				} else {
+					t.Error(err)
+				}
+			}
+
+			etName := generateEventTypeName(tc.addressable.Name, tc.addressable.Namespace, event.Type(), event.Source())
+			et, err := eventingClient.EventingV1beta2().EventTypes(tc.addressable.Namespace).Get(ctx, etName, metav1.GetOptions{})
+			if err != nil {
+				t.Error(err)
+			}
+			if !reflect.DeepEqual(et.Spec.Reference, tc.expectedEventType[i].Spec.Reference) {
+				t.Errorf("test case '%s', expected '%s', got '%s'", tc.name, tc.expectedEventType[i].Spec.Reference, et.Spec.Reference)
+			}
+		}
+	}
+
+}
+
+func TestEventTypeAutoHandler_GenerateEventTypeName(t *testing.T) {
+	testCases := []struct {
+		name         string
+		namespace    string
+		eventType    string
+		eventSource  string
+		expectedName string
+	}{
+		{
+			name:         "example",
+			namespace:    "default",
+			eventType:    "events.type",
+			eventSource:  "events.source",
+			expectedName: "et-example-zxhhbxbszw",
+		},
+		{
+			name:         "EXAMPLE",
+			namespace:    "default",
+			eventType:    "events.type",
+			eventSource:  "events.source",
+			expectedName: "et-example-rvhbtvbmrw",
+		},
+		{
+			name:         "emptyName",
+			namespace:    "default",
+			eventType:    "events.type",
+			eventSource:  "events.source",
+			expectedName: "et-emptyname-zw1wdhloyw",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			result := generateEventTypeName(tc.name, tc.namespace, tc.eventType, tc.eventSource)
+
+			if result != tc.expectedName {
+				t.Errorf("test case '%s', expected '%s', got '%s'", tc.name, tc.expectedName, result)
+			}
+		})
+	}
+}
+
+func initFeatureStore(t *testing.T, enabled string) *feature.Store {
+	featureStore := feature.NewStore(logtesting.TestLogger(t))
+	cm := resources.ConfigMap(
+		"config-features",
+		"default",
+		map[string]string{feature.EvenTypeAutoCreate: enabled},
+	)
+	featureStore.OnConfigChanged(cm)
+	return featureStore
+}
+
+func initEvent(eventType string) v2.Event {
+	e := event.New()
+	e.SetType(eventType)
+	if eventType == "" {
+		e.SetType("test.Type")
+	}
+	e.SetSource("test.source")
+	return e
+}
+
+func initEventTypeObject() v1beta2.EventType {
+	return v1beta2.EventType{
+		Spec: v1beta2.EventTypeSpec{
+			Reference: &duckv1.KReference{
+				APIVersion: "eventing.knative.dev/v1",
+				Kind:       "Broker",
+				Namespace:  "default",
+				Name:       "broker"},
+		},
+	}
+}