Channels working

Signed-off-by: Ali Ok <aliok@redhat.com>

Author: aliok

backends/pkg/eventmesh/v1/api.gen.go

@@ -53,7 +53,8 @@ func BuildEventMesh(ctx context.Context, clientset versioned.Interface, dynamicC
 	// map key: "<namespace>/<name>"
 	brokerMap := make(map[string]*Broker)
 	for _, cbr := range convertedBrokers {
-		brokerMap[cbr.GetNamespacedName()] = cbr
+		key := util.GKNamespacedName("eventing.knative.dev", "Broker", cbr.Namespace, cbr.Name)
+		brokerMap[key] = cbr
 	}
 
 	subscribableMap := make(map[string]*Subscribable)
@@ -69,31 +70,35 @@ func BuildEventMesh(ctx context.Context, clientset versioned.Interface, dynamicC
 		return EventMesh{}, err
 	}
 
-	// register the event types in the brokers
+	// register the event types in the brokers and channels
 	for _, et := range convertedEventTypes {
 		if et.Reference != nil {
 			if br, ok := brokerMap[*et.Reference]; ok {
 				br.ProvidedEventTypes = append(br.ProvidedEventTypes, et.NamespacedName())
+			} else if subscribable, ok := subscribableMap[*et.Reference]; ok {
+				subscribable.ProvidedEventTypes = append(subscribable.ProvidedEventTypes, et.NamespacedName())
+			} else {
+				logger.Infow("Event type reference not found", "eventType", et.NamespacedName(), "reference", *et.Reference)
 			}
 		}
 	}
 
-	// fetch the triggers we will process them later
-	triggers, err := clientset.EventingV1().Triggers(metav1.NamespaceAll).List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		logger.Errorw("Error listing triggers", "error", err)
-		return EventMesh{}, err
-	}
-
 	// build a map for easier access to the ETs by their namespaced name.
 	// we need this map when processing the triggers to find out ET definitions for the ET references
-	// brokers provide.
+	// brokers and channels provide.
 	// map key: "<namespace>/<eventType.name>"
 	etByNamespacedName := make(map[string]*EventType)
 	for _, et := range convertedEventTypes {
 		etByNamespacedName[et.NamespacedName()] = et
 	}
 
