Update sources event CE types, sources, subjects

pkg/apis/events/v1beta1/cloudauditlogssource_types.go

@@ -68,12 +68,20 @@ var auditLogsSourceCondSet = apis.NewLivingConditionSet(
 )
 
 const (
-	CloudAuditLogsSourceEvent = "com.google.cloud.auditlog.event"
+	// Ref: https://github.com/googleapis/google-cloudevents/blob/master/proto/google/events/cloud/audit/v1/events.proto#L27
+	// TODO: somehow reference the proto value directly.
+	CloudAuditLogsSourceLogWrittenEventType = "google.cloud.audit.log.v1.written"
+	CloudAuditLogsSourceEventDataSchema     = "https://github.com/googleapis/google-cloudevents/blob/master/proto/google/events/cloud/audit/v1/events.proto"
 )
 
 // CloudAuditLogsSourceEventSource returns the Cloud Audit Logs CloudEvent source value.
-func CloudAuditLogsSourceEventSource(serviceName, parentResource string) string {
-	return fmt.Sprintf("//%s/%s", serviceName, parentResource)
+// Format e.g. //cloudaudit.googleapis.com/projects/project-id/[activity|data_access]
+func CloudAuditLogsSourceEventSource(parentResource, activity string) string {
+	src := fmt.Sprintf("//cloudaudit.googleapis.com/%s", parentResource)
+	if activity != "" {
+		src = src + "/" + activity
+	}
+	return src
 }
 
 // CloudAuditLogsSourceEventID returns the Cloud Audit Logs CloudEvent id value.
@@ -82,6 +90,11 @@ func CloudAuditLogsSourceEventID(id, logName, timestamp string) string {
 	return fmt.Sprintf("%x", md5.Sum([]byte(id+logName+timestamp)))
 }
 
+// CloudAuditLogsSourceEventSubject returns the Cloud Audit Logs CloudEvent subject value.
+func CloudAuditLogsSourceEventSubject(serviceName, resourceName string) string {
+	return fmt.Sprintf("%s/%s", serviceName, resourceName)
+}
+
 type CloudAuditLogsSourceSpec struct {
 	// This brings in the PubSub based Source Specs. Includes:
 	duckv1beta1.PubSubSpec `json:",inline"`

pkg/apis/events/v1beta1/cloudauditlogssource_types_test.go

@@ -69,15 +69,23 @@ func TestAuditLogsConditionSet(t *testing.T) {
 }
 
 func TestCloudAuditLogsSourceEventSource(t *testing.T) {
-	want := "//pubsub.googleapis.com/projects/PROJECT"
+	want := "//cloudaudit.googleapis.com/projects/PROJECT/activity"
 
-	got := CloudAuditLogsSourceEventSource("pubsub.googleapis.com", "projects/PROJECT")
+	got := CloudAuditLogsSourceEventSource("projects/PROJECT", "activity")
 
 	if diff := cmp.Diff(want, got); diff != "" {
 		t.Errorf("failed to get expected (-want, +got) = %v", diff)
 	}
 }
 
+func TestCloudAuditLogsSourceEventSubject(t *testing.T) {
+	want := "pubsub.googleapis.com/projects/PROJECT/topics/TOPIC"
+	got := CloudAuditLogsSourceEventSubject("pubsub.googleapis.com", "projects/PROJECT/topics/TOPIC")
+	if got != want {
+		t.Errorf("CloudAuditLogsSourceEventSubject got=%s, want=%s", got, want)
+	}
+}
+
 func TestCloudAuditLogsSourceEventID(t *testing.T) {
 	want := "efdb9bf7d6fdfc922352530c1ba51242"
 

pkg/apis/events/v1beta1/cloudpubsubsource_types.go

@@ -116,8 +116,10 @@ func CloudPubSubSourceEventSource(googleCloudProject, topic string) string {
 }
 
 const (
-	// CloudPubSubSource CloudEvent type
-	CloudPubSubSourcePublish = "com.google.cloud.pubsub.topic.publish"
+	// ref: https://github.com/googleapis/google-cloudevents/blob/master/proto/google/events/cloud/pubsub/v1/events.proto#L26
+	// TODO: somehow reference the proto value directly.
+	CloudPubSubSourceMessagePublishedEventType = "google.cloud.pubsub.topic.v1.messagePublished"
+	CloudPubSubSourceEventDataSchema           = "https://github.com/googleapis/google-cloudevents/blob/master/proto/google/events/cloud/pubsub/v1/events.proto"
 )
 
 const (

pkg/apis/events/v1beta1/cloudschedulersource_types.go

@@ -58,15 +58,18 @@ var (
 
 const (
 	// CloudEvent types used by CloudSchedulerSource.
-	CloudSchedulerSourceExecute = "com.google.cloud.scheduler.job.execute"
+	// ref: https://github.com/googleapis/google-cloudevents/blob/master/proto/google/events/cloud/scheduler/v1/events.proto#L26
+	// TODO: somehow reference the proto values directly.
+	CloudSchedulerSourceJobExecutedEventType = "google.cloud.scheduler.job.v1.executed"
+	CloudSchedulerSourceEventDataSchema      = "https://github.com/googleapis/google-cloudevents/blob/master/proto/google/events/cloud/scheduler/v1/events.proto"
 	// 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)
+func CloudSchedulerSourceEventSource(jobName string) string {
+	return fmt.Sprintf("//cloudscheduler.googleapis.com/%s", jobName)
 }
 
 // CloudSchedulerSourceSpec is the spec for a CloudSchedulerSource resource

pkg/apis/events/v1beta1/cloudschedulersource_types_test.go

@@ -43,9 +43,9 @@ func TestCloudSchedulerSourceGetGroupVersionKind(t *testing.T) {
 }
 
 func TestCloudSchedulerSourceEventSource(t *testing.T) {
-	want := "//cloudscheduler.googleapis.com/PARENT/schedulers/SCHEDULER"
+	want := "//cloudscheduler.googleapis.com/JOB_NAME"
 
-	got := CloudSchedulerSourceEventSource("PARENT", "SCHEDULER")
+	got := CloudSchedulerSourceEventSource("JOB_NAME")
 
 	if diff := cmp.Diff(want, got); diff != "" {
 		t.Errorf("failed to get expected (-want, +got) = %v", diff)

pkg/apis/events/v1beta1/cloudstoragesource_defaults.go

@@ -24,7 +24,7 @@ import (
 	duckv1beta1 "github.com/google/knative-gcp/pkg/apis/duck/v1beta1"
 )
 
-var allEventTypes = []string{CloudStorageSourceFinalize, CloudStorageSourceDelete, CloudStorageSourceArchive, CloudStorageSourceMetadataUpdate}
+var allEventTypes = []string{CloudStorageSourceObjectFinalizedEventType, CloudStorageSourceObjectDeletedEventType, CloudStorageSourceObjectArchivedEventType, CloudStorageSourceObjectMetadataUpdatedEventType}
 
 func (s *CloudStorageSource) SetDefaults(ctx context.Context) {
 	ctx = apis.WithinParent(ctx, s.ObjectMeta)

pkg/apis/events/v1beta1/cloudstoragesource_defaults_test.go

@@ -47,7 +47,7 @@ func TestCloudStorageSource_SetDefaults(t *testing.T) {
 		},
 		"defaults present": {
 			orig: &CloudStorageSourceSpec{
-				EventTypes: []string{CloudStorageSourceFinalize, CloudStorageSourceDelete},
+				EventTypes: []string{CloudStorageSourceObjectFinalizedEventType, CloudStorageSourceObjectDeletedEventType},
 				PubSubSpec: duckv1beta1.PubSubSpec{
 					Secret: &corev1.SecretKeySelector{
 						LocalObjectReference: corev1.LocalObjectReference{
@@ -58,7 +58,7 @@ func TestCloudStorageSource_SetDefaults(t *testing.T) {
 				},
 			},
 			expected: &CloudStorageSourceSpec{
-				EventTypes: []string{CloudStorageSourceFinalize, CloudStorageSourceDelete},
+				EventTypes: []string{CloudStorageSourceObjectFinalizedEventType, CloudStorageSourceObjectDeletedEventType},
 				PubSubSpec: duckv1beta1.PubSubSpec{
 					Secret: &corev1.SecretKeySelector{
 						LocalObjectReference: corev1.LocalObjectReference{

pkg/apis/events/v1beta1/cloudstoragesource_types.go

@@ -82,10 +82,12 @@ type CloudStorageSourceSpec struct {
 
 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"
+	// TODO: somehow reference the proto values directly.
+	CloudStorageSourceObjectFinalizedEventType       = "google.cloud.storage.object.v1.finalized"
+	CloudStorageSourceObjectArchivedEventType        = "google.cloud.storage.object.v1.archived"
+	CloudStorageSourceObjectDeletedEventType         = "google.cloud.storage.object.v1.deleted"
+	CloudStorageSourceObjectMetadataUpdatedEventType = "google.cloud.storage.object.v1.metadataUpdated"
+	CloudStorageSourceEventDataSchema                = "https://github.com/googleapis/google-cloudevents/blob/master/proto/google/events/cloud/storage/v1/events.proto"
 
 	// CloudEvent source prefix.
 	storageSourcePrefix = "//storage.googleapis.com/buckets"

pkg/apis/events/v1beta1/cloudstoragesource_validation_test.go

@@ -73,7 +73,7 @@ var (
 	// Bucket, Sink, Secret, Event Type and Project, ObjectNamePrefix and PayloadFormat
 	storageSourceSpec = CloudStorageSourceSpec{
 		Bucket:           "my-test-bucket",
-		EventTypes:       []string{CloudStorageSourceFinalize, CloudStorageSourceDelete},
+		EventTypes:       []string{CloudStorageSourceObjectFinalizedEventType, CloudStorageSourceObjectDeletedEventType},
 		ObjectNamePrefix: "test-prefix",
 		PayloadFormat:    cloudevents.ApplicationJSON,
 		PubSubSpec: duckv1beta1.PubSubSpec{
@@ -366,7 +366,7 @@ func TestCheckImmutableFields(t *testing.T) {
 			orig: &storageSourceSpec,
 			updated: CloudStorageSourceSpec{
 				Bucket:           storageSourceSpec.Bucket,
-				EventTypes:       []string{CloudStorageSourceMetadataUpdate},
+				EventTypes:       []string{CloudStorageSourceObjectMetadataUpdatedEventType},
 				ObjectNamePrefix: storageSourceSpec.ObjectNamePrefix,
 				PayloadFormat:    storageSourceSpec.PayloadFormat,
 				PubSubSpec:       storageSourceSpec.PubSubSpec,

pkg/pubsub/adapter/converters/auditlogs.go

@@ -39,8 +39,6 @@ import (
 )
 
 const (
-	logEntrySchema = "type.googleapis.com/google.logging.v2.LogEntry"
-
 	parentResourcePattern = `^(:?projects|organizations|billingAccounts|folders)/[^/]+`
 
 	serviceNameExtension  = "servicename"
@@ -101,6 +99,16 @@ func resolveAnyUnknowns(typeURL string) (proto.Message, error) {
 	return reflect.New(mt.Elem()).Interface().(proto.Message), nil
 }
 
+// Log name, e.g. "organizations/1234567890/logs/cloudresourcemanager.googleapis.com%2Factivity"
+func logActivity(logName string) string {
+	parts := strings.Split(logName, "%2F")
+	if len(parts) < 2 {
+		return ""
+	}
+	// Could be "activity" or "data_access"
+	return parts[1]
+}
+
 func convertCloudAuditLogs(ctx context.Context, msg *pubsub.Message) (*cev2.Event, error) {
 	entry := logpb.LogEntry{}
 	if err := jsonpbUnmarshaller.Unmarshal(bytes.NewReader(msg.Data), &entry); err != nil {
@@ -111,6 +119,7 @@ func convertCloudAuditLogs(ctx context.Context, msg *pubsub.Message) (*cev2.Even
 	if parentResource == "" {
 		return nil, fmt.Errorf("invalid LogName: %q", entry.LogName)
 	}
+	logActivity := logActivity(entry.LogName)
 
 	// Make a new event and convert the message payload.
 	event := cev2.NewEvent(cev2.VersionV1)
@@ -120,8 +129,10 @@ func convertCloudAuditLogs(ctx context.Context, msg *pubsub.Message) (*cev2.Even
 	} else {
 		event.SetTime(timestamp)
 	}
+	event.SetType(v1beta1.CloudAuditLogsSourceLogWrittenEventType)
+	event.SetSource(v1beta1.CloudAuditLogsSourceEventSource(parentResource, logActivity))
+	event.SetDataSchema(v1beta1.CloudAuditLogsSourceEventDataSchema)
 	event.SetData(cev2.ApplicationJSON, msg.Data)
-	event.SetDataSchema(logEntrySchema)
 
 	switch payload := entry.Payload.(type) {
 	case *logpb.LogEntry_ProtoPayload:
@@ -131,9 +142,8 @@ func convertCloudAuditLogs(ctx context.Context, msg *pubsub.Message) (*cev2.Even
 		}
 		switch proto := unpacked.Message.(type) {
 		case *auditpb.AuditLog:
-			event.SetType(v1beta1.CloudAuditLogsSourceEvent)
-			event.SetSource(v1beta1.CloudAuditLogsSourceEventSource(proto.ServiceName, parentResource))
-			event.SetSubject(proto.ResourceName)
+			event.SetSubject(v1beta1.CloudAuditLogsSourceEventSubject(proto.ServiceName, proto.ResourceName))
+			// TODO: figure out if we want to keep these extensions.
 			event.SetExtension(serviceNameExtension, proto.ServiceName)
 			event.SetExtension(methodNameExtension, proto.MethodName)
 			event.SetExtension(resourceNameExtension, proto.ResourceName)

pkg/pubsub/adapter/converters/auditlogs_test.go

@@ -35,15 +35,15 @@ import (
 
 const (
 	insertID = "test-insert-id"
-	logName  = "projects/test-project/test-log-name"
+	logName  = "projects/test-project/pubsub.googleapis.com%2Factivity"
 	testTs   = "2006-01-02T15:04:05Z"
 )
 
 func TestConvertAuditLog(t *testing.T) {
 	auditLog := auditpb.AuditLog{
-		ServiceName:  "test-service-name",
+		ServiceName:  "pubsub.googleapis.com",
 		MethodName:   "test-method-name",
-		ResourceName: "test-resource-name",
+		ResourceName: "projects/test-project/topics/test-topic",
 	}
 	payload, err := ptypes.MarshalAny(&auditLog)
 	if err != nil {
@@ -87,15 +87,18 @@ func TestConvertAuditLog(t *testing.T) {
 	if !e.Time().Equal(testTime) {
 		t.Errorf("Time '%v' != '%v'", e.Time(), testTime)
 	}
-	if want := v1beta1.CloudAuditLogsSourceEventSource("test-service-name", "projects/test-project"); e.Source() != want {
+	if want := v1beta1.CloudAuditLogsSourceEventSource("projects/test-project", "activity"); e.Source() != want {
 		t.Errorf("Source %q != %q", e.Source(), want)
 	}
-	if e.Type() != "com.google.cloud.auditlog.event" {
-		t.Errorf(`Type %q != "com.google.cloud.auditlog.event"`, e.Type())
+	if e.Type() != "google.cloud.audit.log.v1.written" {
+		t.Errorf(`Type %q != "google.cloud.audit.log.v1.written"`, e.Type())
 	}
-	if want := "test-resource-name"; e.Subject() != want {
+	if want := v1beta1.CloudAuditLogsSourceEventSubject("pubsub.googleapis.com", "projects/test-project/topics/test-topic"); e.Subject() != want {
 		t.Errorf("Subject %q != %q", e.Subject(), want)
 	}
+	if e.DataSchema() != v1beta1.CloudAuditLogsSourceEventDataSchema {
+		t.Errorf("DataSchema got=%s, want=%s", e.DataSchema(), v1beta1.CloudAuditLogsSourceEventDataSchema)
+	}
 
 	var actualLogEntry logpb.LogEntry
 	if err = jsonpb.Unmarshal(bytes.NewReader(e.Data()), &actualLogEntry); err != nil {
@@ -107,9 +110,9 @@ func TestConvertAuditLog(t *testing.T) {
 	}
 
 	wantExtensions := map[string]interface{}{
-		"servicename":  "test-service-name",
+		"servicename":  "pubsub.googleapis.com",
 		"methodname":   "test-method-name",
-		"resourcename": "test-resource-name",
+		"resourcename": "projects/test-project/topics/test-topic",
 	}
 	if diff := cmp.Diff(wantExtensions, e.Extensions()); diff != "" {
 		t.Errorf("unexpected (-want, +got) = %v", diff)

pkg/pubsub/adapter/converters/pubsub.go

@@ -41,13 +41,14 @@ func convertCloudPubSub(ctx context.Context, msg *pubsub.Message) (*cev2.Event,
 	}
 
 	event.SetSource(v1beta1.CloudPubSubSourceEventSource(project, topic))
-	event.SetType(v1beta1.CloudPubSubSourcePublish)
+	event.SetType(v1beta1.CloudPubSubSourceMessagePublishedEventType)
 
 	subscription, err := GetSubscriptionKey(ctx)
 	if err != nil {
 		return nil, err
 	}
 
+	// TODO: use struct generated from proto: https://github.com/googleapis/google-cloudevents/blob/master/proto/google/events/cloud/pubsub/v1/events.proto#L33
 	pushMessage := &PushMessage{
 		Subscription: subscription,
 		Message: &PubSubMessage{

pkg/pubsub/adapter/converters/pubsub_pull.go

@@ -17,9 +17,10 @@ limitations under the License.
 package converters
 
 import (
-	"cloud.google.com/go/pubsub"
 	"context"
 	"fmt"
+
+	"cloud.google.com/go/pubsub"
 	cev2 "github.com/cloudevents/sdk-go/v2"
 	. "github.com/cloudevents/sdk-go/v2/event"
 	"github.com/google/knative-gcp/pkg/apis/events/v1beta1"
@@ -41,7 +42,7 @@ func convertPubSubPull(ctx context.Context, msg *pubsub.Message) (*cev2.Event, e
 	}
 
 	event.SetSource(v1beta1.CloudPubSubSourceEventSource(project, topic))
-	event.SetType(v1beta1.CloudPubSubSourcePublish)
+	event.SetType(v1beta1.CloudPubSubSourceMessagePublishedEventType)
 
 	// We promote attributes to extensions. If there is at least one attribute that cannot be promoted, we fail.
 	if msg.Attributes != nil && len(msg.Attributes) > 0 {

pkg/pubsub/adapter/converters/pubsub_pull_test.go

@@ -18,13 +18,14 @@ package converters
 
 import (
 	"context"
-	"github.com/cloudevents/sdk-go"
-	"github.com/google/knative-gcp/pkg/apis/events/v1alpha1"
 	"testing"
 
+	cloudevents "github.com/cloudevents/sdk-go"
+
 	"cloud.google.com/go/pubsub"
 	cev2 "github.com/cloudevents/sdk-go/v2"
 	"github.com/google/go-cmp/cmp"
+	"github.com/google/knative-gcp/pkg/apis/events/v1beta1"
 	. "github.com/google/knative-gcp/pkg/pubsub/adapter/context"
 )
 
@@ -54,7 +55,7 @@ func TestConvertPubSubPull(t *testing.T) {
 	}, {
 		name: "upper case attributes",
 		message: &pubsub.Message{
-			ID: "id",
+			ID:   "id",
 			Data: []byte("test data"),
 			Attributes: map[string]string{
 				"AttriBUte1": "value1",
@@ -102,8 +103,8 @@ func TestConvertPubSubPull(t *testing.T) {
 func pubSubPull(extensions map[string]string) *cev2.Event {
 	e := cev2.NewEvent(cloudevents.VersionV1)
 	e.SetID("id")
-	e.SetSource(v1alpha1.CloudPubSubSourceEventSource("testproject", "testtopic"))
-	e.SetType(v1alpha1.CloudPubSubSourcePublish)
+	e.SetSource(v1beta1.CloudPubSubSourceEventSource("testproject", "testtopic"))
+	e.SetType(v1beta1.CloudPubSubSourceMessagePublishedEventType)
 	e.SetData("application/octet-stream", []byte("test data"))
 	for k, v := range extensions {
 		e.SetExtension(k, v)