Reconcile EventTypes within Source Duck controller

cmd/controller/main.go

@@ -28,6 +28,7 @@ import (
 	"knative.dev/eventing/pkg/reconciler/parallel"
 	pingsource "knative.dev/eventing/pkg/reconciler/pingsource/controller"
 	"knative.dev/eventing/pkg/reconciler/sequence"
+	sourcecrd "knative.dev/eventing/pkg/reconciler/source/crd"
 	"knative.dev/eventing/pkg/reconciler/subscription"
 	"knative.dev/eventing/pkg/reconciler/trigger"
 )
@@ -50,5 +51,7 @@ func main() {
 		// Sources
 		apiserversource.NewController,
 		pingsource.NewController,
+		// Sources CRD
+		sourcecrd.NewController,
 	)
 }

config/core/resources/eventtype.yaml

@@ -56,6 +56,7 @@ spec:
     - name: Description
       type: string
       JSONPath: ".spec.description"
+    # TODO remove Status https://github.com/knative/eventing/issues/2750
     - name: Ready
       type: string
       JSONPath: ".status.conditions[?(@.type==\"Ready\")].status"

pkg/apis/eventing/register.go

@@ -56,6 +56,10 @@ const (
 	// ScopeCluster indicates the resource must be
 	// handled by the cluster-scoped component
 	ScopeCluster = "cluster"
+
+	// EventTypesAnnotationKey is the annotation key to specify
+	// if a Source has event types defines in its CRD.
+	EventTypesAnnotationKey = "registry.knative.dev/eventTypes"
 )
 
 var (

pkg/apis/eventing/v1alpha1/eventtype_types.go

@@ -40,6 +40,7 @@ type EventType struct {
 	// Status represents the current state of the EventType.
 	// This data may be out of date.
 	// +optional
+	// TODO might be removed https://github.com/knative/eventing/issues/2750
 	Status EventTypeStatus `json:"status,omitempty"`
 }
 
@@ -61,13 +62,16 @@ type EventTypeSpec struct {
 	// Type represents the CloudEvents type. It is authoritative.
 	Type string `json:"type"`
 	// Source is a URI, it represents the CloudEvents source.
-	Source string `json:"source"`
+	// +optional
+	Source string `json:"source,omitempty"`
 	// Schema is a URI, it represents the CloudEvents schemaurl extension attribute.
 	// It may be a JSON schema, a protobuf schema, etc. It is optional.
 	// +optional
 	Schema string `json:"schema,omitempty"`
+	// TODO remove https://github.com/knative/eventing/issues/2750
 	// Broker refers to the Broker that can provide the EventType.
-	Broker string `json:"broker"`
+	// +optional
+	Broker string `json:"broker,omitempty"`
 	// Description is an optional field used to describe the EventType, in any meaningful way.
 	// +optional
 	Description string `json:"description,omitempty"`

pkg/apis/eventing/v1alpha1/eventtype_validation.go

@@ -19,8 +19,6 @@ package v1alpha1
 import (
 	"context"
 
-	"github.com/google/go-cmp/cmp/cmpopts"
-
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/kmp"
 )
@@ -35,15 +33,6 @@ func (ets *EventTypeSpec) Validate(ctx context.Context) *apis.FieldError {
 		fe := apis.ErrMissingField("type")
 		errs = errs.Also(fe)
 	}
-	if ets.Source == "" {
-		// TODO validate is a valid URI.
-		fe := apis.ErrMissingField("source")
-		errs = errs.Also(fe)
-	}
-	if ets.Broker == "" {
-		fe := apis.ErrMissingField("broker")
-		errs = errs.Also(fe)
-	}
 	// TODO validate Schema is a valid URI.
 	return errs
 }
@@ -53,9 +42,8 @@ func (et *EventType) CheckImmutableFields(ctx context.Context, original *EventTy
 		return nil
 	}
 
-	// All but Description field immutable.
-	ignoreArguments := cmpopts.IgnoreFields(EventTypeSpec{}, "Description")
-	if diff, err := kmp.ShortDiff(original.Spec, et.Spec, ignoreArguments); err != nil {
+	// All fields immutable
+	if diff, err := kmp.ShortDiff(original.Spec, et.Spec); err != nil {
 		return &apis.FieldError{
 			Message: "Failed to diff EventType",
 			Paths:   []string{"spec"},

pkg/apis/eventing/v1alpha1/eventtype_validation_test.go

@@ -25,16 +25,16 @@ import (
 )
 
 func TestEventTypeValidation(t *testing.T) {
-	name := "invalid type and source and broker"
-	broker := &EventType{Spec: EventTypeSpec{}}
+	name := "invalid type"
+	et := &EventType{Spec: EventTypeSpec{}}
 
 	want := &apis.FieldError{
-		Paths:   []string{"spec.type", "spec.source", "spec.broker"},
+		Paths:   []string{"spec.type"},
 		Message: "missing field(s)",
 	}
 
 	t.Run(name, func(t *testing.T) {
-		got := broker.Validate(context.TODO())
+		got := et.Validate(context.TODO())
 		if diff := cmp.Diff(want.Error(), got.Error()); diff != "" {
 			t.Errorf("EventType.Validate (-want, +got) = %v", diff)
 		}
@@ -47,42 +47,12 @@ func TestEventTypeSpecValidation(t *testing.T) {
 		ets  *EventTypeSpec
 		want *apis.FieldError
 	}{{
-		name: "invalid eventtype spec",
-		ets:  &EventTypeSpec{},
-		want: func() *apis.FieldError {
-			fe := apis.ErrMissingField("type", "source", "broker")
-			return fe
-		}(),
-	}, {
 		name: "invalid eventtype type",
-		ets: &EventTypeSpec{
-			Source: "test-source",
-			Broker: "test-broker",
-		},
+		ets:  &EventTypeSpec{},
 		want: func() *apis.FieldError {
 			fe := apis.ErrMissingField("type")
 			return fe
 		}(),
-	}, {
-		name: "invalid eventtype source",
-		ets: &EventTypeSpec{
-			Type:   "test-type",
-			Broker: "test-broker",
-		},
-		want: func() *apis.FieldError {
-			fe := apis.ErrMissingField("source")
-			return fe
-		}(),
-	}, {
-		name: "invalid eventtype broker",
-		ets: &EventTypeSpec{
-			Type:   "test-type",
-			Source: "test-source",
-		},
-		want: func() *apis.FieldError {
-			fe := apis.ErrMissingField("broker")
-			return fe
-		}(),
 	},
 	}
 
@@ -232,7 +202,7 @@ func TestEventTypeImmutableFields(t *testing.T) {
 `,
 		},
 	}, {
-		name: "good (description change)",
+		name: "bad (description change)",
 		current: &EventType{
 			Spec: EventTypeSpec{
 				Type:        "test-type",
@@ -251,9 +221,15 @@ func TestEventTypeImmutableFields(t *testing.T) {
 				Description: "original-description",
 			},
 		},
-		want: nil,
-	},
-	}
+		want: &apis.FieldError{
+			Message: "Immutable fields changed (-old +new)",
+			Paths:   []string{"spec"},
+			Details: `{v1alpha1.EventTypeSpec}.Description:
+	-: "original-description"
+	+: "test-description"
+`,
+		},
+	}}
 
 	for _, test := range tests {
 		t.Run(test.name, func(t *testing.T) {

pkg/apis/eventing/v1alpha1/trigger_lifecycle.go

@@ -154,7 +154,7 @@ func (ts *TriggerStatus) MarkDependencyUnknown(reason, messageFormat string, mes
 }
 
 func (ts *TriggerStatus) MarkDependencyNotConfigured() {
-	triggerCondSet.Manage(ts).MarkUnknown(EventTypeConditionBrokerReady,
+	triggerCondSet.Manage(ts).MarkUnknown(TriggerConditionDependency,
 		"DependencyNotConfigured", "Dependency has not yet been reconciled.")
 }
 

pkg/apis/eventing/v1beta1/eventtype_defaults_test.go

@@ -44,15 +44,15 @@ func TestEventTypeDefaults(t *testing.T) {
 			initial: EventType{
 				Spec: EventTypeSpec{
 					Type:   "test-type",
-					Source: *testSource,
+					Source: testSource,
 					Broker: "",
 					Schema: testSchema,
 				},
 			},
 			expected: EventType{
 				Spec: EventTypeSpec{
 					Type:   "test-type",
-					Source: *testSource,
+					Source: testSource,
 					Broker: "default",
 					Schema: testSchema,
 				},
@@ -62,14 +62,14 @@ func TestEventTypeDefaults(t *testing.T) {
 			initial: EventType{
 				Spec: EventTypeSpec{
 					Type:   "test-type",
-					Source: *testSource,
+					Source: testSource,
 					Schema: testSchema,
 				},
 			},
 			expected: EventType{
 				Spec: EventTypeSpec{
 					Type:   "test-type",
-					Source: *testSource,
+					Source: testSource,
 					Broker: "default",
 					Schema: testSchema,
 				},

pkg/apis/eventing/v1beta1/eventtype_types.go

@@ -39,6 +39,7 @@ type EventType struct {
 	// Status represents the current state of the EventType.
 	// This data may be out of date.
 	// +optional
+	// TODO might be removed https://github.com/knative/eventing/issues/2750
 	Status EventTypeStatus `json:"status,omitempty"`
 }
 
@@ -60,7 +61,8 @@ type EventTypeSpec struct {
 	// Type represents the CloudEvents type. It is authoritative.
 	Type string `json:"type"`
 	// Source is a URI, it represents the CloudEvents source.
-	Source apis.URL `json:"source"`
+	// +optional
+	Source *apis.URL `json:"source,omitempty"`
 	// Schema is a URI, it represents the CloudEvents schemaurl extension attribute.
 	// It may be a JSON schema, a protobuf schema, etc. It is optional.
 	// +optional
@@ -70,8 +72,10 @@ type EventTypeSpec struct {
 	// The contents are not validated or manipulated by the system.
 	// +optional
 	SchemaData string `json:"schemaData,omitempty"`
+	// TODO remove https://github.com/knative/eventing/issues/2750
 	// Broker refers to the Broker that can provide the EventType.
-	Broker string `json:"broker"`
+	// +optional
+	Broker string `json:"broker,omitempty"`
 	// Description is an optional field used to describe the EventType, in any meaningful way.
 	// +optional
 	Description string `json:"description,omitempty"`

pkg/apis/eventing/v1beta1/eventtype_validation.go

@@ -19,8 +19,6 @@ package v1beta1
 import (
 	"context"
 
-	"github.com/google/go-cmp/cmp/cmpopts"
-
 	"knative.dev/pkg/apis"
 	"knative.dev/pkg/kmp"
 )
@@ -35,15 +33,7 @@ func (ets *EventTypeSpec) Validate(ctx context.Context) *apis.FieldError {
 		fe := apis.ErrMissingField("type")
 		errs = errs.Also(fe)
 	}
-	if ets.Source.IsEmpty() {
-		// TODO validate is a valid URI.
-		fe := apis.ErrMissingField("source")
-		errs = errs.Also(fe)
-	}
-	if ets.Broker == "" {
-		fe := apis.ErrMissingField("broker")
-		errs = errs.Also(fe)
-	}
+	// TODO validate Source is a valid URI.
 	// TODO validate Schema is a valid URI.
 	// There is no validation of the SchemaData, it is application specific data.
 	return errs
@@ -54,9 +44,8 @@ func (et *EventType) CheckImmutableFields(ctx context.Context, original *EventTy
 		return nil
 	}
 
-	// All but Description field immutable.
-	ignoreArguments := cmpopts.IgnoreFields(EventTypeSpec{}, "Description")
-	if diff, err := kmp.ShortDiff(original.Spec, et.Spec, ignoreArguments); err != nil {
+	// All fields are immutable.
+	if diff, err := kmp.ShortDiff(original.Spec, et.Spec); err != nil {
 		return &apis.FieldError{
 			Message: "Failed to diff EventType",
 			Paths:   []string{"spec"},