Update sources event CE types, sources, subjects (#1340)

* Update sources event CE types, sources, subjects

* separate schemas from eventing api

* fix v1alpha1 api

* fix test

* fix yamls

* fix test images for cal and scheduler targets

* cleanups

config/core/resources/cloudauditlogssource.yaml

@@ -22,7 +22,7 @@ metadata:
   annotations:
     registry.knative.dev/eventTypes: |
       [
-        {"type": "com.google.cloud.auditlog.event", "schema": "type.googleapis.com/google.logging.v2.LogEntry", "description": "Common audit log event type for all Google Cloud Platform API operations." }
+        {"type": "google.cloud.audit.log.v1.written", "schema": "https://raw.githubusercontent.com/googleapis/google-cloudevents/master/proto/google/events/cloud/audit/v1/data.proto", "description": "Common audit log event type for all Google Cloud Platform API operations." }
       ]
   name: cloudauditlogssources.events.cloud.google.com
 spec:

config/core/resources/cloudpubsubsource.yaml

@@ -22,7 +22,7 @@ metadata:
   annotations:
     registry.knative.dev/eventTypes: |
       [
-        { "type": "com.google.cloud.pubsub.topic.publish", "description": "This event is sent when a message is published to a Cloud Pub/Sub topic."}
+        { "type": "google.cloud.pubsub.topic.v1.messagePublished", "schema": "https://raw.githubusercontent.com/googleapis/google-cloudevents/master/proto/google/events/cloud/pubsub/v1/data.proto", "description": "This event is sent when a message is published to a Cloud Pub/Sub topic."}
       ]
   name: cloudpubsubsources.events.cloud.google.com
 spec:

config/core/resources/cloudschedulersource.yaml

@@ -22,7 +22,7 @@ metadata:
   annotations:
     registry.knative.dev/eventTypes: |
       [
-        { "type": "com.google.cloud.scheduler.job.execute", "description": "This event is sent when a job is executed in Cloud Scheduler."}
+        { "type": "google.cloud.scheduler.job.v1.executed", "schema":"https://raw.githubusercontent.com/googleapis/google-cloudevents/master/proto/google/events/cloud/scheduler/v1/data.proto", "description": "This event is sent when a job is executed in Cloud Scheduler."}
       ]
   name: cloudschedulersources.events.cloud.google.com
 spec:

config/core/resources/cloudstoragesource.yaml

@@ -22,10 +22,10 @@ metadata:
   annotations:
     registry.knative.dev/eventTypes: |
       [
-        { "type": "com.google.cloud.storage.object.finalize", "schema": "https://raw.githubusercontent.com/google/knative-gcp/master/schemas/storage/schema.json", "description": "Sent when a new object (or a new generation of an existing object) is successfully created in the bucket. This includes copying or rewriting an existing object. A failed upload does not trigger this event." },
-        { "type": "com.google.cloud.storage.object.delete", "schema": "https://raw.githubusercontent.com/google/knative-gcp/master/schemas/storage/schema.json", "description": "Sent when an object has been permanently deleted. This includes objects that are overwritten or are deleted as part of the bucket's lifecycle configuration. For buckets with object versioning enabled, this is not sent when an object is archived."},
-        { "type": "com.google.cloud.storage.object.archive", "schema": "https://raw.githubusercontent.com/google/knative-gcp/master/schemas/storage/schema.json", "description": "Only sent when a bucket has enabled object versioning. This event indicates that the live version of an object has become an archived version, either because it was archived or because it was overwritten by the upload of an object of the same name."},
-        { "type": "com.google.cloud.storage.object.metadataUpdate", "schema": "https://raw.githubusercontent.com/google/knative-gcp/master/schemas/storage/schema.json", "description": "Sent when the metadata of an existing object changes." }
+        { "type": "google.cloud.storage.object.v1.finalized", "schema": "https://raw.githubusercontent.com/googleapis/google-cloudevents/master/proto/google/events/cloud/storage/v1/data.proto", "description": "Sent when a new object (or a new generation of an existing object) is successfully created in the bucket. This includes copying or rewriting an existing object. A failed upload does not trigger this event." },
+        { "type": "google.cloud.storage.object.v1.deleted", "schema": "https://raw.githubusercontent.com/googleapis/google-cloudevents/master/proto/google/events/cloud/storage/v1/data.proto", "description": "Sent when an object has been permanently deleted. This includes objects that are overwritten or are deleted as part of the bucket's lifecycle configuration. For buckets with object versioning enabled, this is not sent when an object is archived."},
+        { "type": "google.cloud.storage.object.v1.archived", "schema": "https://raw.githubusercontent.com/googleapis/google-cloudevents/master/proto/google/events/cloud/storage/v1/data.proto", "description": "Only sent when a bucket has enabled object versioning. This event indicates that the live version of an object has become an archived version, either because it was archived or because it was overwritten by the upload of an object of the same name."},
+        { "type": "google.cloud.storage.object.v1.metadataUpdated", "schema": "https://raw.githubusercontent.com/googleapis/google-cloudevents/master/proto/google/events/cloud/storage/v1/data.proto", "description": "Sent when the metadata of an existing object changes." }
       ]
   name: cloudstoragesources.events.cloud.google.com
 spec:
@@ -154,10 +154,10 @@ spec:
               items:
                 type: string
                 enum:
-                  - com.google.cloud.storage.object.finalize
-                  - com.google.cloud.storage.object.delete
-                  - com.google.cloud.storage.object.archive
-                  - com.google.cloud.storage.object.metadataUpdate
+                  - google.cloud.storage.object.v1.finalized
+                  - google.cloud.storage.object.v1.deleted
+                  - google.cloud.storage.object.v1.archived
+                  - google.cloud.storage.object.v1.metadataUpdated
         status:
           type: object
           properties:

pkg/apis/events/v1alpha1/cloudauditlogssource_types.go

@@ -17,9 +17,6 @@ limitations under the License.
 package v1alpha1
 
 import (
-	"crypto/md5"
-	"fmt"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -65,21 +62,6 @@ var auditLogsSourceCondSet = apis.NewLivingConditionSet(
 	SinkReady,
 )
 
-const (
-	CloudAuditLogsSourceEvent = "com.google.cloud.auditlog.event"
-)
-
-// CloudAuditLogsSourceEventSource returns the Cloud Audit Logs CloudEvent source value.
-func CloudAuditLogsSourceEventSource(serviceName, parentResource string) string {
-	return fmt.Sprintf("//%s/%s", serviceName, parentResource)
-}
-
-// CloudAuditLogsSourceEventID returns the Cloud Audit Logs CloudEvent id value.
-func CloudAuditLogsSourceEventID(id, logName, timestamp string) string {
-	// Hash the concatenation of the three fields.
-	return fmt.Sprintf("%x", md5.Sum([]byte(id+logName+timestamp)))
-}
-
 type CloudAuditLogsSourceSpec struct {
 	// This brings in the PubSub based Source Specs. Includes:
 	duckv1alpha1.PubSubSpec `json:",inline"`

pkg/apis/events/v1alpha1/cloudauditlogssource_types_test.go

@@ -68,26 +68,6 @@ func TestAuditLogsConditionSet(t *testing.T) {
 	}
 }
 
-func TestCloudAuditLogsSourceEventSource(t *testing.T) {
-	want := "//pubsub.googleapis.com/projects/PROJECT"
-
-	got := CloudAuditLogsSourceEventSource("pubsub.googleapis.com", "projects/PROJECT")
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("failed to get expected (-want, +got) = %v", diff)
-	}
-}
-
-func TestCloudAuditLogsSourceEventID(t *testing.T) {
-	want := "efdb9bf7d6fdfc922352530c1ba51242"
-
-	got := CloudAuditLogsSourceEventID("pt9y76cxw5", "projects/knative-project-228222/logs/cloudaudit.googleapis.com%2Factivity", "2020-01-19T22:45:03.439395442Z")
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("failed to get expected (-want, +got) = %v", diff)
-	}
-}
-
 func TestCloudAuditLogsSourceIdentitySpec(t *testing.T) {
 	s := &CloudAuditLogsSource{
 		Spec: CloudAuditLogsSourceSpec{

pkg/apis/events/v1alpha1/cloudpubsubsource_types.go

@@ -17,7 +17,6 @@ limitations under the License.
 package v1alpha1
 
 import (
-	"fmt"
 	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
@@ -109,16 +108,6 @@ func (ps CloudPubSubSourceSpec) GetRetentionDuration() time.Duration {
 	return defaultRetentionDuration
 }
 
-// CloudPubSubSourceEventSource returns the Cloud Pub/Sub CloudEvent source value.
-func CloudPubSubSourceEventSource(googleCloudProject, topic string) string {
-	return fmt.Sprintf("//pubsub.googleapis.com/projects/%s/topics/%s", googleCloudProject, topic)
-}
-
-const (
-	// CloudPubSubSource CloudEvent type
-	CloudPubSubSourcePublish = "com.google.cloud.pubsub.topic.publish"
-)
-
 const (
 	// CloudPubSubSourceConditionReady has status True when the CloudPubSubSource is
 	// ready to send events.

pkg/apis/events/v1alpha1/cloudpubsubsource_types_test.go

@@ -30,16 +30,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 )
 
-func TestCloudPubSubSourceEventSource(t *testing.T) {
-	want := "//pubsub.googleapis.com/projects/PROJECT/topics/TOPIC"
-
-	got := CloudPubSubSourceEventSource("PROJECT", "TOPIC")
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("failed to get expected (-want, +got) = %v", diff)
-	}
-}
-
 func TestCloudPubSubSourceGetGroupVersionKind(t *testing.T) {
 	want := schema.GroupVersionKind{
 		Group:   "events.cloud.google.com",

pkg/apis/events/v1alpha1/cloudschedulersource_types.go

@@ -17,8 +17,6 @@ limitations under the License.
 package v1alpha1
 
 import (
-	"fmt"
-
 	"k8s.io/apimachinery/pkg/runtime"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -55,19 +53,6 @@ var (
 	_ kngcpduck.PubSubable         = (*CloudSchedulerSource)(nil)
 )
 
-const (
-	// CloudEvent types used by CloudSchedulerSource.
-	CloudSchedulerSourceExecute = "com.google.cloud.scheduler.job.execute"
-	// CloudSchedulerSourceJobName is the Pub/Sub message attribute key with the CloudSchedulerSource's job name.
-	CloudSchedulerSourceJobName = "jobName"
-	// CloudSchedulerSourceName is the Pub/Sub message attribute key with the CloudSchedulerSource's name.
-	CloudSchedulerSourceName = "schedulerName"
-)
-
-func CloudSchedulerSourceEventSource(parent, scheduler string) string {
-	return fmt.Sprintf("//cloudscheduler.googleapis.com/%s/schedulers/%s", parent, scheduler)
-}
-
 // CloudSchedulerSourceSpec is the spec for a CloudSchedulerSource resource
 type CloudSchedulerSourceSpec struct {
 	// This brings in the PubSub based Source Specs. Includes:

pkg/apis/events/v1alpha1/cloudschedulersource_types_test.go

@@ -42,16 +42,6 @@ func TestCloudSchedulerSourceGetGroupVersionKind(t *testing.T) {
 	}
 }
 
-func TestCloudSchedulerSourceEventSource(t *testing.T) {
-	want := "//cloudscheduler.googleapis.com/PARENT/schedulers/SCHEDULER"
-
-	got := CloudSchedulerSourceEventSource("PARENT", "SCHEDULER")
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("failed to get expected (-want, +got) = %v", diff)
-	}
-}
-
 func TestCloudSchedulerSourceConditionSet(t *testing.T) {
 	want := []apis.Condition{{
 		Type: JobReady,

pkg/apis/events/v1alpha1/cloudstoragesource_defaults.go

@@ -23,9 +23,15 @@ import (
 
 	duckv1alpha1 "github.com/google/knative-gcp/pkg/apis/duck/v1alpha1"
 	metadataClient "github.com/google/knative-gcp/pkg/gclient/metadata"
+	schemasv1 "github.com/google/knative-gcp/pkg/schemas/v1"
 )
 
-var allEventTypes = []string{CloudStorageSourceFinalize, CloudStorageSourceDelete, CloudStorageSourceArchive, CloudStorageSourceMetadataUpdate}
+var allEventTypes = []string{
+	schemasv1.CloudStorageObjectFinalizedEventType,
+	schemasv1.CloudStorageObjectDeletedEventType,
+	schemasv1.CloudStorageObjectArchivedEventType,
+	schemasv1.CloudStorageObjectMetadataUpdatedEventType,
+}
 
 func (s *CloudStorageSource) SetDefaults(ctx context.Context) {
 	ctx = apis.WithinParent(ctx, s.ObjectMeta)

pkg/apis/events/v1alpha1/cloudstoragesource_defaults_test.go

@@ -25,6 +25,7 @@ import (
 
 	duckv1alpha1 "github.com/google/knative-gcp/pkg/apis/duck/v1alpha1"
 	testingMetadataClient "github.com/google/knative-gcp/pkg/gclient/metadata/testing"
+	schemasv1 "github.com/google/knative-gcp/pkg/schemas/v1"
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -51,7 +52,7 @@ func TestCloudStorageSourceSpec_SetDefaults(t *testing.T) {
 		},
 		"defaults present": {
 			orig: &CloudStorageSourceSpec{
-				EventTypes: []string{CloudStorageSourceFinalize, CloudStorageSourceDelete},
+				EventTypes: []string{schemasv1.CloudStorageObjectFinalizedEventType, schemasv1.CloudStorageObjectDeletedEventType},
 				PubSubSpec: duckv1alpha1.PubSubSpec{
 					Secret: &corev1.SecretKeySelector{
 						LocalObjectReference: corev1.LocalObjectReference{
@@ -62,7 +63,7 @@ func TestCloudStorageSourceSpec_SetDefaults(t *testing.T) {
 				},
 			},
 			expected: &CloudStorageSourceSpec{
-				EventTypes: []string{CloudStorageSourceFinalize, CloudStorageSourceDelete},
+				EventTypes: []string{schemasv1.CloudStorageObjectFinalizedEventType, schemasv1.CloudStorageObjectDeletedEventType},
 				PubSubSpec: duckv1alpha1.PubSubSpec{
 					Secret: &corev1.SecretKeySelector{
 						LocalObjectReference: corev1.LocalObjectReference{
@@ -106,10 +107,10 @@ func TestCloudStorageSource_SetDefaults(t *testing.T) {
 				},
 				Spec: CloudStorageSourceSpec{
 					EventTypes: []string{
-						"com.google.cloud.storage.object.finalize",
-						"com.google.cloud.storage.object.delete",
-						"com.google.cloud.storage.object.archive",
-						"com.google.cloud.storage.object.metadataUpdate",
+						schemasv1.CloudStorageObjectFinalizedEventType,
+						schemasv1.CloudStorageObjectDeletedEventType,
+						schemasv1.CloudStorageObjectArchivedEventType,
+						schemasv1.CloudStorageObjectMetadataUpdatedEventType,
 					},
 					PubSubSpec: duckv1alpha1.PubSubSpec{
 						Secret: &corev1.SecretKeySelector{
@@ -148,10 +149,10 @@ func TestCloudStorageSource_SetDefaults(t *testing.T) {
 				},
 				Spec: CloudStorageSourceSpec{
 					EventTypes: []string{
-						"com.google.cloud.storage.object.finalize",
-						"com.google.cloud.storage.object.delete",
-						"com.google.cloud.storage.object.archive",
-						"com.google.cloud.storage.object.metadataUpdate",
+						schemasv1.CloudStorageObjectFinalizedEventType,
+						schemasv1.CloudStorageObjectDeletedEventType,
+						schemasv1.CloudStorageObjectArchivedEventType,
+						schemasv1.CloudStorageObjectMetadataUpdatedEventType,
 					},
 					PubSubSpec: duckv1alpha1.PubSubSpec{
 						Secret: &corev1.SecretKeySelector{

pkg/apis/events/v1alpha1/cloudstoragesource_types.go

@@ -17,8 +17,6 @@ limitations under the License.
 package v1alpha1
 
 import (
-	"fmt"
-
 	"k8s.io/apimachinery/pkg/runtime"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -78,17 +76,6 @@ type CloudStorageSourceSpec struct {
 	PayloadFormat string `json:"payloadFormat,omitempty"`
 }
 
-const (
-	// CloudEvent types used by CloudStorageSource.
-	CloudStorageSourceFinalize       = "com.google.cloud.storage.object.finalize"
-	CloudStorageSourceArchive        = "com.google.cloud.storage.object.archive"
-	CloudStorageSourceDelete         = "com.google.cloud.storage.object.delete"
-	CloudStorageSourceMetadataUpdate = "com.google.cloud.storage.object.metadataUpdate"
-
-	// CloudEvent source prefix.
-	storageSourcePrefix = "//storage.googleapis.com/buckets"
-)
-
 const (
 	// CloudStorageSourceConditionReady has status True when the CloudStorageSource is ready to send events.
 	CloudStorageSourceConditionReady = apis.ConditionReady
@@ -98,10 +85,6 @@ const (
 	NotificationReady apis.ConditionType = "NotificationReady"
 )
 
-func CloudStorageSourceEventSource(bucket string) string {
-	return fmt.Sprintf("%s/%s", storageSourcePrefix, bucket)
-}
-
 var storageCondSet = apis.NewLivingConditionSet(
 	duckv1alpha1.PullSubscriptionReady,
 	duckv1alpha1.TopicReady,

pkg/apis/events/v1alpha1/cloudstoragesource_types_test.go

@@ -41,15 +41,6 @@ func TestGetGroupVersionKind(t *testing.T) {
 	}
 }
 
-func TestCloudStorageSourceEventSource(t *testing.T) {
-	want := "//storage.googleapis.com/buckets/bucket"
-	got := CloudStorageSourceEventSource("bucket")
-
-	if diff := cmp.Diff(want, got); diff != "" {
-		t.Errorf("failed to get expected (-want, +got) = %v", diff)
-	}
-}
-
 func TestCloudStorageSourceSourceConditionSet(t *testing.T) {
 	want := []apis.Condition{{
 		Type: NotificationReady,

pkg/apis/events/v1alpha1/cloudstoragesource_validation_test.go

@@ -25,6 +25,7 @@ import (
 
 	duckv1alpha1 "github.com/google/knative-gcp/pkg/apis/duck/v1alpha1"
 	metadatatesting "github.com/google/knative-gcp/pkg/gclient/metadata/testing"
+	schemasv1 "github.com/google/knative-gcp/pkg/schemas/v1"
 
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
@@ -77,7 +78,7 @@ var (
 	// Bucket, Sink, Secret, Event Type and Project, ObjectNamePrefix and PayloadFormat
 	storageSourceSpec = CloudStorageSourceSpec{
 		Bucket:           "my-test-bucket",
-		EventTypes:       []string{CloudStorageSourceFinalize, CloudStorageSourceDelete},
+		EventTypes:       []string{schemasv1.CloudStorageObjectFinalizedEventType, schemasv1.CloudStorageObjectDeletedEventType},
 		ObjectNamePrefix: "test-prefix",
 		PayloadFormat:    cloudevents.ApplicationJSON,
 		PubSubSpec: duckv1alpha1.PubSubSpec{
@@ -381,7 +382,7 @@ func TestCheckImmutableFields(t *testing.T) {
 			orig: &storageSourceSpec,
 			updated: CloudStorageSourceSpec{
 				Bucket:           storageSourceSpec.Bucket,
-				EventTypes:       []string{CloudStorageSourceMetadataUpdate},
+				EventTypes:       []string{schemasv1.CloudStorageObjectMetadataUpdatedEventType},
 				ObjectNamePrefix: storageSourceSpec.ObjectNamePrefix,
 				PayloadFormat:    storageSourceSpec.PayloadFormat,
 				PubSubSpec:       storageSourceSpec.PubSubSpec,

pkg/apis/events/v1beta1/cloudauditlogssource_types.go

@@ -17,9 +17,6 @@ limitations under the License.
 package v1beta1
 
 import (
-	"crypto/md5"
-	"fmt"
-
 	duckv1beta1 "github.com/google/knative-gcp/pkg/apis/duck/v1beta1"
 	kngcpduck "github.com/google/knative-gcp/pkg/duck/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -67,21 +64,6 @@ var auditLogsSourceCondSet = apis.NewLivingConditionSet(
 	SinkReady,
 )
 
-const (
-	CloudAuditLogsSourceEvent = "com.google.cloud.auditlog.event"
-)
-
-// CloudAuditLogsSourceEventSource returns the Cloud Audit Logs CloudEvent source value.
-func CloudAuditLogsSourceEventSource(serviceName, parentResource string) string {
-	return fmt.Sprintf("//%s/%s", serviceName, parentResource)
-}
-
-// CloudAuditLogsSourceEventID returns the Cloud Audit Logs CloudEvent id value.
-func CloudAuditLogsSourceEventID(id, logName, timestamp string) string {
-	// Hash the concatenation of the three fields.
-	return fmt.Sprintf("%x", md5.Sum([]byte(id+logName+timestamp)))
-}
-
 type CloudAuditLogsSourceSpec struct {
 	// This brings in the PubSub based Source Specs. Includes:
 	duckv1beta1.PubSubSpec `json:",inline"`