Make EventType Autocreate Async (#7709)

* feat: eventtype autocreate is now async Signed-off-by: Calum Murray <cmurray@redhat.com> * cleanup: removed unnecessary return statement Signed-off-by: Calum Murray <cmurray@redhat.com> --------- Signed-off-by: Calum Murray <cmurray@redhat.com>
knative · Mar 5, 2024 · 6e7728e · 6e7728e
1 parent 2645b50
commit 6e7728e
Show file tree

Hide file tree

Showing 4 changed files with 57 additions and 60 deletions.
diff --git a/pkg/broker/ingress/ingress_handler.go b/pkg/broker/ingress/ingress_handler.go
@@ -278,9 +278,7 @@ func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
 
 	// EventType auto-create feature handling
 	if h.EvenTypeHandler != nil {
-		if err := h.EvenTypeHandler.AutoCreateEventType(ctx, event, toKReference(broker), broker.GetUID()); err != nil {
-			h.Logger.Error("Even type auto create failed", zap.Error(err))
-		}
+		h.EvenTypeHandler.AutoCreateEventType(ctx, event, toKReference(broker), broker.GetUID())
 	}
 }
 

diff --git a/pkg/channel/fanout/fanout_event_handler.go b/pkg/channel/fanout/fanout_event_handler.go
@@ -208,11 +208,7 @@ func (f *FanoutEventHandler) autoCreateEventType(ctx context.Context, evnt event
 			f.logger.Warn("No channelUID provided, unable to autocreate event type")
 			return
 		}
-		err := f.eventTypeHandler.AutoCreateEventType(ctx, &evnt, f.channelRef, *f.channelUID)
-		if err != nil {
-			f.logger.Warn("EventTypeCreate failed")
-			return
-		}
+		f.eventTypeHandler.AutoCreateEventType(ctx, &evnt, f.channelRef, *f.channelUID)
 	}
 }
 

diff --git a/pkg/eventtype/eventtypes.go b/pkg/eventtype/eventtypes.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"crypto/md5" //nolint:gosec
 	"fmt"
+	"time"
 
 	"github.com/cloudevents/sdk-go/v2/event"
 	"go.uber.org/zap"
@@ -52,55 +53,62 @@ func generateEventTypeName(name, namespace, eventType, eventSource string) strin
 }
 
 // AutoCreateEventType creates EventType object based on processed event's types from addressable KReference objects
-func (h *EventTypeAutoHandler) AutoCreateEventType(ctx context.Context, event *event.Event, addressable *duckv1.KReference, ownerUID types.UID) error {
+func (h *EventTypeAutoHandler) AutoCreateEventType(ctx context.Context, event *event.Event, addressable *duckv1.KReference, ownerUID types.UID) {
 	// Feature flag gate
 	if !h.FeatureStore.IsEnabled(feature.EvenTypeAutoCreate) {
 		h.Logger.Debug("Event Type auto creation is disabled")
-		return nil
+		return
 	}
-	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,
+	ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), time.Second*30)
+	go func() {
+		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))
+			cancel()
+			return
+		}
+		if exists != nil {
+			cancel()
+			return
+		}
+
+		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
+			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))
+			cancel()
+			return
+		}
+		cancel()
+	}()
 }
diff --git a/pkg/eventtype/eventtypes_test.go b/pkg/eventtype/eventtypes_test.go
@@ -20,6 +20,7 @@ import (
 	"context"
 	"reflect"
 	"testing"
+	"time"
 
 	v2 "github.com/cloudevents/sdk-go/v2"
 	"github.com/cloudevents/sdk-go/v2/event"
@@ -94,14 +95,8 @@ func TestEventTypeAutoHandler_AutoCreateEventType(t *testing.T) {
 
 		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)
-				}
-			}
+			handler.AutoCreateEventType(ctx, &event, tc.addressable, ownerUID)
+			time.Sleep(time.Millisecond * 500) // autocreate runs in a different goroutine, need to wait for it to finish
 
 			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{})