Update the event-type cleanup logic

components/event-publisher-proxy/pkg/handler/http/handler.go

@@ -129,7 +129,16 @@ func (h *Handler) publishLegacyEventsAsCE(writer http.ResponseWriter, request *h
 	statusCode, dispatchTime, respBody := h.send(ctx, event)
 	// Change response as per old error codes
 	h.LegacyTransformer.TransformsCEResponseToLegacyResponse(writer, statusCode, event, string(respBody))
-	h.Logger.Infof("Event dispatched id:[%s] statusCode:[%d] duration:[%s] responseBody:[%s]", event.ID(), statusCode, dispatchTime, respBody)
+
+	h.Logger.WithFields(
+		logrus.Fields{
+			"id":           event.ID(),
+			"source":       event.Source(),
+			"type":         event.Type(),
+			"statusCode":   statusCode,
+			"duration":     dispatchTime,
+			"responseBody": respBody,
+		}).Info("Event dispatched")
 }
 
 func (h *Handler) publishCloudEvents(writer http.ResponseWriter, request *http.Request) {
@@ -162,7 +171,15 @@ func (h *Handler) publishCloudEvents(writer http.ResponseWriter, request *http.R
 	statusCode, dispatchTime, respBody := h.send(ctx, event)
 	h.writeResponse(writer, statusCode, respBody)
 
-	h.Logger.Infof("Event dispatched id:[%s] statusCode:[%d] duration:[%s] responseBody:[%s]", event.ID(), statusCode, dispatchTime, respBody)
+	h.Logger.WithFields(
+		logrus.Fields{
+			"id":           event.ID(),
+			"source":       event.Source(),
+			"type":         event.Type(),
+			"statusCode":   statusCode,
+			"duration":     dispatchTime,
+			"responseBody": respBody,
+		}).Info("Event dispatched")
 }
 
 // writeResponse writes the HTTP response given the status code and response body.

components/event-publisher-proxy/pkg/handler/nats/handler.go

@@ -111,7 +111,16 @@ func (h *Handler) publishLegacyEventsAsCE(writer http.ResponseWriter, request *h
 	statusCode, dispatchTime, respBody := h.send(ctx, event)
 	// Change response as per old error codes
 	h.LegacyTransformer.TransformsCEResponseToLegacyResponse(writer, statusCode, event, string(respBody))
-	h.Logger.Infof("Event dispatched id:[%s] statusCode:[%d] duration:[%s] responseBody:[%s]", event.ID(), statusCode, dispatchTime, respBody)
+
+	h.Logger.WithFields(
+		logrus.Fields{
+			"id":           event.ID(),
+			"source":       event.Source(),
+			"type":         event.Type(),
+			"statusCode":   statusCode,
+			"duration":     dispatchTime,
+			"responseBody": respBody,
+		}).Info("Event dispatched")
 }
 
 func (h *Handler) publishCloudEvents(writer http.ResponseWriter, request *http.Request) {
@@ -144,7 +153,15 @@ func (h *Handler) publishCloudEvents(writer http.ResponseWriter, request *http.R
 	statusCode, dispatchTime, respBody := h.send(ctx, event)
 	h.writeResponse(writer, statusCode, respBody)
 
-	h.Logger.Infof("Event dispatched id:[%s] statusCode:[%d] duration:[%s] responseBody:[%s]", event.ID(), statusCode, dispatchTime, respBody)
+	h.Logger.WithFields(
+		logrus.Fields{
+			"id":           event.ID(),
+			"source":       event.Source(),
+			"type":         event.Type(),
+			"statusCode":   statusCode,
+			"duration":     dispatchTime,
+			"responseBody": respBody,
+		}).Info("Event dispatched")
 }
 
 // writeResponse writes the HTTP response given the status code and response body.

components/event-publisher-proxy/pkg/legacy-events/legacy.go

@@ -2,8 +2,10 @@ package legacy
 
 import (
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"regexp"
+	"strings"
 	"time"
 
 	cev2event "github.com/cloudevents/sdk-go/v2/event"
@@ -168,10 +170,23 @@ func (t Transformer) convertPublishRequestToCloudEvent(appName string, publishRe
 		event.SetID(uuid.New().String())
 	}
 
-	eventType := formatEventType4BEB(t.eventTypePrefix, appName, publishRequest.PublishrequestV1.EventType, publishRequest.PublishrequestV1.EventTypeVersion)
+	eventTypeCombined := combineEventTypeSegments(publishRequest.PublishrequestV1.EventType)
+	eventType := formatEventType4BEB(t.eventTypePrefix, appName, eventTypeCombined, publishRequest.PublishrequestV1.EventTypeVersion)
 	event.SetSource(t.bebNamespace)
 	event.SetType(eventType)
 	event.SetExtension(eventTypeVersionExtensionKey, publishRequest.PublishrequestV1.EventTypeVersion)
 	event.SetDataContentType(ContentTypeApplicationJSON)
 	return &event, nil
 }
+
+// combineEventTypeSegments returns an event-type with exactly two segments separated by "." if the given event-type
+// has more than two segments separated by "." (e.g. "Account.Order.Created" becomes "AccountOrder.Created")
+func combineEventTypeSegments(eventType string) string {
+	parts := strings.Split(eventType, ".")
+	if len(parts) > 2 {
+		businessObject := strings.Join(parts[0:len(parts)-1], "")
+		operation := parts[len(parts)-1]
+		eventType = fmt.Sprintf("%s.%s", businessObject, operation)
+	}
+	return eventType
+}

components/event-publisher-proxy/pkg/legacy-events/legacy_test.go

@@ -8,23 +8,27 @@ import (
 	. "github.com/kyma-project/kyma/components/event-publisher-proxy/testing"
 )
 
+const (
+	eventTypeMultiSegment         = "Segment1.Segment2.Segment3.Segment4.Segment5"
+	eventTypeMultiSegmentCombined = "Segment1Segment2Segment3Segment4.Segment5"
+)
+
 func TestConvertPublishRequestToCloudEvent(t *testing.T) {
 	bebNs := MessagingNamespace
 	eventTypePrefix := MessagingEventTypePrefix
 	legacyTransformer := NewTransformer(bebNs, eventTypePrefix, nil)
 	eventID := EventID
 	appName := ApplicationName
-	legacyEventType := LegacyEventType
 	legacyEventVersion := LegacyEventTypeVersion
 	data := LegacyEventData
 	timeNow := time.Now()
-	expectedEventType := formatEventType4BEB(eventTypePrefix, appName, legacyEventType, legacyEventVersion)
+	expectedEventType := formatEventType4BEB(eventTypePrefix, appName, eventTypeMultiSegmentCombined, legacyEventVersion)
 	timeNowStr := timeNow.Format(time.RFC3339)
 	timeNowFormatted, _ := time.Parse(time.RFC3339, timeNowStr)
 	publishReqParams := &legacyapi.PublishEventParametersV1{
 		PublishrequestV1: legacyapi.PublishRequestV1{
 			EventID:          eventID,
-			EventType:        legacyEventType,
+			EventType:        eventTypeMultiSegment,
 			EventTime:        timeNowStr,
 			EventTypeVersion: legacyEventVersion,
 			Data:             data,
@@ -63,3 +67,30 @@ func TestConvertPublishRequestToCloudEvent(t *testing.T) {
 		t.Errorf("incorrect eventtype extension, want: %s, got: %s", legacyEventVersion, gotExtension)
 	}
 }
+
+func TestCombineEventTypeSegments(t *testing.T) {
+	testCases := []struct {
+		name           string
+		givenEventType string
+		wantEventType  string
+	}{
+		{
+			name:           "event-type with two segments",
+			givenEventType: LegacyEventType,
+			wantEventType:  LegacyEventType,
+		},
+		{
+			name:           "event-type with more than two segments",
+			givenEventType: eventTypeMultiSegment,
+			wantEventType:  eventTypeMultiSegmentCombined,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			if gotEventType := combineEventTypeSegments(tc.givenEventType); tc.wantEventType != gotEventType {
+				t.Fatalf("invalid event-type want: %s, got: %s", tc.wantEventType, gotEventType)
+			}
+		})
+	}
+}

components/eventing-controller/pkg/application/clean.go

@@ -37,8 +37,3 @@ func GetCleanTypeOrName(application *applicationv1alpha1.Application) string {
 func GetCleanName(name string) string {
 	return invalidApplicationNameSegment.ReplaceAllString(name, "")
 }
-
-// IsCleanName returns true if the name contains alphanumeric characters only, otherwise returns false
-func IsCleanName(name string) bool {
-	return !invalidApplicationNameSegment.MatchString(name)
-}

components/eventing-controller/pkg/application/clean_test.go

@@ -54,23 +54,3 @@ func TestCleanName(t *testing.T) {
 		}
 	}
 }
-
-func TestIsCleanName(t *testing.T) {
-	testCases := []struct {
-		givenName string
-		wantClean bool
-	}{
-		{givenName: "with.dot", wantClean: false},
-		{givenName: "with-dash", wantClean: false},
-		{givenName: "with_underscore", wantClean: false},
-		{givenName: "with#special$characters", wantClean: false},
-		{givenName: "alphabetical", wantClean: true},
-		{givenName: "alphanumeric0123", wantClean: true},
-	}
-
-	for _, tc := range testCases {
-		if gotClean := IsCleanName(tc.givenName); tc.wantClean != gotClean {
-			t.Errorf("Is clean application name:[%s] failed, want:[%v] but got:[%v]", tc.givenName, tc.wantClean, gotClean)
-		}
-	}
-}

components/eventing-controller/pkg/handlers/eventtype/clean.go

@@ -1,10 +1,18 @@
 package eventtype
 
 import (
+	"regexp"
+
 	"github.com/go-logr/logr"
 	"github.com/kyma-project/kyma/components/eventing-controller/pkg/application"
 )
 
+var (
+	// invalidEventTypeSegment used to match and replace none-alphanumeric characters in the event-type segments
+	// as per SAP Event spec https://github.tools.sap/CentralEngineering/sap-event-specification#type
+	invalidEventTypeSegment = regexp.MustCompile("[^a-zA-Z0-9.]")
+)
+
 type Cleaner interface {
 	Clean(eventType string) (string, error)
 }
@@ -22,23 +30,31 @@ func NewCleaner(eventTypePrefix string, applicationLister *application.Lister, l
 	return &cleaner{eventTypePrefix: eventTypePrefix, applicationLister: applicationLister, logger: logger}
 }
 
-// Clean cleans the application name segment in the event-type from none-alphanumeric characters
-// and returns the clean event-type, or returns an error if the event-type parsing failed
+// Clean cleans the event-type from none-alphanumeric characters and returns it
+// or returns an error if the event-type parsing failed
 func (c cleaner) Clean(eventType string) (string, error) {
 	appName, event, version, err := parse(eventType, c.eventTypePrefix)
 	if err != nil {
 		c.logger.Error(err, "failed to parse event-type", "prefix", c.eventTypePrefix, "type", eventType)
 		return "", err
 	}
 
-	// handle existing applications
-	if appObj, err := c.applicationLister.Get(appName); err == nil {
-		eventType = build(c.eventTypePrefix, application.GetCleanTypeOrName(appObj), event, version)
-		return eventType, nil
+	// clean the application name
+	var eventTypeClean string
+	if appObj, err := c.applicationLister.Get(appName); err != nil {
+		c.logger.Info("cannot find application", "name", appName)
+		eventTypeClean = build(c.eventTypePrefix, application.GetCleanName(appName), event, version)
+	} else {
+		eventTypeClean = build(c.eventTypePrefix, application.GetCleanTypeOrName(appObj), event, version)
 	}
 
-	// handle non-existing applications
-	c.logger.Info("failed to get application", "name", appName)
-	eventType = build(c.eventTypePrefix, application.GetCleanName(appName), event, version)
-	return eventType, nil
+	// clean the event-type segments
+	eventTypeClean = cleanEventType(eventTypeClean)
+	c.logger.Info("clean event-type", "before", eventType, "after", eventTypeClean)
+
+	return eventTypeClean, nil
+}
+
+func cleanEventType(eventType string) string {
+	return invalidEventTypeSegment.ReplaceAllString(eventType, "")
 }