feat: add constructor for full event lineage graph

pkg/graph/constructor.go

@@ -17,14 +17,155 @@ limitations under the License.
 package graph
 
 import (
+	"context"
+	"encoding/json"
 	"fmt"
 
+	"go.uber.org/zap"
+	apierrs "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/client-go/dynamic"
+	rest "k8s.io/client-go/rest"
+
 	eventingv1 "knative.dev/eventing/pkg/apis/eventing/v1"
 	eventingv1beta3 "knative.dev/eventing/pkg/apis/eventing/v1beta3"
 	messagingv1 "knative.dev/eventing/pkg/apis/messaging/v1"
+	eventingclient "knative.dev/eventing/pkg/client/clientset/versioned"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 )
 
+type ConstructorConfig struct {
+	RestConfig            rest.Config
+	Namespaces            []string
+	ShouldAddBroker       func(b eventingv1.Broker) bool
+	ShouldAddChannel      func(c messagingv1.Channel) bool
+	ShouldAddSource       func(s duckv1.Source) bool
+	ShouldAddTrigger      func(t eventingv1.Trigger) bool
+	ShouldAddSubscription func(s messagingv1.Subscription) bool
+	ShouldAddEventType    func(et eventingv1beta3.EventType) bool
+}
+
+func ConstructGraph(ctx context.Context, config ConstructorConfig, logger zap.Logger) (*Graph, error) {
+	eventingClient, err := eventingclient.NewForConfig(&config.RestConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	g := NewGraph()
+
+	for _, ns := range config.Namespaces {
+		brokers, err := eventingClient.EventingV1().Brokers(ns).List(ctx, metav1.ListOptions{})
+		if err != nil && !apierrs.IsNotFound(err) && !apierrs.IsUnauthorized(err) && !apierrs.IsForbidden(err) {
+			return nil, err
+		}
+
+		if apierrs.IsUnauthorized(err) || apierrs.IsForbidden(err) {
+			logger.Warn("failed to list brokers while constructing lineage graph", zap.Error(err))
+		}
+
+		if err == nil {
+			for _, broker := range brokers.Items {
+				if config.ShouldAddBroker(broker) {
+					g.AddBroker(broker)
+				}
+			}
+		}
+
+		channels, err := eventingClient.MessagingV1().Channels(ns).List(ctx, metav1.ListOptions{})
+		if err != nil && !apierrs.IsNotFound(err) && !apierrs.IsUnauthorized(err) && !apierrs.IsForbidden(err) {
+			return nil, err
+		}
+
+		if apierrs.IsUnauthorized(err) || apierrs.IsForbidden(err) {
+			logger.Warn("failed to list channels while constructing lineage graph", zap.Error(err))
+		}
+
+		if err == nil {
+			for _, channel := range channels.Items {
+				if config.ShouldAddChannel(channel) {
+					g.AddChannel(channel)
+				}
+			}
+		}
+
+		sources, err := getSources(ctx, config)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, source := range sources {
+			if config.ShouldAddSource(source) {
+				g.AddSource(source)
+			}
+		}
+
+		triggers, err := eventingClient.EventingV1().Triggers(ns).List(ctx, metav1.ListOptions{})
+		if err != nil && !apierrs.IsNotFound(err) && !apierrs.IsUnauthorized(err) && !apierrs.IsForbidden(err) {
+			return nil, err
+		}
+
+		if apierrs.IsUnauthorized(err) || apierrs.IsForbidden(err) {
+			logger.Warn("failed to list triggers while constructing lineage graph", zap.Error(err))
+		}
+
+		if err == nil {
+			for _, trigger := range triggers.Items {
+				if config.ShouldAddTrigger(trigger) {
+					err := g.AddTrigger(trigger)
+					if err != nil {
+						return nil, err
+					}
+				}
+			}
+		}
+
+		subscriptions, err := eventingClient.MessagingV1().Subscriptions(ns).List(ctx, metav1.ListOptions{})
+		if err != nil && !apierrs.IsNotFound(err) && !apierrs.IsUnauthorized(err) && !apierrs.IsForbidden(err) {
+			return nil, err
+		}
+
+		if apierrs.IsUnauthorized(err) || apierrs.IsForbidden(err) {
+			logger.Warn("failed to list subscriptions while constructing lineage graph", zap.Error(err))
+		}
+
+		if err == nil {
+			for _, subscription := range subscriptions.Items {
+				if config.ShouldAddSubscription(subscription) {
+					err := g.AddSubscription(subscription)
+					if err != nil {
+						return nil, err
+					}
+				}
+			}
+		}
+
+		eventTypes, err := eventingClient.EventingV1beta3().EventTypes(ns).List(ctx, metav1.ListOptions{})
+		if err != nil && !apierrs.IsNotFound(err) && !apierrs.IsUnauthorized(err) && !apierrs.IsForbidden(err) {
+			return nil, err
+		}
+
+		if apierrs.IsUnauthorized(err) || apierrs.IsForbidden(err) {
+			logger.Warn("failed to list eventtypes while constructing lineage graph", zap.Error(err))
+		}
+
+		if err == nil {
+			for _, eventType := range eventTypes.Items {
+				if config.ShouldAddEventType(eventType) {
+					err := g.AddEventType(eventType)
+					if err != nil {
+						return nil, err
+					}
+				}
+			}
+		}
+	}
+
+	return g, nil
+}
+
 func (g *Graph) AddBroker(broker eventingv1.Broker) {
 	ref := &duckv1.KReference{
 		Name:       broker.Name,
@@ -74,7 +215,7 @@ func (g *Graph) AddChannel(channel messagingv1.Channel) {
 	v.AddEdge(to, dest, NoTransform{}, true)
 }
 
-func (g *Graph) AddEventType(et *eventingv1beta3.EventType) error {
+func (g *Graph) AddEventType(et eventingv1beta3.EventType) error {
 	ref := &duckv1.KReference{
 		Name:       et.Name,
 		Namespace:  et.Namespace,
@@ -89,15 +230,15 @@ func (g *Graph) AddEventType(et *eventingv1beta3.EventType) error {
 			return fmt.Errorf("trigger/subscription must have a primary outward edge, but had none")
 		}
 
-		outEdge.To().AddEdge(outEdge.From(), dest, EventTypeTransform{EventType: et}, false)
+		outEdge.To().AddEdge(outEdge.From(), dest, EventTypeTransform{EventType: &et}, false)
 
 		return nil
 	}
 
 	from := g.getOrCreateVertex(dest)
 	to := g.getOrCreateVertex(&duckv1.Destination{Ref: et.Spec.Reference})
 
-	from.AddEdge(to, dest, EventTypeTransform{EventType: et}, false)
+	from.AddEdge(to, dest, EventTypeTransform{EventType: &et}, false)
 
 	return nil
 }
@@ -155,6 +296,7 @@ func (g *Graph) AddTrigger(trigger eventingv1.Trigger) error {
 	return nil
 
 }
+
 func (g *Graph) AddSubscription(subscription messagingv1.Subscription) error {
 	channelRef := &duckv1.KReference{
 		Name:       subscription.Spec.Channel.Name,
@@ -197,6 +339,132 @@ func (g *Graph) AddSubscription(subscription messagingv1.Subscription) error {
 
 }
 
+func getSources(ctx context.Context, config ConstructorConfig) ([]duckv1.Source, error) {
+	client, err := dynamic.NewForConfig(&config.RestConfig)
+	if err != nil {
+		return nil, err
+	}
+
+	sourceCRDs, err := client.Resource(
+		schema.GroupVersionResource{
+			Group:    "apiextentions.k8s.io",
+			Version:  "v1",
+			Resource: "customresourcedefinitions",
+		},
+	).List(ctx, metav1.ListOptions{LabelSelector: labels.Set{"duck.knative.dev/source": "true"}.String()})
+	if err != nil {
+		return nil, fmt.Errorf("unable to list source CRDs: %w", err)
+	}
+
+	duckSources := []duckv1.Source{}
+
+	for i := range sourceCRDs.Items {
+		sourceCrd := sourceCRDs.Items[i]
+		sourceGVR, err := gvrFromUnstructured(&sourceCrd)
+		if err != nil {
+			continue
+		}
+
+		for _, ns := range config.Namespaces {
+			sourcesList, err := client.Resource(sourceGVR).Namespace(ns).List(ctx, metav1.ListOptions{})
+			if err != nil {
+				continue
+			}
+
+			for i := range sourcesList.Items {
+				unstructuredSource := sourcesList.Items[i]
+				duckSource, err := duckSourceFromUnstructured(&unstructuredSource)
+				if err == nil {
+					duckSources = append(duckSources, duckSource)
+				}
+			}
+		}
+	}
+
+	return duckSources, nil
+}
+
+func duckSourceFromUnstructured(u *unstructured.Unstructured) (duckv1.Source, error) {
+	duckSource := duckv1.Source{}
+	marshalled, err := u.MarshalJSON()
+	if err != nil {
+		return duckSource, err
+	}
+
+	err = json.Unmarshal(marshalled, &duckSource)
+	return duckSource, err
+}
+
+func gvrFromUnstructured(u *unstructured.Unstructured) (schema.GroupVersionResource, error) {
+	group, err := groupFromUnstructured(u)
+	if err != nil {
+		return schema.GroupVersionResource{}, err
+	}
+
+	version, err := versionFromUnstructured(u)
+	if err != nil {
+		return schema.GroupVersionResource{}, err
+	}
+
+	resource, err := resourceFromUnstructured(u)
+	if err != nil {
+		return schema.GroupVersionResource{}, err
+	}
+
+	return schema.GroupVersionResource{
+		Group:    group,
+		Version:  version,
+		Resource: resource,
+	}, nil
+}
+
+func groupFromUnstructured(u *unstructured.Unstructured) (string, error) {
+	content := u.UnstructuredContent()
+	group, found, err := unstructured.NestedString(content, "spec", "group")
+	if !found || err != nil {
+		return "", fmt.Errorf("can't find source kind from source CRD: %w", err)
+	}
+
+	return group, nil
+}
+
+func versionFromUnstructured(u *unstructured.Unstructured) (string, error) {
+	content := u.UnstructuredContent()
+	var version string
+	versions, found, err := unstructured.NestedSlice(content, "spec", "versions")
+	if !found || err != nil || len(versions) == 0 {
+		version, found, err = unstructured.NestedString(content, "spec", "version")
+		if !found || err != nil {
+			return "", fmt.Errorf("can't find source version from source CRD: %w", err)
+		}
+	} else {
+		for _, v := range versions {
+			if vmap, ok := v.(map[string]interface{}); ok {
+				if vmap["served"] == true {
+					version = vmap["name"].(string)
+					break
+				}
+			}
+		}
+	}
+
+	if version == "" {
+		return "", fmt.Errorf("can't find source version from source CRD: %w", err)
+	}
+
+	return version, nil
+}
+
+func resourceFromUnstructured(u *unstructured.Unstructured) (string, error) {
+	content := u.UnstructuredContent()
+	resource, found, err := unstructured.NestedString(content, "spec", "names", "plural")
+	if !found || err != nil {
+		return "", fmt.Errorf("can't find source resource from source CRD: %w", err)
+	}
+
+	return resource, nil
+}
+
 func getTransformForTrigger(trigger eventingv1.Trigger) Transform {
 	if len(trigger.Spec.Filters) == 0 && trigger.Spec.Filter != nil {
 		return &AttributesFilterTransform{Filter: trigger.Spec.Filter}

pkg/graph/constructor_test.go

@@ -1788,12 +1788,12 @@ func TestAddChannel(t *testing.T) {
 func TestAddEventType(t *testing.T) {
 	tests := []struct {
 		name     string
-		et       *eventingv1beta3.EventType
+		et       eventingv1beta3.EventType
 		expected map[comparableDestination]*Vertex
 	}{
 		{
 			name: "ET references source",
-			et: &eventingv1beta3.EventType{
+			et: eventingv1beta3.EventType{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      "my-EventType",
 					Namespace: "default",