Sources working - finally

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

Author: aliok

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

@@ -0,0 +1,7 @@
+package v1
+
+import "knative.dev/backstage-plugins/backends/pkg/util"
+
+func (gknn GroupKindNamespacedName) String() string {
+	return util.GKNamespacedName(gknn.Group, gknn.Kind, gknn.Namespace, gknn.Name)
+}

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

@@ -48,6 +48,12 @@ func BuildEventMesh(ctx context.Context, clientset versioned.Interface, dynamicC
 		return EventMesh{}, err
 	}
 
+	convertedSourceEntries, err := fetchSources(ctx, dynamicClient, logger)
+	if err != nil {
+		logger.Errorw("Error fetching and converting sources", "error", err)
+		return EventMesh{}, err
+	}
+
 	// build a broker map and a subscribable map for easier access.
 	// we need this map to register the event types in the brokers when we are processing the event types.
 	// map key: "<namespace>/<name>"
@@ -73,9 +79,9 @@ func BuildEventMesh(ctx context.Context, clientset versioned.Interface, dynamicC
 	// register the event types in the brokers and channels
 	for _, et := range convertedEventTypes {
 		if et.Reference != nil {
-			if br, ok := brokerMap[*et.Reference]; ok {
+			if br, ok := brokerMap[et.Reference.String()]; ok {
 				br.ProvidedEventTypes = append(br.ProvidedEventTypes, et.NamespacedName())
-			} else if subscribable, ok := subscribableMap[*et.Reference]; ok {
+			} else if subscribable, ok := subscribableMap[et.Reference.String()]; ok {
 				subscribable.ProvidedEventTypes = append(subscribable.ProvidedEventTypes, et.NamespacedName())
 			} else {
 				logger.Infow("Event type reference not found", "eventType", et.NamespacedName(), "reference", *et.Reference)
@@ -92,6 +98,29 @@ func BuildEventMesh(ctx context.Context, clientset versioned.Interface, dynamicC
 		etByNamespacedName[et.NamespacedName()] = et
 	}
 
+	// build a map for easier access to the ETs by their type.
+	// there can be multiple ETs with the same type but different names.
+	// we need this map when processing the sources to find out ET definitions for the ET types.
+	// map key does not have a namespace: "<eventType.type>"
+	etsByType := make(map[string][]*EventType)
+	for _, et := range convertedEventTypes {
+		etsByType[et.Type] = append(etsByType[et.Type], et)
+	}
+
+	// register the event types in the sources
+	for _, source := range convertedSourceEntries {
+		if source.ProvidedEventTypeTypes == nil {
+			continue
+		}
+		for _, providedType := range *source.ProvidedEventTypeTypes {
+			if ets, ok := etsByType[providedType]; ok {
+				for _, et := range ets {
+					source.ProvidedEventTypes = append(source.ProvidedEventTypes, et.NamespacedName())
+				}
+			}
+		}
+	}
+
 	// fetch the triggers we will process them later
 	triggers, err := clientset.EventingV1().Triggers(metav1.NamespaceAll).List(context.Background(), metav1.ListOptions{})
 	if err != nil {
@@ -135,11 +164,16 @@ func BuildEventMesh(ctx context.Context, clientset versioned.Interface, dynamicC
 	for _, s := range convertedSubscribables {
 		outputSubscribables = append(outputSubscribables, *s)
 	}
+	outputSources := make([]Source, 0, len(convertedSourceEntries))
+	for _, s := range convertedSourceEntries {
+		outputSources = append(outputSources, *s)
+	}
 
 	eventMesh := EventMesh{
 		EventTypes:    outputEventTypes,
 		Brokers:       outputBrokers,
 		Subscribables: outputSubscribables,
+		Sources:       outputSources,
 	}
 
 	return eventMesh, nil
@@ -341,6 +375,58 @@ func fetchSubscribables(ctx context.Context, dynamicClient dynamic.Interface, lo
 	return subscribables, nil
 }
 
+func fetchSources(ctx context.Context, dynamicClient dynamic.Interface, logger *zap.SugaredLogger) ([]*Source, error) {
+	// first, fetch the source CRDs
+	sourceCRDs, err := dynamicClient.Resource(
+		schema.GroupVersionResource{
+			Group:    "apiextensions.k8s.io",
+			Version:  "v1",
+			Resource: "customresourcedefinitions",
+		},
+	).List(ctx, metav1.ListOptions{LabelSelector: labels.Set{"duck.knative.dev/source": "true"}.String()})
+
+	if errors.IsNotFound(err) {
+		return nil, nil
+	}
+
+	if err != nil {
+		logger.Errorw("Error listing source CRDs", "error", err)
+		return nil, err
+	}
+
+	// then, fetch the sources
+	sources := make([]*Source, 0)
+	for _, crd := range sourceCRDs.Items {
+		gvr, err := util.GVRFromUnstructured(&crd)
+		if err != nil {
+			logger.Errorw("Error getting GVR from CRD", "error", err)
+			return nil, err
+		}
+
+		sourceResources, err := dynamicClient.Resource(gvr).Namespace(metav1.NamespaceAll).List(ctx, metav1.ListOptions{})
+
+		if errors.IsNotFound(err) {
+			continue
+		}
+
+		if err != nil {
+			logger.Errorw("Error listing source resources", "error", err)
+			return nil, err
+		}
+
+		for _, resource := range sourceResources.Items {
+			sourceEntry, err := convertSource(gvr, crd, &resource)
+			if err != nil {
+				logger.Errorw("Error converting source", "error", err)
+				return nil, err
+			}
+			sources = append(sources, &sourceEntry)
+		}
+	}
+
+	return sources, nil
+}
+
 // fetchEventTypes fetches the event types and converts them to the representation that's consumed by the Backstage plugin.
 func fetchEventTypes(clientset versioned.Interface, logger *zap.SugaredLogger) ([]*EventType, error) {
 	eventTypeResponse, err := clientset.EventingV1beta2().EventTypes(metav1.NamespaceAll).List(context.Background(), metav1.ListOptions{})

backends/pkg/eventmesh/v1/builder.go

@@ -63,5 +63,6 @@ func (e Endpoint) GetEventMesh(ctx context.Context, _ GetEventMeshRequestObject)
 		Brokers:       eventMesh.Brokers,
 		EventTypes:    eventMesh.EventTypes,
 		Subscribables: eventMesh.Subscribables,
+		Sources:       eventMesh.Sources,
 	}, nil
 }

backends/pkg/eventmesh/v1/endpoint.go

@@ -15,13 +15,17 @@ func (et EventType) NamespacedType() string {
 	return util.NamespacedName(et.Namespace, et.Type)
 }
 
-// TODO: remove
 // 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 := ""
+	var reference *GroupKindNamespacedName
 	if et.Spec.Reference != nil {
-		reference = util.GKNamespacedName(util.APIVersionToGroup(et.Spec.Reference.APIVersion), et.Spec.Reference.Kind, et.Namespace, et.Spec.Reference.Name)
+		reference = &GroupKindNamespacedName{
+			Group:     util.APIVersionToGroup(et.Spec.Reference.APIVersion),
+			Kind:      et.Spec.Reference.Kind,
+			Namespace: et.Namespace,
+			Name:      et.Spec.Reference.Name,
+		}
 	}
 	return EventType{
 		Name:        et.Name,
@@ -33,7 +37,7 @@ func convertEventType(et *v1beta2.EventType) EventType {
 		SchemaURL:   util.ToStrPtrOrNil(et.Spec.Schema.String()),
 		Labels:      et.Labels,
 		Annotations: util.FilterAnnotations(et.Annotations),
-		Reference:   util.ToStrPtrOrNil(reference),
+		Reference:   reference,
 		// this field will be populated later on, when we have process the triggers
 		ConsumedBy: make([]string, 0),
 	}

backends/pkg/eventmesh/v1/eventtype.go

@@ -0,0 +1,89 @@
+package v1
+
+import (
+	"encoding/json"
+	"errors"
+
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+
+	"knative.dev/eventing/pkg/apis/eventing"
+
+	"knative.dev/backstage-plugins/backends/pkg/util"
+)
+
+// eventTypeEntry refers to an entry in the registry.knative.dev/eventTypes annotation.
+type eventTypeEntry struct {
+	Type        string `json:"type"`
+	Schema      string `json:"schema,omitempty"`
+	Description string `json:"description,omitempty"`
+}
+
+func convertSource(gvr schema.GroupVersionResource, crd unstructured.Unstructured, source *unstructured.Unstructured) (Source, error) {
+	providedEventTypeTypes := []string{}
+
+	crdAnnotations := crd.GetAnnotations()
+	if eventTypesJson, ok := crdAnnotations[eventing.EventTypesAnnotationKey]; ok {
+		var providedEventTypeEntries []eventTypeEntry
+		if err := json.Unmarshal([]byte(eventTypesJson), &providedEventTypeEntries); err != nil {
+			return Source{}, errors.New("failed to unmarshal event types")
+		}
+
+		providedEventTypeTypes = make([]string, len(providedEventTypeEntries))
+		for i, entry := range providedEventTypeEntries {
+			providedEventTypeTypes[i] = entry.Type
+		}
+	}
+
+	src := Source{
+		Namespace:              source.GetNamespace(),
+		Name:                   source.GetName(),
+		UID:                    string(source.GetUID()),
+		Annotations:            util.FilterAnnotations(source.GetAnnotations()),
+		Labels:                 source.GetLabels(),
+		ProvidedEventTypeTypes: &providedEventTypeTypes,
+		// this field will be populated later on
+		ProvidedEventTypes: []string{},
+		Group:              gvr.Group,
+		Kind:               source.GetKind(),
+	}
+
+	if sinkRef, ok := getSinkRef(source); ok {
+		src.Sink = &sinkRef
+	}
+
+	return src, nil
+}
+
+func getSinkRef(u *unstructured.Unstructured) (GroupKindNamespacedName, bool) {
+	stringMap, ok, err := unstructured.NestedStringMap(u.Object, "spec", "sink", "ref")
+	if err != nil {
+		return GroupKindNamespacedName{}, false
+	}
+
+	if !ok {
+		return GroupKindNamespacedName{}, false
+	}
+
+	apiVersion, ok := stringMap["apiVersion"]
+	if !ok {
+		// if apiVersion is not present (e.g. using a URL as the sink), we don't care/
+		// same story with others
+		return GroupKindNamespacedName{}, false
+	}
+	kind, ok := stringMap["kind"]
+	if !ok {
+		return GroupKindNamespacedName{}, false
+	}
+	name, ok := stringMap["name"]
+	if !ok {
+		return GroupKindNamespacedName{}, false
+	}
+
+	return GroupKindNamespacedName{
+		Group:     util.APIVersionToGroup(apiVersion),
+		Kind:      kind,
+		Namespace: u.GetNamespace(),
+		Name:      name,
+	}, true
+}