Inline 🦆 SourceSpec

Relates to PDY-54

registry/pkg/adapter/adapter.go

@@ -64,9 +64,6 @@ type envConfig struct {
 	EnvOwnerRepo string `envconfig:"REGISTRY_OWNER_REPO" required:"true"`
 	// Environment variable containing information about tags to filter
 	Tags *string `envconfig:"TAGS"`
-	// Extra extension attributes included in every Cloud Event this source emits
-	// Example: "key1:value1,key2:value2"
-	Data string `envconfig:"DATA"`
 }
 
 // NewEnvConfig function reads env variables defined in envConfig structure and
@@ -262,10 +259,16 @@ func (a *registryAdapter) sendEvent(eventType string, desc *remote.Descriptor) e
 	event.SetType(cloudEventType)
 	event.SetSource(a.env.EnvRegistryBaseUrl)
 	event.SetSubject(a.env.EnvOwnerRepo)
-
-	extraAttributes := a.parseExtraAttributes()
-	for key, value := range extraAttributes {
-		event.SetExtension(key, value)
+	overrides, err := a.env.GetCloudEventOverrides()
+	if err != nil {
+		return fmt.Errorf("failed to unmarshal cloudevent overrides: %w", err)
+	}
+	for key, value := range overrides.Extensions {
+		if key == "action" {
+			a.logger.Warnf("'action' is a reserved CloudEvent override key for RegistrySource, skipping value: %s", value)
+		} else {
+			event.SetExtension(key, value)
+		}
 	}
 	event.SetExtension("action", eventType)
 
@@ -288,20 +291,6 @@ func (a *registryAdapter) sendEvent(eventType string, desc *remote.Descriptor) e
 	return nil
 }
 
-func (a *registryAdapter) parseExtraAttributes() map[string]string {
-	result := map[string]string{}
-	keyValuePairs := strings.Split(a.env.Data, ",")
-	for _, keyValue := range keyValuePairs {
-		elements := strings.SplitN(keyValue, ":", 2)
-		if len(elements) < 2 {
-			a.logger.Warnf("failed to parse extra attribute %q, missing colon", keyValue)
-			continue
-		}
-		result[elements[0]] = elements[1]
-	}
-	return result
-}
-
 func buildImageStrWithDigest(desc *remote.Descriptor) string {
 	repo := strings.Split(desc.Ref.String(), ":")
 	return fmt.Sprintf("%s@%s", repo[0], desc.Digest.String())

registry/pkg/apis/sources/v1alpha1/registrysource_types.go

@@ -36,6 +36,13 @@ var _ resourcesemantics.GenericCRD = (*RegistrySource)(nil)
 // RegistrySourceSpec defines the desired state of RegistrySource
 // +kubebuilder:categories=all,knative,eventing,sources
 type RegistrySourceSpec struct {
+	// inherits duck/v1 SourceSpec, which currently provides:
+	// * Sink - a reference to an object that will resolve to a domain name or
+	//   a URI directly to use as the sink.
+	// * CloudEventOverrides - defines overrides to control the output format
+	//   and modifications of the event sent to the sink.
+	duckv1.SourceSpec `json:",inline"`
+
 	// ServiceAccountName holds the name of the Kubernetes service account
 	// as which the underlying K8s resources should be run. If unspecified
 	// this will default to the "default" service account for the namespace
@@ -63,19 +70,10 @@ type RegistrySourceSpec struct {
 	// +kubebuilder:validation:Enum=create,delete,update
 	EventTypes []string `json:"eventTypes"`
 
-	// Sink is a reference to an object that will resolve to a domain
-	// name to use as the sink.
-	// +optional
-	Sink *duckv1.Destination `json:"sink,omitempty"`
-
 	// Tags the registry source emits events for
 	// Defaults to all tags of the specified repository
 	// +optional
 	Tags []string `json:"tags,omitempty"`
-	// Extra attributes the registry source includes in the Cloud Events it emits
-	// Each element is a colon-separated key-value pair. E.g.: "foo:bar"
-	// +optional
-	Data []string `json:"data,omitempty"`
 }
 
 const (

registry/pkg/apis/sources/v1alpha1/registrysource_validation.go

@@ -32,10 +32,7 @@ func (gs *RegistrySourceSpec) Validate(ctx context.Context) *apis.FieldError {
 	// TODO: there are more requirements for RegistrySource. Add them here.
 
 	// Validate sink
-	if gs.Sink == nil {
-		fe := apis.ErrMissingField("sink")
-		errs = errs.Also(fe)
-	} else if fe := gs.Sink.Validate(ctx); fe != nil {
+	if fe := gs.Sink.Validate(ctx); fe != nil {
 		errs = errs.Also(fe.ViaField("sink"))
 	}
 	return errs

registry/pkg/reconciler/source/resources/service.go

@@ -73,13 +73,6 @@ func MakeDeployment(args *ServiceArgs) *appsv1.Deployment {
 			Value: strings.Join(tags, ","),
 		})
 	}
-	data := args.Source.Spec.Data
-	if data != nil {
-		env = append(env, corev1.EnvVar{
-			Name: "DATA",
-			Value: strings.Join(data, ","),
-		})
-	}
 	return &appsv1.Deployment{
 		ObjectMeta: metav1.ObjectMeta{
 			GenerateName: fmt.Sprintf("%s-", args.Source.Name),