+	// fetch the triggers we will process them later
+	triggers, err := clientset.EventingV1().Triggers(metav1.NamespaceAll).List(context.Background(), metav1.ListOptions{})
+	if err != nil {
+		logger.Errorw("Error listing triggers", "error", err)
+		return EventMesh{}, err
+	}
+
 	for _, trigger := range triggers.Items {
 		err := processTrigger(ctx, &trigger, brokerMap, etByNamespacedName, dynamicClient, logger)
 		if err != nil {
@@ -110,7 +115,7 @@ func BuildEventMesh(ctx context.Context, clientset versioned.Interface, dynamicC
 	}
 
 	for _, subscription := range subscriptions.Items {
-		err := processSubscription(ctx, &subscription, subscribableMap, dynamicClient, logger)
+		err := processSubscription(ctx, &subscription, subscribableMap, etByNamespacedName, dynamicClient, logger)
 		if err != nil {
 			logger.Errorw("Error processing subscription", "error", err)
 			// do not stop the Backstage plugin from rendering the rest of the data, e.g. because
@@ -166,7 +171,7 @@ func processTrigger(ctx context.Context, trigger *eventingv1.Trigger, brokerMap
 		logger.Errorw("Trigger has no broker", "namespace", trigger.Namespace, "trigger", trigger.Name)
 		return nil
 	}
-	brokerRef := util.NamespacedName(trigger.Namespace, trigger.Spec.Broker)
+	brokerRef := util.GKNamespacedName("eventing.knative.dev", "Broker", trigger.Namespace, trigger.Spec.Broker)
 	if _, ok := brokerMap[brokerRef]; !ok {
 		logger.Infow("Broker not found", "namespace", trigger.Namespace, "trigger", trigger.Name, "broker", trigger.Spec.Broker)
 		return nil
@@ -182,7 +187,7 @@ func processTrigger(ctx context.Context, trigger *eventingv1.Trigger, brokerMap
 	return nil
 }
 
-func processSubscription(ctx context.Context, subscription *v1.Subscription, subscribableMap map[string]*Subscribable, dynamicClient dynamic.Interface, logger *zap.SugaredLogger) error {
+func processSubscription(ctx context.Context, subscription *v1.Subscription, subscribableMap map[string]*Subscribable, etByNamespacedName map[string]*EventType, dynamicClient dynamic.Interface, logger *zap.SugaredLogger) error {
 	// if the subscription has no subscriber, we can skip it, there's no relation to show on Backstage side
 	if subscription.Spec.Subscriber.Ref == nil {
 		logger.Debugw("Subscription has no subscriber ref; cannot process this subscription", "namespace", subscription.Namespace, "subscription", subscription.Name)
@@ -209,6 +214,16 @@ func processSubscription(ctx context.Context, subscription *v1.Subscription, sub
 		return nil
 	}
 
+	eventTypes := subscribableMap[channelRef].ProvidedEventTypes
+	logger.Infow("Collected provided event types", "namespace", subscription.Namespace, "subscription", subscription.Name, "channel", channel.Name, "eventTypes", eventTypes)
+
+	for _, eventType := range eventTypes {
+		key := util.NamespacedName(subscription.Namespace, eventType)
+		if et, ok := etByNamespacedName[key]; ok {
+			et.ConsumedBy = append(et.ConsumedBy, subscriberBackstageId)
+		}
+	}
+
 	return nil
 }
 

backends/pkg/eventmesh/v1/builder.go

@@ -1,10 +1,8 @@
 package v1
 
 import (
-	"knative.dev/eventing/pkg/apis/eventing/v1beta2"
-	"knative.dev/eventing/pkg/apis/eventing/v1beta3"
-
 	"knative.dev/backstage-plugins/backends/pkg/util"
+	"knative.dev/eventing/pkg/apis/eventing/v1beta2"
 )
 
 // NamespacedName returns the name and namespace of the event type in the format "<namespace>/<name>"
@@ -21,6 +19,10 @@ func (et EventType) NamespacedType() string {
 // convertEventType converts a Knative Eventing EventType to a simplified representation that is easier to consume by the Backstage plugin.
 // see EventType.
 func convertEventType(et *v1beta2.EventType) EventType {
+	reference := ""
+	if et.Spec.Reference != nil {
+		reference = util.GKNamespacedName(util.APIVersionToGroup(et.Spec.Reference.APIVersion), et.Spec.Reference.Kind, et.Namespace, et.Spec.Reference.Name)
+	}
 	return EventType{
 		Name:        et.Name,
 		Namespace:   et.Namespace,
@@ -31,39 +33,8 @@ func convertEventType(et *v1beta2.EventType) EventType {
 		SchemaURL:   util.ToStrPtrOrNil(et.Spec.Schema.String()),
 		Labels:      et.Labels,
 		Annotations: util.FilterAnnotations(et.Annotations),
-		Reference:   util.ToStrPtrOrNil(util.NamespacedRefName(et.Spec.Reference)),
-		// this field will be populated later on, when we have process the triggers
-		ConsumedBy: make([]string, 0),
-	}
-}
-
-// convertEventType converts a Knative Eventing EventType to a simplified representation that is easier to consume by the Backstage plugin.
-// see EventType.
-func convertEventTypev1beta3(et *v1beta3.EventType) EventType {
-	cet := EventType{
-		Name:        et.Name,
-		Namespace:   et.Namespace,
-		Uid:         string(et.UID),
-		Description: util.ToStrPtrOrNil(et.Spec.Description),
-		Labels:      et.Labels,
-		Annotations: util.FilterAnnotations(et.Annotations),
-		Reference:   util.ToStrPtrOrNil(util.NamespacedRefName(et.Spec.Reference)),
+		Reference:   util.ToStrPtrOrNil(reference),
 		// this field will be populated later on, when we have process the triggers
 		ConsumedBy: make([]string, 0),
 	}
-
-	if len(et.Spec.Attributes) == 0 {
-		return cet
-	}
-
-	for _, attr := range et.Spec.Attributes {
-		switch attr.Name {
-		case "type": // TODO: any CE constant for these?
-			cet.Type = attr.Value
-		case "schemadata":
-			cet.SchemaURL = util.ToStrPtrOrNil(attr.Value)
-		}
-	}
-
-	return cet
 }

backends/pkg/eventmesh/v1/eventtype.go

@@ -15,5 +15,7 @@ func convertSubscribable(gvr schema.GroupVersionResource, u *unstructured.Unstru
 		Labels:      u.GetLabels(),
 		Group:       gvr.Group,
 		Kind:        u.GetKind(),
+		// this field will be populated later on
+		ProvidedEventTypes: []string{},
 	}
 }

backends/pkg/eventmesh/v1/server.gen.go

@@ -52,6 +52,7 @@ export type Subscribable = {
     annotations?:Record<string, string>;
     group:string;
     kind:string;
+    providedEventTypes?:string[];
 };
 
 type EventMesh = {
@@ -306,6 +307,7 @@ export class KnativeEventMeshProvider implements EntityProvider {
                 lifecycle: this.env,
                 system: SystemKnative,
                 owner: OwnerKnative,
+                providesApis: !subscribable.providedEventTypes ? [] : subscribable.providedEventTypes.map((eventType:string) => `api:${eventType}`),
             }
         }
     }

backends/pkg/eventmesh/v1/subscribable.go

@@ -128,6 +128,13 @@ components:
             description: Kubernetes API kind of the subscribable.
             format: string
             example: InMemoryChannel
+          providedEventTypes:
+            type: array
+            items:
+              type: string
+              format: string
+            description: List of event types provided by the subscribable.
+            example: [ "my-namespace/my-event-type" ]
         required:
           - namespace
           - name
@@ -136,6 +143,7 @@ components:
           - annotations
           - group
           - kind
+          - providedEventTypes
     Source:
       type: object
       description: Source is a simplified representation of a Knative Eventing Source that is easier to consume by the Backstage plugin.
@@ -251,10 +259,7 @@ components:
           description: Annotations of the event type. These are passed as is, except that are filtered out by the `FilterAnnotations` function.
           example: { "key": "value" }
         reference:
-          type: string
-          format: string
-          description: Reference is the EventTypes's reference to a resource like a broker or a channel. It is in the format `<namespace>/<name>`.
-          example: my-namespace/my-broker
+          $ref: '#/components/schemas/GroupKindNamespacedName'
         consumedBy:
           type: array
           items:
@@ -271,6 +276,35 @@ components:
         - labels
         - annotations
         - consumedBy
+    GroupKindNamespacedName:
+        type: object
+        description: GroupKindNamespacedName is a struct that holds the group, kind, namespace, and name of a Kubernetes resource.
+        properties:
+          group:
+            type: string
+            description: Kubernetes API group of the resource, without the version.
+            format: string
+            example: eventing.knative.dev
+          kind:
+            type: string
+            description: Kubernetes API kind of the resource.
+            format: string
+            example: Broker
+          namespace:
+            type: string
+            description: Namespace of the resource.
+            format: string
+            example: my-namespace
+          name:
+            type: string
+            description: Name of the resource.
+            format: string
+            example: my-broker
+        required:
+            - group
+            - kind
+            - namespace
+            - name
     EventMesh:
       type: object
       description: EventMesh is the top-level struct that holds the event mesh data. It's the struct that's serialized and sent to the Backstage plugin.