Add labels for topics and pullsubscription

pkg/reconciler/events/auditlogs/auditlogs_test.go

@@ -161,7 +161,8 @@ func TestAllCases(t *testing.T) {
 					PropagationPolicy: "CreateDelete",
 				}),
 				WithTopicLabels(map[string]string{
-					"receive-adapter": receiveAdapterName,
+					"receive-adapter":                     receiveAdapterName,
+					"events.cloud.google.com/source-name": sourceName,
 				}),
 				WithTopicOwnerReferences([]metav1.OwnerReference{sourceOwnerRef(sourceName, sourceUID)}),
 			),
@@ -371,7 +372,8 @@ func TestAllCases(t *testing.T) {
 				}),
 				WithPullSubscriptionSink(sinkGVK, sinkName),
 				WithPullSubscriptionLabels(map[string]string{
-					"receive-adapter": receiveAdapterName,
+					"receive-adapter":                     receiveAdapterName,
+					"events.cloud.google.com/source-name": sourceName,
 				}),
 				WithPullSubscriptionAnnotations(map[string]string{
 					"metrics-resource-group": resourceGroup,

pkg/reconciler/events/pubsub/pubsub.go

@@ -150,14 +150,14 @@ func (r *Reconciler) reconcilePullSubscription(ctx context.Context, source *v1al
 			return nil, fmt.Errorf("failed to get PullSubscription: %w", err)
 		}
 		args := &resources.PullSubscriptionArgs{
-			Namespace:      source.Namespace,
-			Name:           source.Name,
-			Spec:           &source.Spec.PubSubSpec,
-			Owner:          source,
-			Topic:          source.Spec.Topic,
-			ReceiveAdapter: r.receiveAdapterName,
-			ResourceGroup:  resourceGroup,
-			Mode:           pubsubv1alpha1.ModePushCompatible,
+			Namespace:     source.Namespace,
+			Name:          source.Name,
+			Spec:          &source.Spec.PubSubSpec,
+			Owner:         source,
+			Topic:         source.Spec.Topic,
+			ResourceGroup: resourceGroup,
+			Mode:          pubsubv1alpha1.ModePushCompatible,
+			Labels:        resources.GetLabels(r.receiveAdapterName, source.Name),
 		}
 		newPS := resources.MakePullSubscription(args)
 		logging.FromContext(ctx).Desugar().Debug("Creating PullSubscription", zap.Any("ps", newPS))

pkg/reconciler/events/pubsub/pubsub_test.go

@@ -161,7 +161,8 @@ func TestAllCases(t *testing.T) {
 				WithPullSubscriptionSink(sinkGVK, sinkName),
 				WithPullSubscriptionMode(pubsubv1alpha1.ModePushCompatible),
 				WithPullSubscriptionLabels(map[string]string{
-					"receive-adapter": receiveAdapterName,
+					"receive-adapter":                     receiveAdapterName,
+					"events.cloud.google.com/source-name": pubsubName,
 				}),
 				WithPullSubscriptionAnnotations(map[string]string{
 					"metrics-resource-group": resourceGroup,

pkg/reconciler/events/scheduler/scheduler_test.go

@@ -189,7 +189,8 @@ func TestAllCases(t *testing.T) {
 					PropagationPolicy: "CreateDelete",
 				}),
 				WithTopicLabels(map[string]string{
-					"receive-adapter": receiveAdapterName,
+					"receive-adapter":                     receiveAdapterName,
+					"events.cloud.google.com/source-name": schedulerName,
 				}),
 				WithTopicOwnerReferences([]metav1.OwnerReference{ownerRef()}),
 			),
@@ -434,8 +435,8 @@ func TestAllCases(t *testing.T) {
 					}),
 					WithPullSubscriptionSink(sinkGVK, sinkName),
 					WithPullSubscriptionLabels(map[string]string{
-						"receive-adapter": receiveAdapterName,
-					}),
+						"receive-adapter":                     receiveAdapterName,
+						"events.cloud.google.com/source-name": schedulerName}),
 					WithPullSubscriptionAnnotations(map[string]string{
 						"metrics-resource-group": resourceGroup,
 					}),

pkg/reconciler/events/storage/storage_test.go

@@ -185,7 +185,8 @@ func TestAllCases(t *testing.T) {
 					PropagationPolicy: "CreateDelete",
 				}),
 				WithTopicLabels(map[string]string{
-					"receive-adapter": receiveAdapterName,
+					"receive-adapter":                     receiveAdapterName,
+					"events.cloud.google.com/source-name": storageName,
 				}),
 				WithTopicOwnerReferences([]metav1.OwnerReference{ownerRef()}),
 			),
@@ -416,7 +417,8 @@ func TestAllCases(t *testing.T) {
 				}),
 				WithPullSubscriptionSink(sinkGVK, sinkName),
 				WithPullSubscriptionLabels(map[string]string{
-					"receive-adapter": receiveAdapterName,
+					"receive-adapter":                     receiveAdapterName,
+					"events.cloud.google.com/source-name": storageName,
 				}),
 				WithPullSubscriptionAnnotations(map[string]string{
 					"metrics-resource-group": resourceGroup,

pkg/reconciler/pubsub/reconciler.go

@@ -29,7 +29,7 @@ import (
 	"go.uber.org/zap"
 	corev1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/logging"
 )
@@ -69,7 +69,15 @@ func (psb *PubSubBase) ReconcilePubSub(ctx context.Context, pubsubable duck.PubS
 			logging.FromContext(ctx).Desugar().Error("Failed to get Topics", zap.Error(err))
 			return nil, nil, fmt.Errorf("failed to get Topics: %w", err)
 		}
-		newTopic := resources.MakeTopic(namespace, name, spec, pubsubable, topic, psb.receiveAdapterName)
+		args := &resources.TopicArgs{
+			Namespace: namespace,
+			Name:      name,
+			Spec:      spec,
+			Owner:     pubsubable,
+			Topic:     topic,
+			Labels:    resources.GetLabels(psb.receiveAdapterName, name),
+		}
+		newTopic := resources.MakeTopic(args)
 		t, err = topics.Create(newTopic)
 		if err != nil {
 			logging.FromContext(ctx).Desugar().Error("Failed to create Topic", zap.Any("topic", newTopic), zap.Error(err))
@@ -92,14 +100,14 @@ func (psb *PubSubBase) ReconcilePubSub(ctx context.Context, pubsubable duck.PubS
 			return t, nil, fmt.Errorf("failed to get Pullsubscription: %w", err)
 		}
 		args := &resources.PullSubscriptionArgs{
-			Namespace:      namespace,
-			Name:           name,
-			Spec:           spec,
-			Owner:          pubsubable,
-			Topic:          topic,
-			ReceiveAdapter: psb.receiveAdapterName,
-			AdapterType:    psb.adapterType,
-			ResourceGroup:  resourceGroup,
+			Namespace:     namespace,
+			Name:          name,
+			Spec:          spec,
+			Owner:         pubsubable,
+			Topic:         topic,
+			AdapterType:   psb.adapterType,
+			ResourceGroup: resourceGroup,
+			Labels:        resources.GetLabels(psb.receiveAdapterName, name),
 		}
 		newPS := resources.MakePullSubscription(args)
 		ps, err = pullSubscriptions.Create(newPS)